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.

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.

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

// Subscribe to account change events
const eventEmitter = await client.subscribe(
    '/data/AccountChangeEvent'
);

// Handle incoming events
eventEmitter.on('data', (event) => {
    // Event handling logic goes here
}):

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.

const subscribeCallback = (subscription, callbackType, data) => {
    // Event handling logic goes here
};

// Subscribe to account change events
await client.subscribe('/data/AccountChangeEvent', subscribeCallback);

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

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)

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

const client = new PubSubApiClient({
    authType: 'user-supplied',
    accessToken: sfConnection.accessToken,
    instanceUrl: sfConnection.instanceUrl,
    organizationId: sfConnection.userInfo.organizationId
});

const client = new PubSubApiClient({
    authType: 'username-password',
    loginUrl: process.env.SALESFORCE_LOGIN_URL,
    username: process.env.SALESFORCE_USERNAME,
    password: process.env.SALESFORCE_PASSWORD,
    userToken: process.env.SALESFORCE_TOKEN
});

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

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

// Read private key file
const privateKey = fs.readFileSync(process.env.SALESFORCE_PRIVATE_KEY_FILE);

// Build PubSub client
const client = new PubSubApiClient({
    authType: 'oauth-jwt-bearer',
    loginUrl: process.env.SALESFORCE_JWT_LOGIN_URL,
    clientId: process.env.SALESFORCE_JWT_CLIENT_ID,
    username: process.env.SALESFORCE_USERNAME,
    privateKey
});

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:

import pino from 'pino';

const config = {
    /* your config goes here */
};
const logger = pino();
const client = new PubSubApiClient(config, logger);

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:

   npm install salesforce-pubsub-api-client dotenv

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

   SALESFORCE_LOGIN_URL=...
   SALESFORCE_USERNAME=...
   SALESFORCE_PASSWORD=...
   SALESFORCE_TOKEN=...

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

   import * as dotenv from 'dotenv';
   import PubSubApiClient from 'salesforce-pubsub-api-client';
   
   async function run() {
       try {
           // Load config from .env file
           dotenv.config();
   
           // Build and connect Pub/Sub API client
           const client = new PubSubApiClient({
               authType: 'username-password',
               loginUrl: process.env.SALESFORCE_LOGIN_URL,
               username: process.env.SALESFORCE_USERNAME,
               password: process.env.SALESFORCE_PASSWORD,
               userToken: process.env.SALESFORCE_TOKEN
           });
           await client.connect();
   
           // Prepare event callback
           const subscribeCallback = (subscription, callbackType, data) => {
               switch (callbackType) {
                   case 'event':
                       // Event received
                       console.log(
                           `${subscription.topicName} - Handling ${data.payload.ChangeEventHeader.entityName} change event ` +
                               `with ID ${data.replayId} ` +
                               `(${subscription.receivedEventCount}/${subscription.requestedEventCount} ` +
                               `events received so far)`
                       );
                       // Safely log event payload as a JSON string
                       console.log(
                           JSON.stringify(
                               data,
                               (key, value) =>
                                   /* Convert BigInt values into strings and keep other types unchanged */
                                   typeof value === 'bigint'
                                       ? value.toString()
                                       : value,
                               2
                           )
                       );
                       break;
                   case 'lastEvent':
                       // Last event received
                       console.log(
                           `${subscription.topicName} - Reached last of ${subscription.requestedEventCount} requested event on channel. Closing connection.`
                       );
                       break;
                   case 'end':
                       // Client closed the connection
                       console.log('Client shut down gracefully.');
                       break;
               }
           };
   
           // Subscribe to 3 account change event
           client.subscribe('/data/AccountChangeEvent', subscribeCallback, 3);
       } catch (error) {
           console.error(error);
       }
   }
   
   run();

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)
   {
       "id": "9b77cea1-a923-4766-ad50-a1797d9b39fd",
       "schemaId": "a01VpgrsZNJdn-7KStJcxQ",
       "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.

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