Node client for the Salesforce Pub/Sub API

See the official Pub/Sub API repo and the documentation for more information on the Salesforce gRPC-based Pub/Sub API.

• v4 to v5 Migration
• v4 Documentation
• Installation and Configuration
  • Authentication
    • User supplied authentication
    • Username/password flow
    • OAuth 2.0 client credentials flow (client_credentials)
    • OAuth 2.0 JWT bearer flow
  • Logging
• Quick Start Example
• Other Examples
  • Publish a single platform event
  • Publish a batch of platform events
  • Subscribe with a replay ID
  • Subscribe to past events in retention window
  • Subscribe using a managed subscription
  • Work with flow control for high volumes of events
  • Handle gRPC stream lifecycle events
• Common Issues
• Reference
  • PubSubApiClient
  • PublishCallback
  • SubscribeCallback
  • SubscriptionInfo
  • EventParseError
  • Configuration

This project bundles and uses CA root certificates from the python-ceritfi project.

v4 to v5 Migration

[!WARNING] Version 5 of the Pub/Sub API client introduces a couple of breaking changes which require a small migration effort. Read this section for an overview of the changes.

Configuration and Connection

In v4 and earlier versions of this client:

• you specify the configuration in a .env file with specific property names.
• you connect with either the connect() or connectWithAuth() method depending on the authentication flow.

In v5:

• you pass your configuration with an object in the client constructor. The .env file is no longer a requirement, you are free to store your configuration where you want.
• you connect with a unique connect() method.

Event handling

In v4 and earlier versions of this client you use an asynchronous EventEmitter to receive updates such as incoming messages or lifecycle events:

1// Subscribe to account change events
2const eventEmitter = await client.subscribe(
3    '/data/AccountChangeEvent'
4);
5
6// Handle incoming events
7eventEmitter.on('data', (event) => {
8    // Event handling logic goes here
9}):

In v5 you use a synchronous callback function to receive the same information. This helps to ensure that events are received in the right order.

1const subscribeCallback = (subscription, callbackType, data) => {
2    // Event handling logic goes here
3};
4
5// Subscribe to account change events
6await client.subscribe('/data/AccountChangeEvent', subscribeCallback);

Installation and Configuration

Install the client library with npm install salesforce-pubsub-api-client.

Authentication

Pick one of these authentication flows and pass the relevant configuration to the PubSubApiClient constructor:

• User supplied authentication
• Username/password flow (recommended for tests)
• OAuth 2.0 client flow
• OAuth 2.0 JWT Bearer flow (recommended for production)

User supplied authentication

If you already have a Salesforce client in your app, you can reuse its authentication information. In the example below, we assume that sfConnection is a connection obtained with jsforce

1const client = new PubSubApiClient({
2    authType: 'user-supplied',
3    accessToken: sfConnection.accessToken,
4    instanceUrl: sfConnection.instanceUrl,
5    organizationId: sfConnection.userInfo.organizationId
6});

Username/password flow

[!WARNING] Relying on a username/password authentication flow for production is not recommended. Consider switching to JWT auth for extra security.

1const client = new PubSubApiClient({
2    authType: 'username-password',
3    loginUrl: process.env.SALESFORCE_LOGIN_URL,
4    username: process.env.SALESFORCE_USERNAME,
5    password: process.env.SALESFORCE_PASSWORD,
6    userToken: process.env.SALESFORCE_TOKEN
7});

OAuth 2.0 client credentials flow (client_credentials)

1const client = new PubSubApiClient({
2    authType: 'oauth-client-credentials',
3    loginUrl: process.env.SALESFORCE_LOGIN_URL,
4    clientId: process.env.SALESFORCE_CLIENT_ID,
5    clientSecret: process.env.SALESFORCE_CLIENT_SECRET
6});

OAuth 2.0 JWT bearer flow

This is the most secure authentication option. Recommended for production use.

1// Read private key file
2const privateKey = fs.readFileSync(process.env.SALESFORCE_PRIVATE_KEY_FILE);
3
4// Build PubSub client
5const client = new PubSubApiClient({
6    authType: 'oauth-jwt-bearer',
7    loginUrl: process.env.SALESFORCE_JWT_LOGIN_URL,
8    clientId: process.env.SALESFORCE_JWT_CLIENT_ID,
9    username: process.env.SALESFORCE_USERNAME,
10    privateKey
11});

Logging

The client uses debug level messages so you can lower the default logging level if you need more information.

The documentation examples use the default client logger (the console). The console is fine for a test environment but you'll want to switch to a custom logger with asynchronous logging for increased performance.

You can pass a logger like pino in the client constructor:

1import pino from 'pino';
2
3const config = {
4    /* your config goes here */
5};
6const logger = pino();
7const client = new PubSubApiClient(config, logger);

Quick Start Example

Here's an example that will get you started quickly. It listens to up to 3 account change events. Once the third event is reached, the client closes gracefully.

1. Activate Account change events in Salesforce Setup > Change Data Capture.

2. Install the client and dotenv in your project:

   1npm install salesforce-pubsub-api-client dotenv

3. Create a .env file at the root of the project and replace the values:

   1SALESFORCE_LOGIN_URL=...
   2SALESFORCE_USERNAME=...
   3SALESFORCE_PASSWORD=...
   4SALESFORCE_TOKEN=...

4. Create a sample.js file with the following content:

   1import * as dotenv from 'dotenv';
   2import PubSubApiClient from 'salesforce-pubsub-api-client';
   3
   4async function run() {
   5    try {
   6        // Load config from .env file
   7        dotenv.config();
   8
   9        // Build and connect Pub/Sub API client
   10        const client = new PubSubApiClient({
   11            authType: 'username-password',
   12            loginUrl: process.env.SALESFORCE_LOGIN_URL,
   13            username: process.env.SALESFORCE_USERNAME,
   14            password: process.env.SALESFORCE_PASSWORD,
   15            userToken: process.env.SALESFORCE_TOKEN
   16        });
   17        await client.connect();
   18
   19        // Prepare event callback
   20        const subscribeCallback = (subscription, callbackType, data) => {
   21            switch (callbackType) {
   22                case 'event':
   23                    // Event received
   24                    console.log(
   25                        `${subscription.topicName} - Handling ${data.payload.ChangeEventHeader.entityName} change event ` +
   26                            `with ID ${data.replayId} ` +
   27                            `(${subscription.receivedEventCount}/${subscription.requestedEventCount} ` +
   28                            `events received so far)`
   29                    );
   30                    // Safely log event payload as a JSON string
   31                    console.log(
   32                        JSON.stringify(
   33                            data,
   34                            (key, value) =>
   35                                /* Convert BigInt values into strings and keep other types unchanged */
   36                                typeof value === 'bigint'
   37                                    ? value.toString()
   38                                    : value,
   39                            2
   40                        )
   41                    );
   42                    break;
   43                case 'lastEvent':
   44                    // Last event received
   45                    console.log(
   46                        `${subscription.topicName} - Reached last of ${subscription.requestedEventCount} requested event on channel. Closing connection.`
   47                    );
   48                    break;
   49                case 'end':
   50                    // Client closed the connection
   51                    console.log('Client shut down gracefully.');
   52                    break;
   53            }
   54        };
   55
   56        // Subscribe to 3 account change event
   57        client.subscribe('/data/AccountChangeEvent', subscribeCallback, 3);
   58    } catch (error) {
   59        console.error(error);
   60    }
   61}
   62
   63run();

5. Run the project with node sample.js

   If everything goes well, you'll see output like this:

   Connected to Salesforce org https://pozil-dev-ed.my.salesforce.com (00D58000000arpqEAA) as grpc@pozil.com
   Connected to Pub/Sub API endpoint api.pubsub.salesforce.com:7443
   /data/AccountChangeEvent - Subscribe request sent for 3 events
   

   At this point, the script is on hold and waits for events.

6. Modify an account record in Salesforce. This fires an account change event.

   Once the client receives an event, it displays it like this:

   /data/AccountChangeEvent - Received 1 events, latest replay ID: 18098167
   /data/AccountChangeEvent - Handling Account change event with ID 18098167 (1/3 events received so far)
   {
       "replayId": 18098167,
       "payload": {
           "ChangeEventHeader": {
           "entityName": "Account",
           "recordIds": [
               "0014H00002LbR7QQAV"
           ],
           "changeType": "UPDATE",
           "changeOrigin": "com/salesforce/api/soap/58.0;client=SfdcInternalAPI/",
           "transactionKey": "000046c7-a642-11e2-c29b-229c6786473e",
           "sequenceNumber": 1,
           "commitTimestamp": 1696444513000,
           "commitNumber": 11657372702432,
           "commitUser": "00558000000yFyDAAU",
           "nulledFields": [],
           "diffFields": [],
           "changedFields": [
               "LastModifiedDate",
               "BillingAddress.City",
               "BillingAddress.State"
           ]
           },
           "Name": null,
           "Type": null,
           "ParentId": null,
           "BillingAddress": {
               "Street": null,
               "City": "San Francisco",
               "State": "CA",
               "PostalCode": null,
               "Country": null,
               "StateCode": null,
               "CountryCode": null,
               "Latitude": null,
               "Longitude": null,
               "Xyz": null,
               "GeocodeAccuracy": null
           },
           "ShippingAddress": null,
           "Phone": null,
           "Fax": null,
           "AccountNumber": null,
           "Website": null,
           "Sic": null,
           "Industry": null,
           "AnnualRevenue": null,
           "NumberOfEmployees": null,
           "Ownership": null,
           "TickerSymbol": null,
           "Description": null,
           "Rating": null,
           "Site": null,
           "OwnerId": null,
           "CreatedDate": null,
           "CreatedById": null,
           "LastModifiedDate": 1696444513000,
           "LastModifiedById": null,
           "Jigsaw": null,
           "JigsawCompanyId": null,
           "CleanStatus": null,
           "AccountSource": null,
           "DunsNumber": null,
           "Tradestyle": null,
           "NaicsCode": null,
           "NaicsDesc": null,
           "YearStarted": null,
           "SicDesc": null,
           "DandbCompanyId": null
       }
   }
   

   Note that the change event payloads include all object fields but fields that haven't changed are null. In the above example, the only changes are the Billing State, Billing City and Last Modified Date.

   Use the values from ChangeEventHeader.nulledFields, ChangeEventHeader.diffFields and ChangeEventHeader.changedFields to identify actual value changes.

Other Examples

Publish a single platform event

[!NOTE] For best performances, use publishBatch when publishing event batches.

Publish a single Sample__e platform events with a Message__c field using publish:

1const payload = {
2    CreatedDate: new Date().getTime(), // Non-null value required but there's no validity check performed on this field
3    CreatedById: '005_________', // Valid user ID
4    Message__c: { string: 'Hello world' } // Field is nullable so we need to specify the 'string' type
5};
6const publishResult = await client.publish('/event/Sample__e', payload);
7console.log('Published event: ', JSON.stringify(publishResult));

Publish a batch of platform events

Publish a batch of Sample__e platform events using publishBatch:

1// Prepare publish callback
2const publishCallback = (info, callbackType, data) => {
3    switch (callbackType) {
4        case 'publishResponse':
5            console.log(JSON.stringify(data));
6            break;
7    }
8};
9
10// Prepare events
11const events = [
12    {
13        payload: {
14            CreatedDate: new Date().getTime(), // Non-null value required but there's no validity check performed on this field
15            CreatedById: '005_________', // Valid user ID
16            Message__c: { string: 'Hello world' } // Field is nullable so we need to specify the 'string' type
17        }
18    }
19];
20
21// Publish event batch
22client.publishBatch('/event/Sample__e', events, publishCallback);

Subscribe with a replay ID

Subscribe to 5 account change events starting from a replay ID:

1await client.subscribeFromReplayId(
2    '/data/AccountChangeEvent',
3    subscribeCallback,
4    5,
5    17092989
6);

Subscribe to past events in retention window

Subscribe to the 3 earliest past account change events in the retention window:

1await client.subscribeFromEarliestEvent(
2    '/data/AccountChangeEvent',
3    subscribeCallback,
4    3
5);

Subscribe using a managed subscription

You can turn your Pub/Sub client application stateless by delegating the tracking of replay IDs to the server thanks to managed event subscriptions.

1. Create a managed event subscription using the tooling API. You can use API request templates from the Salesforce Platform APIs Postman collection to do so.
2. Subscribe to 3 events from a managed event subscription (Managed_Sample_PE in this expample):

   1await client.subscribeWithManagedSubscription(
   2    'Managed_Sample_PE',
   3    subscribeCallback,
   4    3
   5);

3. Using the subscription information sent in the subscribe callback, frequently commit the last replay ID that you receveive:

   1client.commitReplayId(
   2    subscription.subscriptionId,
   3    subscription.lastReplayId
   4);

4. Optionnaly, request additional events to be sent (3 more in this example):

   1client.requestAdditionalManagedEvents(subscription.subscriptionId, 3);

Work with flow control for high volumes of events

When working with high volumes of events you can control the incoming flow of events by requesting a limited batch of events. This event flow control ensures that the client doesn’t get overwhelmed by accepting more events that it can handle if there is a spike in event publishing.

This is the overall process:

1. Pass a number of requested events in your subscribe call.
2. Handle the lastevent callback type from subscribe callback to detect the end of the event batch.
3. Subscribe to an additional batch of events with client.requestAdditionalEvents(...). If you don't request additional events at this point, the gRPC subscription will close automatically (default Pub/Sub API behavior).

The code below illustrate how you can achieve event flow control:

1try {
2    // Connect with the Pub/Sub API
3    const client = new PubSubApiClient(/* config goes here */);
4    await client.connect();
5
6    // Prepare event callback
7    const subscribeCallback = (subscription, callbackType, data) => {
8        switch (callbackType) {
9            case 'event':
10                // Logic for handling a single event.
11                // Unless you request additional events later, this should get called up to 10 times
12                // given the initial subscription boundary.
13            break;
14            case 'lastEvent':
15                // Last event received
16                console.log(
17                    `${eventEmitter.getTopicName()} - Reached last requested event on channel.`
18                );
19                // Request 10 additional events
20                client.requestAdditionalEvents(eventEmitter, 10);
21            break;
22            case 'end':
23                // Client closed the connection
24                console.log('Client shut down gracefully.');
25            break;
26        }
27    };
28
29    // Subscribe to a batch of 10 account change event
30    await client.subscribe('/data/AccountChangeEvent', subscribeCallback 10);
31} catch (error) {
32    console.error(error);
33}

Handle gRPC stream lifecycle events

Use callback types from subscribe callback to handle gRPC stream lifecycle events:

1const subscribeCallback = (subscription, callbackType, data) => {
2    if (callbackType === 'grpcStatus') {
3        // Stream status update
4        console.log('gRPC stream status: ', status);
5    } else if (callbackType === 'error') {
6        // Stream error
7        console.error('gRPC stream error: ', JSON.stringify(error));
8    } else if (callbackType === 'end') {
9        // Stream end
10        console.log('gRPC stream ended');
11    }
12};

Common Issues

TypeError: Do not know how to serialize a BigInt

If you attempt to call JSON.stringify on an event you will likely see the following error:

TypeError: Do not know how to serialize a BigInt

This happens when an integer value stored in an event field exceeds the range of the Number JS type (this typically happens with commitNumber values). In this case, we use a BigInt type to safely store the integer value. However, the BigInt type is not yet supported in standard JSON representation (see step 10 in the BigInt TC39 spec) so this triggers a TypeError.

To avoid this error, use a replacer function to safely escape BigInt values so that they can be serialized as a string (or any other format of your choice) in JSON:

1// Safely log event as a JSON string
2console.log(
3    JSON.stringify(
4        event,
5        (key, value) =>
6            /* Convert BigInt values into strings and keep other types unchanged */
7            typeof value === 'bigint' ? value.toString() : value,
8        2
9    )
10);

Reference

PubSubApiClient

Client for the Salesforce Pub/Sub API

PubSubApiClient(configuration, [logger])

Builds a new Pub/Sub API client.

Name	Type	Description
configuration	Configuration	The client configuration (authentication...).
logger	Logger	An optional custom logger. The client uses the console if no value is supplied.

close()

Closes the gRPC connection. The client will no longer receive events for any topic.

commitReplayId(subscriptionId, replayId) → string

Commits a replay ID on a managed subscription.

Returns: commit request UUID.

Name	Type	Description
subscriptionId	string	managed subscription ID
replayId	number	event replay ID

async connect() → {Promise.<void>}

Authenticates with Salesforce then connects to the Pub/Sub API.

Returns: Promise that resolves once the connection is established.

async getConnectivityState() → {Promise<connectivityState>}

Gets the gRPC connectivity state from the current channel.

Returns: Promise that holds the channel's connectivity state.

async publish(topicName, payload, [correlationKey]) → {Promise.<PublishResult>}

Publishes an payload to a topic using the gRPC client. This is a synchronous operation, use publishBatch when publishing event batches.

Returns: Promise holding a PublishResult object with replayId and correlationKey.

Name	Type	Description
topicName	string	name of the topic that we're publishing on
payload	Object	payload of the event that is being published
correlationKey	string	optional correlation key. If you don't provide one, we'll generate a random UUID for you.

async publishBatch(topicName, events, publishCallback)

Publishes a batch of events using the gRPC client's publish stream.

Name	Type	Description
topicName	string	name of the topic that we're publishing on
events	PublisherEvent[]	events to be published
publishCallback	PublishCallback	callback function for handling publish responses.

async subscribe(topicName, subscribeCallback, [numRequested])

Subscribes to a topic.

Name	Type	Description
topicName	string	name of the topic that we're subscribing to
subscribeCallback	SubscribeCallback	subscribe callback function
numRequested	number	optional number of events requested. If not supplied or null, the client keeps the subscription alive forever.

async subscribeFromEarliestEvent(topicName, subscribeCallback, [numRequested])

Subscribes to a topic and retrieves all past events in retention window.

Name	Type	Description
topicName	string	name of the topic that we're subscribing to
subscribeCallback	SubscribeCallback	subscribe callback function
numRequested	number	optional number of events requested. If not supplied or null, the client keeps the subscription alive forever.

async subscribeFromReplayId(topicName, subscribeCallback, numRequested, replayId)

Subscribes to a topic and retrieves past events starting from a replay ID.

Name	Type	Description
topicName	string	name of the topic that we're subscribing to
subscribeCallback	SubscribeCallback	subscribe callback function
numRequested	number	number of events requested. If null, the client keeps the subscription alive forever.
replayId	number	replay ID

async subscribeWithManagedSubscription(subscriptionIdOrName, subscribeCallback, [numRequested])

Subscribes to a topic thanks to a managed subscription.

Throws an error if the managed subscription does not exist or is not in the RUN state.

Name	Type	Description
subscriptionIdOrName	string	managed subscription ID or developer name
subscribeCallback	SubscribeCallback	subscribe callback function
numRequested	number	optional number of events requested. If not supplied or null, the client keeps the subscription alive forever.

requestAdditionalEvents(topicName, numRequested)

Request additional events on an existing subscription.

Name	Type	Description
topicName	string	name of the topic.
numRequested	number	number of events requested.

requestAdditionalManagedEvents(subscriptionId, numRequested)

Request additional events on an existing managed subscription.

Name	Type	Description
subscriptionId	string	managed subscription ID.
numRequested	number	number of events requested.

PublishCallback

Callback function that lets you process batch publish responses.

The function takes three parameters:

Name	Type	Description
info	{ topicName: string }	callback information
callbackType	string	name of the callback type (see table below).
data	[Object]	data that is passed with the callback (depends on the callback type).

Callback types:

Name	Callback Data	Description
publishResponse	PublishResponse	Client received a publish response. The attached data is the publish confirmation for a batch of events.
error	Object	Signals an event publishing error or a gRPC stream error.
grpcKeepalive	{ schemaId: string, rpcId: string }	Server publishes this gRPC keep alive message every 270 seconds (or less) if there are no events.
grpcStatus	Object	Misc gRPC stream status information.

PublishResponse

Name	Type	Description
schemaId	string	topic schema ID
rpcId	string	RPC ID
results	{ replayId: string, correlationKey: string }[]	Event publish confirmations. Each confirmation contains the replay ID and a correlation key.

SubscribeCallback

Callback function that lets you process incoming Pub/Sub API events while keeping track of the topic name and the volume of events requested/received.

The function takes three parameters:

Name	Type	Description
subscription	SubscriptionInfo	subscription information
callbackType	string	name of the callback type (see table below).
data	[Object]	data that is passed with the callback (depends on the callback type).

Callback types:

Name	Callback Data	Description
data	Object	Client received a new event. The attached data is the parsed event data.
error	EventParseError or Object	Signals an event parsing error or a gRPC stream error.
lastevent	void	Signals that we received the last event that the client requested. The stream will end shortly.
end	void	Signals the end of the gRPC stream.
grpcKeepalive	{ latestReplayId: number, pendingNumRequested: number }	Server publishes this gRPC keep alive message every 270 seconds (or less) if there are no events.
grpcStatus	Object	Misc gRPC stream status information.

SubscriptionInfo

Holds the information related to a subscription.

Name	Type	Description
isManaged	boolean	whether this is a managed event subscription or not.
topicName	string	topic name for this subscription.
subscriptionId	string	managed subscription ID. Undefined for regular subscriptions.
subscriptionName	string	managed subscription name. Undefined for regular subscriptions.
requestedEventCount	number	number of events that were requested when subscribing.
receivedEventCount	number	the number of events that were received since subscribing.
lastReplayId	number	replay ID of the last processed event or null if no event was processed yet.

EventParseError

Holds the information related to an event parsing error. This class attempts to extract the event replay ID from the event that caused the error.

Name	Type	Description
message	string	The error message.
cause	Error	The cause of the error.
replayId	number	The replay ID of the event at the origin of the error. Could be undefined if we're not able to extract it from the event data.
event	Object	The un-parsed event data at the origin of the error.
latestReplayId	number	The latest replay ID that was received before the error.

Configuration

Check out the authentication section for more information on how to provide the right values.

Name	Type	Description
authType	string	Authentication type. One of user-supplied, username-password, oauth-client-credentials or oauth-jwt-bearer.
pubSubEndpoint	string	A custom Pub/Sub API endpoint. The default endpoint api.pubsub.salesforce.com:7443 is used if none is supplied.
accessToken	string	Salesforce access token.
instanceUrl	string	Salesforce instance URL.
organizationId	string	Optional organization ID. If you don't provide one, we'll attempt to parse it from the accessToken.
loginUrl	string	Salesforce login host. One of https://login.salesforce.com, https://test.salesforce.com or your domain specific host.
clientId	string	Connected app client ID.
clientSecret	string	Connected app client secret.
privateKey	string	Private key content.
username	string	Salesforce username.
password	string	Salesforce user password.
userToken	string	Salesforce user security token.
rejectUnauthorizedSsl	boolean	Optional flag used to accept self-signed SSL certificates for testing purposes when set to false. Default is true (client rejects self-signed SSL certificates).