See the official Pub/Sub API repo and the documentation for more information on the Salesforce gRPC-based Pub/Sub API.
[!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.
In v4 and earlier versions of this client:
.env
file with specific property names.connect()
or connectWithAuth()
method depending on the authentication flow.In v5:
.env
file is no longer a requirement, you are free to store your configuration where you want.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:
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
});
[!WARNING] Relying on a username/password authentication flow for production is not recommended. Consider switching to JWT auth for extra security.
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.
Activate Account change events in Salesforce Setup > Change Data Capture.
Install the client and dotenv
in your project:
npm install salesforce-pubsub-api-client dotenv
Create a .env
file at the root of the project and replace the values:
SALESFORCE_LOGIN_URL=...
SALESFORCE_USERNAME=...
SALESFORCE_PASSWORD=...
SALESFORCE_TOKEN=...
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) => {
if (callbackType === 'event') {
// Event received
console.log(
`${subscription.topicName} - ``Handling ${event.payload.ChangeEventHeader.entityName} change event ` +
`with ID ${event.replayId} ` +
`(${subscription.receivedEventCount}/${subscription.requestedEventCount} ` +
`events received so far)`
);
// Safely log event payload as a JSON string
console.log(
JSON.stringify(
event,
(key, value) =>
/* Convert BigInt values into strings and keep other types unchanged */
typeof value === 'bigint'
? value.toString()
: value,
2
)
);
} else if (callbackType === 'lastEvent') {
// Last event received
console.log(
`${subscription.topicName} - Reached last of ${subscription.requestedEventCount} requested event on channel. Closing connection.`
);
} else if (callbackType === 'end') {
// Client closed the connection
console.log('Client shut down gracefully.');
}
};
// Subscribe to 3 account change event
client.subscribe('/data/AccountChangeEvent', subscribeCallback, 3);
} catch (error) {
console.error(error);
}
}
run();
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.
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.
Publish a Sample__e
Platform Event with a Message__c
field:
const payload = {
CreatedDate: new Date().getTime(), // Non-null value required but there's no validity check performed on this field
CreatedById: '005_________', // Valid user ID
Message__c: { string: 'Hello world' } // Field is nullable so we need to specify the 'string' type
};
const publishResult = await client.publish('/event/Sample__e', payload);
console.log('Published event: ', JSON.stringify(publishResult));
Subscribe to 5 account change events starting from a replay ID:
await client.subscribeFromReplayId(
'/data/AccountChangeEvent',
subscribeCallback,
5,
17092989
);
Subscribe to the 3 earliest past account change events in the retention window:
await client.subscribeFromEarliestEvent(
'/data/AccountChangeEvent',
subscribeCallback,
3
);
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:
lastevent
callback type from subscribe callback to detect the end of the event batch.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:
try {
// Connect with the Pub/Sub API
const client = new PubSubApiClient(/* config goes here */);
await client.connect();
// Prepare event callback
const subscribeCallback = (subscription, callbackType, data) => {
if (callbackType === 'event') {
// Logic for handling a single event.
// Unless you request additional events later, this should get called up to 10 times
// given the initial subscription boundary.
} else if (callbackType === 'lastEvent') {
// Last event received
console.log(
`${eventEmitter.getTopicName()} - Reached last requested event on channel.`
);
// Request 10 additional events
client.requestAdditionalEvents(eventEmitter, 10);
} else if (callbackType === 'end') {
// Client closed the connection
console.log('Client shut down gracefully.');
}
};
// Subscribe to a batch of 10 account change event
await client.subscribe('/data/AccountChangeEvent', subscribeCallback 10);
} catch (error) {
console.error(error);
}
Use callback types from subscribe callback to handle gRPC stream lifecycle events:
const subscribeCallback = (subscription, callbackType, data) => {
if (callbackType === 'grpcStatus') {
// Stream status update
console.log('gRPC stream status: ', status);
} else if (callbackType === 'error') {
// Stream error
console.error('gRPC stream error: ', JSON.stringify(error));
} else if (callbackType === 'end') {
// Stream end
console.log('gRPC stream ended');
}
};
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:
// Safely log event as a JSON string
console.log(
JSON.stringify(
event,
(key, value) =>
/* Convert BigInt values into strings and keep other types unchanged */
typeof value === 'bigint' ? value.toString() : value,
2
)
);
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.
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>}
Get connectivity state from current channel.
Returns: Promise that holds the channel's connectivity state.
async publish(topicName, payload, [correlationKey]) → {Promise.<PublishResult>}
Publishes a payload to a topic using the gRPC client.
Returns: Promise holding a PublishResult
object with replayId
and correlationKey
.
Name | Type | Description |
---|---|---|
topicName |
string | name of the topic that we're subscribing to |
payload |
Object | |
correlationKey |
string | optional correlation key. If you don't provide one, we'll generate a random UUID for you. |
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 |
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. |
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. |
Holds the information related to a subscription.
Name | Type | Description |
---|---|---|
topicName |
string | topic name for this subscription. |
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. |
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. |
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. |