# developer Pub/Sub API (event-bus) - Full Documentation # API End-of-Life Policy Salesforce is committed to supporting Pub/Sub API for a minimum of three years from the date of first release, which was June 30, 2022. In order to mature and improve the quality and performance of the API, versions that are more than three years old sometimes are no longer supported. Salesforce notifies customers who use an API version scheduled for deprecation at least one year before support for the version ends. # Authentication Pub/Sub API supports any authentication mechanism that enables retrieving the session ID, including OAuth, and username and password authentication. The OAuth access token value or session ID is part of the authentication metadata header that the client passes to the Pub/Sub API RPC methods. For more information about gRPC metadata, see [Metadata](https://www.grpc.io/docs/what-is-grpc/core-concepts/#metadata) in [Core concepts, architecture and lifecycle](https://www.grpc.io/docs/what-is-grpc/core-concepts) in the gRPC documentation. For more information about authorizing your app with OAuth, see [OAuth Authorization Flows](https://help.salesforce.com/s/articleView?id=xcloud.remoteaccess_oauth_flows.htm&type=5) in _Salesforce Help_. In each RPC method, use these key-value pairs in the metadata headers. For more information about using headers in Pub/Sub API, see [Include Authorization Headers in RPC Method Calls](https://developer.salesforce.com/docs/platform/pub-sub-api/guide/rpc-method-headers.md). | Metadata Header Name | Metadata Header Value | | -------------------- | ----------------------------------------------------------------------------------------------------------- | | `accesstoken` | The access token if using OAuth, or the Salesforce session ID if using username and password authentication | | `instanceurl` | URL of the Salesforce instance. For example, `https://MyDomainName.my.salesforce.com` | | `tenantid` | The Salesforce org ID | If you use OAuth, the access token expiration is controlled by the Salesforce session timeout value in the connected app, or if not set, in the user’s profile or the session settings. The session times out after 2 hours of inactivity by default, unless a custom timeout value is set. For more information, see [Manage Session Policies for a Connected App](https://help.salesforce.com/s/articleView?id=xcloud.connected_app_manage_session_policies.htm&type=5) in _Salesforce Help_. However, the session timeout doesn’t matter for a subscriber client because the Subscribe RPC method keeps the connection alive through keepalive messages if there are still pending events to be delivered. For more details, see [Subscribe Keepalive Behavior](https://developer.salesforce.com/docs/platform/pub-sub-api/references/methods/subscribe-rpc.html#subscribe-keepalive-behavior). For the PublishStream RPC method, the connection closes if no new publish request is sent within a short time. See [PublishStream RPC Method.](https://developer.salesforce.com/docs/platform/pub-sub-api/references/methods/publishstream-rpc.html) The rest of the RPC methods execute promptly and don’t need a long-lived connection. # Bidirectional Streaming Bidirectional streaming is one of the four types of RPC methods that can be defined in a gRPC API. With bidirectional streaming, both the client and the server can send a sequence of messages to each other by using two independent streams. It’s not necessary for the client to wait until the server finishes sending all the messages to send new requests. Similarly, the server doesn’t wait until the client has sent all the messages before responding. ![Flow of bidirectional streaming between the client and the Pub/Sub API service](https://developer.salesforce.com/docs/media/pubsub-bidirectional-streaming.png '{"class": "image-lg", "title": "Flow of bidirectional streams between the client and the Pub/Sub API service"}') The Subscribe and ManagedSubscribe (beta) methods use bidirectional streaming to subscribe to an Event Bus topic. The PublishStream method also uses bidirectional streaming. For more information, see [Core concepts, architecture and lifecycle](https://grpc.io/docs/what-is-grpc/core-concepts/) in the gRPC documentation. **See Also** - [Subscribe RPC Method](/docs/platform/pub-sub-api/references/methods/subscribe-rpc.html) - [ManagedSubscribe RPC Method (Beta)](/docs/platform/pub-sub-api/references/methods/managedsubscribe-rpc.html) - [PublishStream RPC Method](/docs/platform/pub-sub-api/references/methods/publishstream-rpc.html) - [Pull Subscription and Flow Control](https://developer.salesforce.com/docs/platform/pub-sub-api/guide/flow-control.md) # Code Examples In addition to the quick-start examples, the [Pub/Sub API GitHub repository](https://github.com/forcedotcom/pub-sub-api) contains additional code examples in various programming languages so that you can learn how to use Pub/Sub API in those languages. **note:** Even though Pub/Sub API supports multiple programming languages through the gRPC API, we’re committed to officially support the Java code samples. Use the Java code samples to learn how to implement Pub/Sub API. Before you deploy the code to production, perform enhancements and thorough functional and performance testing. See [Java Quick Start](https://developer.salesforce.com/docs/platform/pub-sub-api/guide/qs-java-quick-start-parent.md) and the [java folder](https://github.com/forcedotcom/pub-sub-api/tree/main/java) in the Pub/Sub API GitHub repository. Samples in other programming languages are provided in this guide and in the [Pub/Sub API GitHub repository](https://github.com/forcedotcom/pub-sub-api) but aren't officially supported. ## Go Code Examples The [Go code examples](https://github.com/forcedotcom/pub-sub-api/tree/main/go) are a collection of code examples in the Go programming language for each RPC method call. The Go examples use OAuth for authorization. You can use these examples to learn how to implement Pub/Sub API in Go. ## Community-Provided Code Examples For code examples that the developer community created, see the [Code Samples from the Developer Community](https://github.com/forcedotcom/pub-sub-api#code-samples-from-the-developer-community) section in the [Pub/Sub API GitHub repository](https://github.com/forcedotcom/pub-sub-api). Salesforce doesn’t support these examples, so use them at your own discretion. **See Also** - [Quick Starts](https://developer.salesforce.com/docs/platform/pub-sub-api/guide/quick-start.md) # Event Data Serialization with Apache Avro Pub/Sub API publishes and delivers binary-encoded events with payloads encoded in the Apache Avro schema. Apache Avro is a data serialization system that provides a binary or JSON data encoding and a schema. The Avro binary encoding is more efficient than the Avro JSON encoding because it enables faster serialization and produces smaller sizes of serialized data. For more information, see [Data Serialization and Deserialization](https://avro.apache.org/docs/++version++/specification/#data-serialization-and-deserialization) in the Apache Avro specification. When sending or receiving events from Pub/Sub API, your app must use Apache Avro to serialize and deserialize event payloads. Before you publish an event message, encode it to the Avro format. And when you receive an event message, decode it using the Avro format before you can retrieve the contents of the event payload. For more information, see [Pub/Sub API as a gRPC API](https://developer.salesforce.com/docs/platform/pub-sub-api/guide/grpc-api.md) and the [Apache Avro Documentation](https://avro.apache.org/docs/current/index.html). The [Python Quick Start for Pub/Sub API](https://developer.salesforce.com/docs/platform/pub-sub-api/guide/qs-python-quick-start.md) includes example functions for encoding and decoding the event messages using Avro. Pub/Sub API doesn’t validate the Avro format of the event payload that’s passed in to the [Publish](/docs/platform/pub-sub-api/references/methods/publish-rpc.html) or [PublishStream](/docs/platform/pub-sub-api/references/methods/publishstream-rpc.html) calls. It’s up to the client to make sure that the event payload is in Avro format and that it’s valid. The `Publish` and `PublishStream` calls complete successfully even if an event payload isn’t in valid Avro format. Because an event with an invalid payload fails to deserialize, a subscriber must filter out the events with invalid payloads. **See Also** - [Event Deserialization Considerations](https://developer.salesforce.com/docs/platform/pub-sub-api/guide/event-deserialization-considerations.md) # Event Deserialization Considerations The Pub/Sub API delivers events in the raw Avro binary format without modifying the fields in the event message. As a result, some fields in the deserialized event aren’t readable when you print them for debugging, and they must be decoded before they can be processed in the subscriber. Also, the fields received in change data capture events with Pub/Sub API and Streaming API (CometD) have differences. ## Bitmap Fields in ChangeEventHeader in Change Events Change data capture events contain bitmap fields whose contents aren’t readable when printed, and they must be decoded for processing in the subscriber app logic. The Pub/Sub API delivers the events in the raw Avro binary format without converting these fields. The bitmap fields are: - `changedFields` - : Contains the fields that were changed. - `diffFields` - : Contains the fields whose values were sent as a data diff. - `nulledFields` - : Contains the fields that were set to null. When you print these bitmap fields, you get a list of hexadecimal values and not the field names. For example: `'changedFields': ['0x650004E0','4-0x10','7-0x06']` A bitmap field uses a bitmap, encoded as a hexadecimal string, to represent the fields. This method is more space efficient than using a list of field names. For example, a bit set to 1 indicates that the field at the corresponding position in the Avro schema was changed for `changedFields`. A bitmap field is an array of strings. The first element of the array contains a bitmap for the individual non-compound fields. Compound fields are placed in additional elements of the array, with bitmaps indicating nested fields. The format for the additional array elements is `'{ParentFieldPosition}-{NestedFieldBitmap}'`. In the previous example,`'4-0x10'` represents a compound field. `4` indicates that the compound field is at the fourth position in the Avro schema and `0x10` is the bitmap for the nested fields it contains. **note:** For `nulledFields`, if the only change when updating a record is to set the compound nested fields to null, the first element of the `nulledFields` array doesn’t contain the individual non-compound fields. In that case, the array contains one entry, which is the compound field specification in the `'{ParentFieldPosition}-{NestedFieldBitmap}'` format. For example, the array value can be `['4-0x08']`. ### Bitmap Fields and Field-Level Security The fields referenced in the field bitmap respect the field permissions that the subscriber has. Any bits in the field bitmap for fields that the subscriber doesn’t have permission for are cleared to 0. For compound field bitmaps, `'{ParentFieldPosition}-{NestedFieldBitmap}'`, the nested compound field bits aren’t cleared. Instead, the entry for the compound field is excluded from the field bitmap array. The reason is because field permissions are checked at the compound field level and not at the nested field level. ### Decoding a Bitmap Field To decode the bitmap, match it against the fields in the event schema. You can get the event schema by calling the [GetSchema RPC Method](/docs/platform/pub-sub-api/references/methods/getschema-rpc.html). The GitHub repository contains code examples that show how to decode bitmap fields. In the Java code example, [EventParser.java](https://github.com/forcedotcom/pub-sub-api/blob/main/java/src/main/java/utility/EventParser.java) expands the bitmap fields. [CommonContext.java](https://github.com/forcedotcom/pub-sub-api/blob/main/java/src/main/java/utility/CommonContext.java) uses the `getFieldListFromBitmap()` method from `EventParser` to handle bitmap fields for change events. In the Python code example, [ChangeEventHeaderUtility.py](https://github.com/forcedotcom/pub-sub-api/blob/main/python/util/ChangeEventHeaderUtility.py) in [Pub/Sub API Examples - Utility Code](https://github.com/forcedotcom/pub-sub-api/tree/main/python/util) expands the bitmap fields. In the `SalesforceListener.py` example, the `process_confirmation` function shows how to get and convert the `changedFields` bitmap field by calling the `process_bitmap` function. ### Byte Field Storage Byte fields in protocol buffer messages map to specific field types in various programming languages. See [Scalar Value Types](https://developers.google.com/protocol-buffers/docs/proto3#scalar) in the gRPC documentation. For example, bitmap fields in `ChangeEventHeader` are in bytes and so is the `replay_id` field. You can store byte values in a variable that holds bytes in your programming language. **note:** If you use `ByteBuffer` in Java or JavaScript to store byte values for the bitmap fields in `ChangeEventHeader` or `replay_id`, reverse the bytes. `ByteBuffer` puts the bytes in big-endian order, but Pub/Sub API expects those fields to be in little-endian order. ## Date Fields After you deserialize a received platform event or change event in your client, Date and Date/Time fields are in Epoch time. As a result, when you print them, you get a number. For example: `'CreatedDate': 1632858587281` If you want to make the Date field readable for debugging purposes, convert the Epoch format into another date format. For more information, see [Unix time](https://en.wikipedia.org/wiki/Unix_time). ## Change Event Differences with Streaming API (CometD) Change data capture events that are delivered with the Pub/Sub API contain some fields that aren’t present on the change events received with Streaming API (CometD). This difference is because Streaming API uses JSON encoding for delivered events while Pub/Sub API uses the Avro binary format. - Change events received with Pub/Sub API contain all the record fields, including the unchanged fields. Unchanged fields have an empty value in the change event, for example, `'Description': None`, even if they have a value in the Salesforce record. The `changedFields` field indicates which fields changed. The `nulledFields` field indicates which fields were set to null. In contrast, change events received with Streaming API (CometD) contain only the changed fields and exclude unchanged fields. - Change events received with Pub/Sub API contain header fields that aren’t included in Streaming API events. The header fields are `diffFields` and `nulledFields`. These fields are present on events received in Apex triggers. For more information, see [Change Event Triggers](https://developer.salesforce.com/docs/atlas.en-us.change_data_capture.meta/change_data_capture/cdc_trigger_intro.htm) in the _Change Data Capture Developer Guide_. **See Also** - _Change Data Capture Developer Guide_: [ChangeEventHeader Fields](https://developer.salesforce.com/docs/atlas.en-us.change_data_capture.meta/change_data_capture/cdc_event_fields_header.htm) - _Change Data Capture Developer Guide_: [Get Compound Fields in Change Events](https://developer.salesforce.com/docs/atlas.en-us.change_data_capture.meta/change_data_capture/cdc_subscribe_compound_fields.htm) # Event Message Durability Salesforce stores platform events and change data capture events for 72 hours. Using Pub/Sub API, you can retrieve events that are within the retention window. Retrieving stored events is helpful after the client disconnects and wants to catch up on missed events. **note:** After the retention period, events are purged from the event bus. The purging process sometimes starts later, and as a result, platform events and change data capture events that are older than 72 hours can still be available. Salesforce doesn’t guarantee the storage of events beyond the retention period of 72 hours. ## Event Replay ID Field Each received event message is assigned an opaque ID contained in the Replay ID field, represented as `ConsumerEvent.replay_id` in the protocol specification. The Replay ID field value, which is populated by the system when the event is delivered to subscribers, refers to the position of the event in the event stream. Replay ID values aren’t guaranteed to be contiguous for consecutive events. A subscriber can store a replay ID value and use it on resubscription to retrieve events that are within the retention window. For example, a subscriber can retrieve missed events after a connection failure. Subscribers must not compute new Replay IDs based on a stored Replay ID to refer to other events in the stream. The `replay_id` field value isn’t guaranteed to be unique when Salesforce maintenance activities occur, such as an org migration. To uniquely identify a platform event message, use the event `id` field. When publishing an event, the field is `ProducerEvent.id`. For more information, see [Identify an Event with ProducerEvent.id and PublishResponse.correlationKey](/docs/platform/pub-sub-api/references/methods/publishstream-rpc.html#identify-an-event-with-producereventid-and-publishresponsecorrelationkey). For a received platform event message, use the `ConsumerEvent.event.id` field. The `id` field is present on platform events and not on change events. ## Retrieving Stored Events To catch up on missed events during client shutdown, you can retrieve events that are temporarily stored in the event bus. By using the [Subscribe RPC Method](/docs/platform/pub-sub-api/references/methods/subscribe-rpc.html) with a `replay_preset` option, a subscriber can choose which events to receive, such as all events starting after a particular Replay ID. To do so, the client stores the Replay ID of the last processed event. This high-level diagram shows how event consumers can read a stream of events by using various replay options. ![Replay stored events](https://developer.salesforce.com/docs/media/pubsub-replaying-events.png '{ "title": "Replay stored events"}') Instead of managing the Replay ID on the client with the `Subscribe` RPC method, you can use the [ManagedSubscribe RPC Method (Beta)](/docs/platform/pub-sub-api/references/methods/managedsubscribe-rpc.html) to more easily track the events received. With `ManagedSubscribe`, you commit the Replay ID of the last processed event on the server. You don’t worry about managing a replay ID store on the client. Then when resubscribing, the subscription resumes after the stored Replay ID. On rare occasions, the stream of retained events can be reset if the Salesforce org is moved to a new instance. As a result, you can no longer access the retained events. See [Salesforce Maintenance Activities and Sandbox Refresh](https://developer.salesforce.com/docs/atlas.en-us.platform_events.meta/platform_events/platform_events_api_considerations.htm#maintenance_activities) in the _Platform Events Developer Guide_. **See Also** - [_GitHub_: Pub/Sub API proto file](https://github.com/forcedotcom/pub-sub-api/blob/main/pubsub_api.proto) # Event Types Pub/Sub API supports these types of events. - High-volume platform events, including custom and standard events. For more information about custom platform events, see [Defining Your Custom Platform Event](https://developer.salesforce.com/docs/atlas.en-us.platform_events.meta/platform_events/platform_events_define.htm) in the _Platform Events Developer Guide_. For more information about standard events, see [Standard Platform Event Object List](https://developer.salesforce.com/docs/atlas.en-us.platform_events.meta/platform_events/platform_events_objects_list.htm) in the _Platform Events Developer Guide_. - Real-time event monitoring events. See [Real-Time Event Monitoring Objects](https://developer.salesforce.com/docs/atlas.en-us.platform_events.meta/platform_events/platform_events_objects_monitoring.htm) in the _Platform Events Developer Guide_. - Change data capture events. See [_Change Data Capture Developer Guide_](https://developer.salesforce.com/docs/atlas.en-us.change_data_capture.meta/change_data_capture/cdc_intro.htm). Pub/Sub API doesn’t support legacy events, such as standard-volume platform events, PushTopic events, and generic streaming events. # General Considerations Keep these considerations in mind when using Pub/Sub API. ## AWS Regions Pub/Sub API uses, for processing purposes, infrastructure that’s provided by Amazon Web Services, Inc. and located in the regions listed below. By using the Pub/Sub API, you agree to the use of AWS infrastructure for processing purposes and acknowledge that such processing can occur outside the region where your production org is located. Event processing can cause data in transit to pass through and exist temporarily in an external region. If you use the Pub/Sub API endpoint specific to the region where your org is located, event processing occurs in the same region as your org. For more information, see [Pub/Sub API Endpoints](https://developer.salesforce.com/docs/platform/pub-sub-api/guide/pub-sub-endpoints.md) and [Regions, Availability Zones, and Local Zones](https://docs.aws.amazon.com/AmazonRDS/latest/UserGuide/Concepts.RegionsAndAvailabilityZones.html) in the AWS documentation. Regions: - Asia Pacific (Mumbai, Sydney, Tokyo) - Canada (Central) - Europe (Frankfurt, London) - South America (São Paulo) - US East (N. Virginia) - US West (Oregon) Events in the Salesforce event bus aren't persisted or stored outside the region where your org is located. ## Channel Membership and User Permission Changes If channel membership or user permissions change after a subscription starts and while it’s active, the changes take effect within 10 minutes. The same delay applies to publishing platform events. Here are some examples. - If a new member is added to a channel, the client receives events corresponding to the new member within 10 minutes. - If a user loses their View All permission on a Salesforce object included in a channel member, they stop receiving the object’s change events within 10 minutes. - If a channel is deleted, the subscription ends with an error within 10 minutes. A channel can be a change data capture channel or a custom platform event channel. - If a user loses their create permission for a platform event, they can’t publish the event anymore after 10 minutes or less. For more information about change data capture channels and permissions, see [Subscription Channels](https://developer.salesforce.com/docs/atlas.en-us.change_data_capture.meta/change_data_capture/cdc_subscribe_channels.htm) and [Required Permissions for Change Events](https://developer.salesforce.com/docs/atlas.en-us.change_data_capture.meta/change_data_capture/cdc_security_perms.htm) in the _Change Data Capture Developer Guide_. For more information about custom platform event channels, see [Group Platform Events into One Stream with a Custom Channel](https://developer.salesforce.com/docs/atlas.en-us.platform_events.meta/platform_events/platform_events_subscribe_custom_channels.htm) in the _Platform Events Developer Guide_. ## Multiple Subscriptions and Duplicate Events If you start multiple subscriptions for the same Salesforce org and topic, each subscription stream is independent of the others. The same events are delivered to each subscriber client. Add some logic in your client to handle duplicate events. The system-generated `id` field value in the received events uniquely identifies each event. You can dedupe events using the `id` field. If a Pub/Sub API publisher overrides the `id` field value, it's possible that the `id` isn't unique. See [Identify an Event with ProducerEvent.id and PublishResponse.correlationKey](https://review-app-sdevendoc-pr-88.herokuapp.com/docs/platform/pub-sub-api/references/methods/publishstream-rpc.html#identify-an-event-with-producereventid-and-publishresponsecorrelationkey). The `id` field corresponds to the `EventUuid` field in events delivered to non-Pub/Sub API clients. See [Identify and Match Event Messages with the EventUuid Field](https://developer.salesforce.com/docs/atlas.en-us.platform_events.meta/platform_events/platform_events_event_uuid.htm) in the _Platform Events Developer Guide_. ## IP Address Session Setting The `Lock sessions to the IP address from which they originated` setting in the Session Settings page in Setup must be disabled for Pub/Sub API. If you enable this setting, Pub/Sub API RPC calls fail. For more information, see [Modify Session Security Settings](https://help.salesforce.com/s/articleView?id=xcloud.admin_sessions.htm&type=5) in _Salesforce Help_. # Generate a Client Trace ID to Troubleshoot Non-Pub/Sub API Errors To help with troubleshooting errors that don’t originate from Pub/Sub API and don’t include the RPC ID, add the `x-client-trace-id` metadata to every RPC request that the client makes to Pub/Sub API as part of the authorization headers. Some gRPC exception errors that Pub/Sub API returns don’t originate from the Pub/Sub API service, such as network errors or errors returned from the data center infrastructure. These gRPC exceptions don’t have an [RPC ID](https://developer.salesforce.com/docs/platform/pub-sub-api/guide/rpc-id-field.md) value in the error message or the `rpc-id` field in the exception trailer because Pub/Sub API adds the RPC ID only to errors it generates. It doesn’t add the RPC ID to errors that an external source returns. The RPC ID helps Salesforce Customer Support with tracing a request and troubleshooting the error. To help with troubleshooting errors that don’t contain an RPC ID, the client can add the `x-client-trace-id` metadata with a generated unique ID. Here are some examples of gRPC exceptions that don’t contain an RPC ID. These errors aren’t listed in [Error Codes](https://developer.salesforce.com/docs/platform/pub-sub-api/guide/handling-errors.md#error-codes) because they aren’t generated by Pub/Sub API. | gRPC Status Code | Error Message | | ---------------- | --------------------------------------------------------------------------- | | UNAVAILABLE | Http2StreamException: The HTTP/2 server reset the stream. HTTP/2 error code | | UNAVAILABLE | This application is down for maintenance | With the `x-client-trace-id` metadata value present in every request, it’s possible to trace the request from the client to the Pub/Sub API service without the RPC ID value. When an error occurs, you can contact Salesforce Customer Support and provide the x`-client-trace-id` and timestamp for the failed request. Salesforce can use this information for debugging purposes. Here are the steps to add the `x-client-trace-id` metadata header to each RPC request. 1. Create a client interceptor that adds a unique `x-client-trace-id` to the metadata headers. The client interceptor also logs the start and end of the request. Clients can use this information to trace the `x-client-trace-id` for each request and provide this information to Salesforce Customer Support when needed. To learn more about interceptors, see [Interceptors](https://grpc.io/docs/guides/interceptors/) in the gRPC Docs. For an example implementation in the sample `pub-sub-api` Java client, see [XClientTraceIdClientInterceptor.java](https://github.com/forcedotcom/pub-sub-api/blob/main/java/src/main/java/utility/XClientTraceIdClientInterceptor.java) in GitHub. 2. Configure the gRPC channel to go through the client interceptor that you created so that each request has a unique `x-client-trace-id`. For an example implementation in the sample `pub-sub-api` Java client, see `CommonContext()` in [CommonContext.java](https://github.com/forcedotcom/pub-sub-api/blob/main/java/src/main/java/utility/CommonContext.java) in GitHub. **note:** The types of errors that originate from an external source are typically transient. When a client catches these errors, we recommend that it retries the RPC method with exponential backoff. **See Also** - [Handle Errors](https://developer.salesforce.com/docs/platform/pub-sub-api/guide/handling-errors.md) - [Provide the RPC ID When Contacting Support to Troubleshoot an Error](https://developer.salesforce.com/docs/platform/pub-sub-api/guide/rpc-id-field.md) # Generate Code from the Proto File To generate code from the proto file, use a gRPC plug-in with protoc. This process generates client code, server code, and protocol buffer code for populating, serializing, and retrieving message types. For more information, see [Introduction to gRPC](https://grpc.io/docs/what-is-grpc/introduction/) in the gRPC documentation. The [quick starts](https://developer.salesforce.com/docs/platform/pub-sub-api/guide/quick-start.md) walk you through the steps of generating the code from the proto file using gRPC tools. ## Execute RPC Methods from the Generated Code The client has a local object known as a stub that implements the service methods. When a gRPC client calls the API, the corresponding API implementation is called on the server. The gRPC infrastructure decodes incoming requests, executes service methods, and encodes service responses. The client calls the methods on the local object, wrapping the parameters for the call in the appropriate protocol buffer message type. gRPC handles sending the requests to the server and returning the server’s protocol buffer responses. The gRPC server serializes the response of a Subscribe RPC call based on the protocol buffer message type specified in the proto file. Similarly, when an event is published through one of the publish RPC methods, the publish request is serialized based on the protocol buffer message type. The payload of a published or received event is in the Avro binary format. For more information, see [Protocol Buffers](https://developers.google.com/protocol-buffers/docs/overview) in the gRPC documentation and [Event Data Serialization with Apache Avro](https://developer.salesforce.com/docs/platform/pub-sub-api/guide/event-avro-serialization.md). **See Also** - [_GitHub_: Pub/Sub API proto file](https://github.com/forcedotcom/pub-sub-api/blob/main/pubsub_api.proto) # Get Started with Pub/Sub API Pub/Sub API provides a single interface for publishing and subscribing to platform events, including real-time event monitoring events, and change data capture events. Based on [gRPC API](https://grpc.io/docs/) and HTTP/2, Pub/Sub API efficiently publishes and delivers binary event messages in the Apache Avro format. | REQUIRED EDITIONS | | ---------------------------------------------------------------------------------------- | | Available in: **Enterprise**, **Performance**, **Unlimited**, and **Developer** Editions | | Not Available in: Government Cloud | ## Discover Learn about Pub/Sub API by checking out the pages under these sections. - [Get Started with Pub/Sub API](https://developer.salesforce.com/docs/platform/pub-sub-api/guide/intro.md) (this section) - [Use Pub/Sub API](https://developer.salesforce.com/docs/platform/pub-sub-api/guide/use-pub-sub-api.md) ## Build an App Learn how to create your first subscriber app and discover code samples. - [Quick Starts](https://developer.salesforce.com/docs/platform/pub-sub-api/guide/quick-start.md) - [Code Examples](https://developer.salesforce.com/docs/platform/pub-sub-api/guide/other-code-examples.md) - [Reference](/docs/platform/pub-sub-api/references/) ## Use Cases With Pub/Sub API, you can build event-driven integration apps. Here are some examples of what you can do. - Publish and subscribe to custom platform events to send and receive real-time notifications in another system for business process integration. Some examples of business process integrations are sending real-time notifications for maintenance service or product orders. - Subscribe to change data capture events, and synchronize order data in an external inventory system. - Subscribe to Event Monitoring real-time events, and publish a platform event back into Salesforce to restrict a user’s profile when they log into Salesforce after working hours. - Subscribe to a standard platform event, such as AppointmentSchedulingEvent, and integrate with Google Calendar to update users’ calendars. ## Pub/Sub API Benefits Learn about what Pub/Sub API offers. - Publishing, subscribing, and event schema retrieval all in one API. - Final publish results of publish operations, and not intermediate queueing results. - Flow control that lets you specify how many events to receive in a subscribe call based on event processing speed in the client. - Scalable and secure publishing and delivery of platform events, change data capture events, and real-time event monitoring events. - Real-time, highly performant data streaming that uses compression through HTTP/2. - Support for 11 programming languages in the client that are offered by the gRPC API, such as Python, Java, Node, and C++. For all the supported languages, see [https://grpc.io/docs/languages/](https://grpc.io/docs/languages/). - An active online developer community presence for gRPC. - Bidirectional data streaming through the gRPC API. The client and the server can send a sequence of messages to each other using two independent streams. - Cross-cloud integration capabilities enabling the development of event-driven apps that integrate across Salesforce clouds. ## Pub/Sub API vs. Other Salesforce APIs Simplify your development with Pub/Sub API by using a single API to work with events instead of using disparate APIs. This table compares the tasks that you perform with Pub/Sub API versus other APIs. | **Task** | **Can accomplish with Pub/Sub API** | **How to accomplish with other APIs** | | ------------------------------------------------------------------------------------------------------------------------ | ----------------------------------- | --------------------------------------- | | Publish platform events | Pub/Sub API | Data APIs: SOAP API, REST API, Bulk API | | Subscribe to events: platform events, change data capture events, real-time event monitoring events, and standard events | Pub/Sub API | Streaming API | | Get the event schema | Pub/Sub API | REST API | **See Also** - [Pub/Sub API and the Expanded Event Bus](https://developer.salesforce.com/docs/platform/pub-sub-api/guide/expanded-event-bus.md) - [Pub/Sub API Features](https://developer.salesforce.com/docs/platform/pub-sub-api/guide/pub-sub-features.md) # GetSchema RPC Method The `GetSchema` method returns the schema of an event topic using the schema ID. Use the schema to encode the payload in the Avro format of the event to publish, or to decode the payload of a received event. **important:** Don’t call `GetSchema` for every RPC method call. The schema typically doesn’t change often, so call `GetSchema` only one time and store the returned schema for subsequent calls. If the event schema changes, for example, when an administrator adds a field to the event definition, the schema ID changes. Compare the schema ID with the latest schema ID retrieved from PublishResponse or FetchResponse. If the schema ID changes, call `GetSchema` to retrieve the new schema. Calling `GetSchema` unnecessarily can slow down your app’s performance. ``` rpc GetSchema (SchemaRequest) returns (SchemaInfo); ``` To get the schema ID for the `SchemaRequest` parameter, do one of the following. - Call `GetTopic`. The return value of this method is TopicInfo. TopicInfo contains `schema_id`, which represents the latest schema. We recommend that you publish events with the latest schema. - For events received from the event bus, get the schema ID from the event message in the FetchResponse, `ProducerEvent.schema_id`. Use this schema for deserialization. For events published to the event bus, get the schema ID from the PublishResponse. **note:** You can still publish events with an old schema saved from an earlier `GetTopic` call. You use an Avro code generator to generate classes based on Avro types and deploy your app. If the event schema changes later, you can still publish and subscribe to the events as long as the schema differences are resolvable by the Avro schema resolution rules. For more information about the fields in SchemaRequest and SchemaInfo, see the [Pub/Sub API proto file](https://github.com/forcedotcom/pub-sub-api/blob/main/pubsub_api.proto). # GetTopic RPC Method The `GetTopic` RPC method returns information for an event, such as the event topic name and the schema ID. The topic is the API name of the event object preceded by a path. The topic indicates the type of event to publish and the type of event to subscribe to. For example, the topic of a custom platform event with the API name of `Order_Event__e` is `/event/Order_Event__e`. Pass in the topic name in `TopicRequest`. The schema ID, that's returned in `TopicInfo`, is used to get the event schema with `GetSchema`. ``` rpc GetTopic (TopicRequest) returns (TopicInfo); ``` For more information about the fields in TopicRequest and TopicInfo, see the [Pub/Sub API proto file](https://github.com/forcedotcom/pub-sub-api/blob/main/pubsub_api.proto). # Handle Errors If an error occurs while an RPC method executes, Pub/Sub API throws a gRPC [StatusRuntimeException](https://grpc.github.io/grpc-java/javadoc/io/grpc/StatusRuntimeException.html) that contains a status code. The exception causes the long-lived connection for the `PublishStream`, `Subscribe`, and `ManagedSubscribe (beta)` RPC methods to stop. Retry the RPC methods to create a new long-lived connection. For details on how to retry the RPC methods, see [Retry Long-Lived RPC Calls After an Error Occurs](https://developer.salesforce.com/docs/platform/pub-sub-api/guide/retry-rpc-calls.md). ## Inspect the Error Returned The gRPC status codes can be found [here](https://grpc.github.io/grpc/core/md_doc_statuscodes.html). In your code, catch the exceptions after performing an RPC method call, and handle the error. The status fields differ depending on the programming language. For example, in Java, the fields are `code` and `description` and in Python they're `code` and `details`. To get status fields in Java, after catching the exception, you can call the [getStatus()](https://grpc.github.io/grpc-java/javadoc/io/grpc/StatusRuntimeException.html#getStatus--) Java method on the exception to get the [Status](https://grpc.github.io/grpc-java/javadoc/io/grpc/Status.html) object. Next, call [getCode()](https://developer.salesforce.com/docs/platform/pub-sub-api/guide/%3Chttps://grpc.github.io/grpc-java/javadoc/io/grpc/Status.html#getCode()>) and [getDescription()](https://developer.salesforce.com/docs/platform/pub-sub-api/guide/%3Chttps://grpc.github.io/grpc-java/javadoc/io/grpc/Status.html#getDescription()>) on the status. To get status fields in Python, after catching the exception, you can call the [code](https://grpc.github.io/grpc/python/grpc.html#grpc.Call.code) and [details](https://grpc.github.io/grpc/python/grpc.html#grpc.Call.details) methods on the exception. Pub/Sub API adds a custom error code in the Trailers section of the exception. You can retrieve the custom error code by calling [getTrailers()](https://developer.salesforce.com/docs/platform/pub-sub-api/guide/%3Chttps://grpc.github.io/grpc-java/javadoc/io/grpc/StatusRuntimeException.html#getTrailers()>) on the exception. The custom error code provides information about the cause of the failure. Also, the exception contains an RPC ID that the Pub/Sub API appends to the error message after the “`rpcId:`" prefix. The RPC ID is also included in the Trailers section of the exception. The RPC ID identifies the method execution that caused the exception and can aid Salesforce Customer Support in troubleshooting the error. If you can’t resolve the error by looking up the error code and the documentation, contact Salesforce Customer Support for help and provide the `rpcId` value. ### Exception Example In the example gRPC exception below, the status returned is `INVALID_ARGUMENT`, and the custom error code from the Pub/Sub API is: `[Trailer] = error-code [Value] = sfdc.platform.eventbus.grpc.subscription.fetch.replayid.corrupted` ``` === GRPC Exception === io.grpc.StatusRuntimeException: INVALID_ARGUMENT: The Replay ID validation failed. Ensure that the Replay ID is valid. rpcId: 2f6b4cee-3525-49d1-8fdb-0bd3d662062f === Trailers === [Trailer] = content-type [Value] = application/grpc [Trailer] = rpc-id [Value] = 2f6b4cee-3525-49d1-8fdb-0bd3d662062f [Trailer] = error-code [Value] = sfdc.platform.eventbus.grpc.subscription.fetch.replayid.corrupted [Trailer] = type [Value] = Subscribe ``` ## Error Codes This table lists the error codes that the Pub/Sub API returns as part of the gRPC [StatusRuntimeException](https://grpc.github.io/grpc-java/javadoc/io/grpc/StatusRuntimeException.html). The gRPC status codes can be found in the [_gRPC documentation_](https://grpc.github.io/grpc/core/md_doc_statuscodes.html). ### General Errors | **Error Code** | **Error Description** | **gRPC Status Code** | | ------------------------------------------------------------------------ | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -------------------- | | `io.grpc.StatusRuntimeException: UNAVAILABLE: RST_STREAM closed stream.` | An internal error occurred. Temporary network connectivity issues can cause this error. Retry the `PublishStream` or `Subscribe` RPC method. | `UNAVAILABLE` | | `sfdc.platform.eventbus.grpc.concurrent.client.limit.exceeded` | The maximum number of concurrent clients across all topics has been exceeded. This limit applies to publisher and subscriber clients. | `RESOURCE_EXHAUSTED` | | `sfdc.platform.eventbus.grpc.service.unavailable` | The Pub/Sub API service is unavailable. Try again later. | `INTERNAL` | | `sfdc.platform.eventbus.grpc.service.auth.error` | An authentication exception occurred. Provide valid authentication via metadata headers. This error can occur if the setting "Lock sessions to the IP address from which they originated" is enabled in Setup. See [IP Address Session Setting](https://developer.salesforce.com/docs/platform/pub-sub-api/guide/considerations.html#ip-address-session-setting). Also, a major Salesforce release update can cause this error because it invalidates the Salesforce session. If you received this error during a major update, fetch a new session token and retry the Pub/Sub API call. | `UNAUTHENTICATED` | | `sfdc.platform.eventbus.grpc.service.auth.headers.invalid` | The auth headers value for the specified key is invalid. Provide valid auth headers. The auth headers can’t be blank. | `UNAUTHENTICATED` | | `sfdc.platform.eventbus.grpc.service.auth.refresh.invalid` | The auth refresh token value is invalid. Provide a valid auth refresh value. | `UNAUTHENTICATED` | | `sfdc.platform.eventbus.grpc.service.protection.stream.limit.triggered` | The service received too many PublishRequests or FetchRequests and doesn’t have the resources to accept new requests. | `RESOURCE_EXHAUSTED` | | `sfdc.platform.eventbus.grpc.service.protection.triggered` | The service received too many connections or requests. It doesn’t have the resources to accept new connections or requests. | `RESOURCE_EXHAUSTED` | | `sfdc.platform.eventbus.grpc.service.tenant.license` | The Salesforce org isn’t licensed to access Pub/Sub API. Contact Salesforce to enable the API. If you get this error even though your org is enabled for Pub/Sub API, retry the RPC method call. The cause of this error can be an intermittent issue with the service when it fails to verify the org’s access to Pub/Sub API. | `PERMISSION_DENIED` | | `sfdc.platform.eventbus.grpc.org.migrated` | The org is currently being migrated to a new instance. Implement a conservative backoff strategy with retries at least 30 minutes apart. | `UNAVAILABLE` | ### Schema Errors | **Error Code** | **Error Description** | **gRPC Status Code** | | ------------------------------------------------------ | ----------------------------------------------------------------------------------------------------------------- | -------------------- | | `sfdc.platform.eventbus.grpc.schema.meta.permission` | An error occurred while getting the schema. Possible reasons are an incorrect schema ID or incorrect credentials. | `PERMISSION_DENIED` | | `sfdc.platform.eventbus.grpc.schema.api.unavailable` | The schema information is unavailable. | `UNAVAILABLE` | | `sfdc.platform.eventbus.grpc.schema.validation.failed` | Schema validation failed. The schema ID can’t be blank. | `INVALID_ARGUMENT` | ### Topic Errors | **Error Code** | **Error Description** | **gRPC Status Code** | | ---------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------ | -------------------- | | `sfdc.platform.eventbus.grpc.topic.meta.permission` | An error occurred while getting the metadata for the specified topic. Ensure the credentials and topic name are correct. | `PERMISSION_DENIED` | | `sfdc.platform.eventbus.grpc.topic.not.found` | The topic information for the specified topic wasn’t found. Ensure that the topic name is correct and the topic exists. | `NOT_FOUND` | | `sfdc.platform.eventbus.grpc.topic.api.unavailable` | The topic information is unavailable for the specified topic. | `UNAVAILABLE` | | `sfdc.platform.eventbus.grpc.topic.validation.empty` | An error occurred while validating the topic information provided. The topic can’t be blank. | `INVALID_ARGUMENT` | ### Publish Errors | **Error Code** | **Error Description** | **gRPC Status Code** | | ------------------------------------------------------------ | ---------------------------------------------------------------------------------------------------------------------- | -------------------- | | `sfdc.platform.eventbus.grpc.publish.event.count.invalid` | The request contains no events. | `INVALID_ARGUMENT` | | `sfdc.platform.eventbus.grpc.publish.topic.mismatch` | There’s a mismatch of topic names between requests. | `INVALID_ARGUMENT` | | `sfdc.platform.eventbus.grpc.publish.auth.refresh.invalid` | The refresh token shouldn’t be provided in the initial request. | `INVALID_ARGUMENT` | | `sfdc.platform.eventbus.grpc.publish.topic.validation.empty` | An error occurred while validating the provided topic information. The topic can’t be blank. | `INVALID_ARGUMENT` | | `sfdc.platform.eventbus.grpc.publish.stream.sweeper.timeout` | No publish request was received during the timeout period of 1,800 seconds (30 minutes). The publish stream timed out. | `CANCELLED` | ### Subscription Errors These error codes apply to both subscriptions and managed subscriptions. These errors cause the long-lived connection for the `Subscribe` or `ManagedSubscribe` RPC method to stop. Retry the RPC method call to create a new long-lived connection. | **Error Code** | **Error Description** | **gRPC Status Code** | | --------------------------------------------------------------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -------------------- | | `sfdc.platform.eventbus.grpc.subscription.fetch.requested.events.invalid` | The requested number of events in a fetch request must be greater than zero. | `INVALID_ARGUMENT` | | `sfdc.platform.eventbus.grpc.subscription.fetch.topic.mismatch` | There’s a mismatch of topic names between requests. | `INVALID_ARGUMENT` | | `sfdc.platform.eventbus.grpc.subscription.fetch.replayid.validation.failed` | The Replay ID validation failed. The `replay_preset` field in FetchRequest is set to `ReplayPreset.CUSTOM`, but no Replay ID value is set in `replay_id`. Provide a Replay ID in the `replay_id` field to subscribe after the custom Replay ID. | `INVALID_ARGUMENT` | | `sfdc.platform.eventbus.grpc.subscription.fetch.replayid.corrupted` | The Replay ID validation failed. Ensure that the `replay_id` field value in FetchRequest is valid and refers to an event that is within the retention window. | `INVALID_ARGUMENT` | | `sfdc.platform.eventbus.grpc.subscription.fetch.replay.repeated` | Due to an internal error, the server received an event that’s older than the one received earlier. Its Replay ID value is lower than that of the last received event. | `INTERNAL` | | `sfdc.platform.eventbus.grpc.subscription.fetch.request.invalid` | The fetch request is missing an auth refresh. Include an auth refresh when sending a fetch request with zero events requested. The auth refresh is for Salesforce internal use only. | `INVALID_ARGUMENT` | | `sfdc.platform.eventbus.grpc.subscription.fetch.overflow` | Too many total events were requested for this subscription. Process some events first before sending a new fetch request. | `INVALID_ARGUMENT` | | `sfdc.platform.eventbus.grpc.subscription.internal.error` | The subscription encountered an internal error and can’t continue. Try restarting the subscription. | `INTERNAL` | | `sfdc.platform.eventbus.grpc.subscription.limit.exceeded` | The maximum number of delivered events to all clients in the last 24 hours has been exceeded. | `RESOURCE_EXHAUSTED` | | `sfdc.platform.eventbus.grpc.subscription.topic.cannot.subscribe` | Can’t subscribe to the specified topic. Check that you have the required permissions. | `PERMISSION_DENIED` | ### Managed Subscription Errors These error codes apply to managed subscriptions only. These errors cause the long-lived connection for the `ManagedSubscribe` RPC method to stop. Retry the RPC method call to create a new long-lived connection. | **Error Code** | **Error Description** | **gRPC Status Code** | | ------------------------------------------------------------------------------------------ | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -------------------- | | `sfdc.platform.eventbus.grpc.managed.subscription.already.active` | An active managed subscription with the same ID already exists. If a previous subscription with the same ID was stopped, there can be a delay before it expires. Try subscribing again later. | `ALREADY_EXISTS` | | `sfdc.platform.eventbus.grpc.managed.subscription.config.deleted` | The managed subscription configuration was deleted. | `NOT_FOUND` | | `sfdc.platform.eventbus.grpc.managed.subscription.config.stopped` | The managed subscription configuration state is STOP. Set state to RUN before subscribing. | `NOT_FOUND` | | `sfdc.platform.eventbus.grpc.managed.subscription.developer.name.not.found` | The managed subscription configuration isn't found. Ensure the developer name is correct and that the managed subscription configuration exists. If the configuration was recently created, try again. | `NOT_FOUND` | | `sfdc.platform.eventbus.grpc.managed.subscription.fetch.request.events.invalid` | The requested number of events in a fetch request must be greater or equal to zero. | `INVALID_ARGUMENT` | | `sfdc.platform.eventbus.grpc.managed.subscription.fetch.request.invalid` | The managed fetch request is missing a CommitReplayRequest or an auth refresh. Include a CommitReplayRequest or an auth refresh when sending a request with zero events requested. The auth refresh is for Salesforce internal use only. | `INVALID_ARGUMENT` | | `sfdc.platform.eventbus.grpc.managed.subscription.id.or.developer.name.empty` | An error occurred while validating the managed subscription ID or developer name. Provide only one of those values. | `INVALID_ARGUMENT` | | `sfdc.platform.eventbus.grpc.managed.subscription.fetch.request.commit.request.restricted` | The initial managed fetch request can’t include a CommitReplayRequest. | `INVALID_ARGUMENT` | | `sfdc.platform.eventbus.grpc.managed.subscription.fetch.request.auth.request.restricted` | The initial managed fetch request can’t include an auth refresh. The auth refresh is for Salesforce internal use only. | `INVALID_ARGUMENT` | | `sfdc.platform.eventbus.grpc.managed.subscription.subscription.id.invalid` | The managed subscription ID is invalid. The subscription ID must be 15 or 18 characters long. | `INVALID_ARGUMENT` | | `sfdc.platform.eventbus.grpc.managed.subscription.subscription.id.not.found` | The managed subscription configuration isn't found. Ensure the subscription ID is correct and that the managed subscription configuration exists. If the configuration was recently created, try again. | `NOT_FOUND` | ### Errors in CommitResponse These errors don't cause the long-lived connection for the `ManagedSubscribe` RPC method to stop. These errors are part of the `CommitResponse` that is returned after sending a `CommitReplayRequest` to commit a Replay ID. | **Error Code** | **Error Description** | | -------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | | `COMMIT` | No Replay ID was provided to commit. Specify a Replay ID in CommitReplayRequest. | | `COMMIT` | The Replay ID provided in the commit request is invalid. Ensure the Replay ID in CommitReplayRequest is valid. If the Replay ID is old, it can be outside the event retention window and is no longer valid. | | `COMMIT` | The commit request was received before the subscription delivered any events. Wait for event delivery before attempting to commit a Replay ID. | | `COMMIT` | The Replay ID commit request failed. For information about how to handle the Replay ID commit failure, check the documentation. | | `COMMIT` | The Replay ID in this commit request is older than the currently committed Replay ID. | **See Also** - [Provide the RPC ID When Contacting Support to Troubleshoot an Error](https://developer.salesforce.com/docs/platform/pub-sub-api/guide/rpc-id-field.md) - [Generate a Client Trace ID to Troubleshoot Non-Pub/Sub API Errors](https://developer.salesforce.com/docs/platform/pub-sub-api/guide/client-trace-id.md) # Include Authorization Headers in RPC Method Calls Every time an RPC method executes, it uses the headers to authorize access to resources. How you set the headers depends on your programming language. In Python, the headers are provided in a tuple as a parameter to each call. For more information, see [Python Quick Start for Pub/Sub API](https://developer.salesforce.com/docs/platform/pub-sub-api/guide/qs-python-quick-start.md). In other languages, such as Java, you can provide the headers in the stub separately before making a gRPC call. ## `accesstoken` The OAuth access token or the session ID returned in the `login()` call response. For more information about getting an OAuth access token, see [OAuth Authorization Flows](https://help.salesforce.com/s/articleView?id=xcloud.remoteaccess_oauth_flows.htm&type=5) in _Salesforce Help_. For more information about the `login()` call, see [login()](https://developer.salesforce.com/docs/atlas.en-us.api.meta/api/sforce_api_calls_login.htm) in the _SOAP API Developer Guide_. ## `instanceurl` The first part of your Salesforce server URL without the ending path portion. You can get this URL in one of two ways. - From the `login()` call result in `serverUrl`, as outlined in the [quick start example](https://developer.salesforce.com/docs/platform/pub-sub-api/guide/qs-python-quick-start.md) in this guide. For more information, see [LoginResult](https://developer.salesforce.com/docs/atlas.en-us.api.meta/api/sforce_api_calls_login_loginresult.htm) in the _SOAP API Developer Guide_. - From the result that Salesforce returns as part of the OAuth authorization flow in `instance_url`. For more information, see [OAuth 2.0 User-Agent Flow for Desktop or Mobile App Integration](https://help.salesforce.com/s/articleView?id=remoteaccess_oauth_user_agent_flow.htm&type=5&language=en_US) in _Salesforce Help_. The Salesforce URL can be: - A MyDomain URL. For example, `https://MyDomainName.my.salesforce.com`. - A custom domain URL. For example, `https://www.example.com`. ## `tenantid` An ID that uniquely identifies your org. If you set this header to the org ID, the Pub/Sub API constructs the entire value for you. These values are valid. - `{Org ID}` - `core/{Subdomain}/{Org ID}` Examples: - Org ID: `00D5e000003TIrB` - Entire value: `core/MyDomainName/00D5e000003TIrB` The org ID is returned in the [login()](https://developer.salesforce.com/docs/atlas.en-us.api.meta/api/sforce_api_calls_login.htm) call response. Alternatively, you can get your org ID value by following the steps in [Find your Salesforce Organization ID](https://help.salesforce.com/articleView?id=000325251&mode=1&type=1) in _Salesforce Help_. If you build the entire value, use your org’s My Domain login URL. You can find this value in **Setup > My Domain**. For more information, see [What Is My Domain?](https://help.salesforce.com/s/articleView?id=xcloud.faq_domain_name_what.htm&type=5) in _Salesforce Help_. # Java Quick Start To get started with Pub/Sub API and publish and subscribe to events, follow the steps in [Java Quick Start for Publishing and Subscribing to Events](https://developer.salesforce.com/docs/platform/pub-sub-api/guide/qs-java-quick-start.md). To try the managed subscriptions beta feature, which lets you track and resume your subscriptions easily, follow the steps in [Java Quick Start for Managed Event Subscriptions (Beta)](https://developer.salesforce.com/docs/platform/pub-sub-api/guide/qs-java-managed-quick-start.md). **In this section:** - [Java Quick Start for Publishing and Subscribing to Events](https://developer.salesforce.com/docs/platform/pub-sub-api/guide/qs-java-quick-start.md) - [Java Quick Start for Managed Event Subscriptions (Beta)](https://developer.salesforce.com/docs/platform/pub-sub-api/guide/qs-java-managed-quick-start.md) # Java Quick Start for Managed Event Subscriptions (Beta) In this quick start, you configure a managed subscription, and then use it to subscribe to `Order_Event__e` events by passing in the managed subscription name. You verify that the managed subscription commits the Replay ID of the last processed event and resumes after the last event when the client disconnects and reconnects. To learn more about the beta feature, see [Managed Event Subscriptions (Beta)](https://developer.salesforce.com/docs/platform/pub-sub-api/guide/managed-sub.md). **note:** The `ManagedSubscribe` RPC method is part of a Beta Service. Customer may opt to try such Beta Service in its sole discretion. Any use of the Beta Service is subject to the applicable Beta Services Terms provided at [Agreements and Terms](https://www.salesforce.com/company/legal/agreements/). **note:** The quick start provides enough instructions and code snippets to build your own client. The examples are for learning purposes only. They aren’t intended for production use and haven’t undergone thorough functional and performance testing. You can use them as a starting point. **In this section:** - [Step 1: Create ManagedEventSubscription with Tooling API](https://developer.salesforce.com/docs/platform/pub-sub-api/guide/qs-java-managed-sub-create.md) - [Step 2: Start the Managed Subscription](https://developer.salesforce.com/docs/platform/pub-sub-api/guide/qs-java-managed-sub-start.md) - [Step 3: Publish Events and Verify the Managed Subscription](https://developer.salesforce.com/docs/platform/pub-sub-api/guide/qs-java-managed-sub-publish-verify.md) # Java Quick Start for Publishing and Subscribing to Events In this quick start, you learn how to build a Pub/Sub API client in Java. The steps walk you through building the package, configuring the client, and running the client to publish and subscribe to events. **note:** The quick start provides enough instructions and code snippets to build your own client. The examples are for learning purposes only. They aren’t intended for production use and haven’t undergone thorough functional and performance testing. Use them as a starting point. **In this section:** - [Step 1: Define a Platform Event](https://developer.salesforce.com/docs/platform/pub-sub-api/guide/qs-java-prereqs.md) - [Step 2: Clone and Build the Client](https://developer.salesforce.com/docs/platform/pub-sub-api/guide/qs-java-build-client.md) - [Step 3: Configure Client Parameters](https://developer.salesforce.com/docs/platform/pub-sub-api/guide/qs-java-parameters.md) - [Step 4: Publish and Subscribe to Platform Events](https://developer.salesforce.com/docs/platform/pub-sub-api/guide/qs-java-publish-subscribe.md) - [Step 5: Subscribe to Change Events](https://developer.salesforce.com/docs/platform/pub-sub-api/guide/qs-java-change-events.md) # Managed Event Subscriptions (Beta) Use managed event subscriptions to track the events that a subscriber client consumed and to resume a subscription from where it left off after a client disconnects. When you resubscribe with the new `ManagedSubscribe` RPC method, the subscription resumes after the last Replay ID that the client successfully committed on the server. **note:** This feature is a Beta Service. Customer may opt to try such Beta Service in its sole discretion. Any use of the Beta Service is subject to the applicable Beta Services Terms provided at [Agreements and Terms](https://www.salesforce.com/company/legal/agreements/). With managed subscriptions, you don’t worry about storing the Replay ID and managing your subscriptions on the client. After you commit a Replay ID, the system stores it on the server. After the subscription disconnects and is restarted, it resumes seamlessly from the events stored in the event bus. For more information about event storage in the event bus, see [Event Message Durability](https://developer.salesforce.com/docs/platform/pub-sub-api/guide/event-message-durability.md). You can subscribe to the supported [event types](https://developer.salesforce.com/docs/platform/pub-sub-api/guide/supported-event-types.html), or to custom or standard channels for platform events or change data capture events. A managed subscription is unique per client and can’t be shared with other clients for the same Salesforce org. When using a managed subscription, you configure the subscription behavior and state by using the ManagedEventSubscription in Metadata API and Tooling API. For more information, see [Configuring the Managed Event Subscription](#configuring-a-managed-event-subscription). For a step-by-step walkthrough for configuring a managed subscription and using it to subscribe to events, see [Java Quick Start for Managed Event Subscriptions (Beta)](https://developer.salesforce.com/docs/platform/pub-sub-api/guide/qs-java-managed-quick-start.md). Managed subscriptions have the same allocations as subscriptions that use the `Subscribe` RPC method. In addition, there’s a limit on how many managed subscriptions you can create per Salesforce org. See [Pub/Sub API and Event Allocations](https://developer.salesforce.com/docs/platform/pub-sub-api/guide/allocations.md). ## Running a Managed Subscription To start a managed subscription, call the [ManagedSubscribe RPC Method](/docs/platform/pub-sub-api/references/methods/managedsubscribe-rpc.html) to subscribe by using the developer name of the ManagedEventSubscription Metadata API or Tooling API record, or use the ID of the Tooling API record. ManagedEventSubscription contains configuration options for the managed subscription that controls subscription behavior after an error occurs and the subscription state. For more information, see [Configuring the Managed Event Subscription](#configuring-a-managed-event-subscription). If the subscriber client disconnects, the client can resume event delivery after the committed Replay ID, avoiding delivery of duplicate events or missed events. Events are returned from the event bus as long as they’re within the retention window. To learn about committing Replay IDs and the keepalive behavior, see [ManagedSubscribe RPC Method (Beta)](/docs/platform/pub-sub-api/references/methods/managedsubscribe-rpc.html). This diagram shows the sequence of managed fetch requests and responses that a client performs to subscribe to events, commit the Replay ID of the last processed event, and resubscribe. ![A sequence of managed fetch requests and responses](https://developer.salesforce.com/docs/media/pubsub-managed-sub-flow-2.png '{"class": "image-lg"}') Here’s a detailed description of the diagram. 1. The client starts the subscription by calling the `ManagedSubscribe` RPC method. It specifies the number of events to request in ManagedFetchRequest. The first time the client sends a managed fetch request, it sets the `subscription_id` field in ManagedFetchRequest. The `subscription_id` field is the ID of the Tooling API ManagedEventSubscription record. 2. The server returns a batch of events in ManagedFetchResponse. 3. After processing the events, the client commits the Replay ID of the last processed event by setting the `CommitReplayRequest` field of ManagedFetchRequest. Also, the client requests new events by setting the `num_requested` field of ManagedFetchRequest. 4. The server returns a CommitReplayResponse nested in a ManagedFetchResponse that contains the result of the Replay ID commit request, indicating success or failure. 5. The client receives the response of the Replay ID commit. After a successful commit, the server holds the Replay ID that the client committed (`event100.replay_id`). 6. The client doesn’t process any events and disconnects. 7. The client resubscribes by making a new `ManagedSubscribe` RPC method call. It sends the same subscription ID and requests new events in ManagedFetchRequest as part of the `ManagedSubscribe` RPC method call. 8. The server returns a batch of events in ManagedFetchResponse. The batch of events starts from the event that’s after the one last processed (`event100`), in addition to new events received since the client shutdown. Events are returned from the event bus as long as they’re within the retention window. **note:** The previous diagram shows one use case. It doesn’t show an example of when an empty ManagedFetchResponse is returned with the latest Replay ID. For best practices on when to commit a Replay ID, see [Committing a Replay ID](/docs/platform/pub-sub-api/references/methods/managedsubscribe-rpc.html#committing-a-replay-id). ## Configuring a Managed Event Subscription Managed subscriptions use configuration options that you set in ManagedEventSubscription by using Tooling API or Metadata API. You can define the topic to subscribe to, the position in the stream where the subscription starts, and where it resumes after an error. **important:** When you create, update, or delete a ManagedEventSubscription record, there can be a delay before the new or changed configuration takes effect in Pub/Sub API. Wait a few minutes before you try to subscribe. Even if you can see the changes via Tooling API or Metadata API, the changes sometimes aren't immediately available to the Pub/Sub API service. To reduce the risk of inconsistencies after you delete a ManagedEventSubscription record, don't use the same developer name to create a new ManagedEventSubscription record. This example shows a managed event subscription configuration for Order Event. This JSON example is the body of a REST API POST request that creates a managed event subscription named `My_Managed_Subscription`. The corresponding REST endpoint is `/services/data/v61.0/tooling/sobjects/ManagedEventSubscription`. ``` { "FullName": "My_Managed_Subscription", "Metadata": { "label" : "My Managed Subscription", "topicName" : "/event/Order_Event__e", "defaultReplay": "LATEST", "state" : "RUN", "errorRecoveryReplay" : "LATEST" } } ``` Here’s a brief description of the ManagedEventSubscription fields. For more information about each field, see [ManagedEventSubscription (Beta)](https://developer.salesforce.com/docs/atlas.en-us.api_meta.meta/api_meta/meta_managedeventsubscription.htm) in the _Metadata API Developer Guide_ and [ManagedEventSubscription (Beta)](https://developer.salesforce.com/docs/atlas.en-us.api_tooling.meta/api_tooling/tooling_api_objects_managedeventsubscription.htm) in the _Tooling API Developer Guide_. - The `topicName` specifies the event subscription path. In this example, the topic name references `Order_Event__e`. - The `defaultReplay` field specifies from where the subscription starts for the first time or after it was stopped, or when no replay ID is stored. In this example, it starts from the latest events received. - The `state` field specifies whether the subscription is running or stopped. In this example, it’s set to be running. The administrator can stop the subscription by changing the changing state field to `STOP`. For more information, see [Stopping a Managed Event Subscription](#stopping-a-managed-event-subscription-via-the-api). - The `errorRecoveryReplay` field specifies the position in the stream where the subscription restarts if the committed Replay ID is invalid. The Replay ID can be invalid if it corresponds to an event outside the event retention window. In this example, the subscription restarts from the latest event received. ## Stopping a Managed Event Subscription via the API An administrator can stop a managed subscription through ManagedEventSubscription in Metadata API or Tooling API. When the administrator stops the subscription via the API, the stored Replay ID value that was committed previously is deleted. Unlike a subscription that ends due to an error or by shutting down the client, the subscription that is stopped via the API doesn’t resume where it left off the next time it’s restarted. It starts from the position specified in the `defaultReplay` field, which is either the latest or earliest position in the stream. When the ManagedEventSubscription `state` field is changed from `RUN` to `STOP`, the system notifies the Pub/Sub API client of the new state value after a small delay and the subscription disconnects. To stop a subscription with ManagedEventSubscription by using Tooling API, send a PATCH request to this REST endpoint after replacing the `` placeholder with the ID of ManagedEventSubscription record: `/services/data/v61.0/tooling/sobjects/ManagedEventSubscription/`. Include the `FullName` field and the `state` field in the Metadata object. ``` { "FullName": "My_Managed_Subscription", "Metadata": { "state" : "STOP" } } ``` To get the ID of a ManagedEventSubscription record, perform this query in Tooling API after replacing the developer name placeholder with the developer name. ``` SELECT Id FROM ManagedEventSubscription WHERE DeveloperName='' ``` To start a managed subscription, wait at least 5 minutes after changing the `state` field to `STOP`, update the `state` field to `RUN` and, after waiting for a short while to allow for the state change to take effect in the service, call the `ManagedSubscribe` RPC method. The subscription starts from the position specified in the `defaultReplay` field. If you call `ManagedSubscribe` RPC when the `state` field is set to `STOP`, the subscription doesn’t start. **See Also** - [ManagedSubscribe RPC Method (Beta)](/docs/platform/pub-sub-api/references/methods/managedsubscribe-rpc.html) - _Metadata API Developer Guide_: [ManagedEventSubscription (Beta)](https://developer.salesforce.com/docs/atlas.en-us.api_meta.meta/api_meta/meta_managedeventsubscription.htm) - _Tooling API Developer Guide_: [ManagedEventSubscription (Beta)](https://developer.salesforce.com/docs/atlas.en-us.api_tooling.meta/api_tooling/tooling_api_objects_managedeventsubscription.htm) # ManagedSubscribe RPC Method (Beta) The `ManagedSubscribe` method starts a managed subscription. With a managed subscription, you can store the Replay ID of the event last processed on the server and resume your subscription from where it left off by using the stored Replay ID. The `ManagedSubscribe` method uses bidirectional streaming and is pull-based, which means that it requests new events. ``` rpc ManagedSubscribe (stream ManagedFetchRequest) returns (stream ManagedFetchResponse) ``` **note:** This feature is a Beta Service. Customer may opt to try such Beta Service in its sole discretion. Any use of the Beta Service is subject to the applicable Beta Services Terms provided at [Agreements and Terms](https://www.salesforce.com/company/legal/agreements/). A client calls `ManagedSubscribe` to subscribe to events by using the ID of the Tooling API record. ManagedEventSubscription contains configuration options that control subscription behavior after an error occurs and the subscription state. For more information, see [Configuring the Managed Event Subscription](/docs/platform/pub-sub-api/guide/managed-sub.html#configuring-a-managed-event-subscription). A managed subscription is unique per client and can’t be shared with other clients for the same Salesforce org. For more information about the fields in ManagedFetchRequest and ManagedFetchResponse, see the [Pub/Sub API proto file](https://github.com/forcedotcom/pub-sub-api/blob/main/pubsub_api.proto). ## Requesting Events A client can request the number of events that it can process by setting the `num_requested` field in ManagedFetchRequest. The maximum value for `num_requested` is 100. The server returns the events in one or more ManagedFetchResponses. Each ManagedFetchResponse contains the number of events that are pending delivery in the `pending_num_requested` field. If a client requests more events before the server finishes the last requested amount, the server appends the new amount to the current batch of events that it still needs to fetch and deliver. ## Committing a Replay ID The client commits the Replay ID from each received ManagedFetchResponse whether the response contains events or is empty. - If a ManagedFetchResponse contains one or more events, the client processes the events first and then commits the Replay ID of the last processed event in CommitReplayRequest as part of ManagedFetchRequest. - If a ManagedFetchResponse contains an empty batch of events, the client commits the latest Replay ID that it gets from the `latest_replay_id` field in ManagedFetchResponse. The client sets the Replay ID in CommitReplayRequest as part of ManagedFetchRequest. That Replay ID points to an advanced position in the stream beyond the last consumed event. An advanced position in the stream is possible even if you didn’t receive events because the event bus combines streams from multiple orgs. No org-specific information or personally identifiable information is shared with other orgs. The commit request response is returned in a ManagedFetchResponse in CommitReplayResponse, which indicates success or failure. This particular ManagedFetchResponse doesn’t include any events. There is no one-to-one correspondence between a CommitReplayRequest and a CommitReplayResponse. The server can return a CommitReplayResponse for one or more CommitReplayRequests. If the server receives several commit requests in a short amount of time, the server commits the latest Replay ID from all the requests and returns one CommitReplayResponse. If the Replay ID commit request is successful, as indicated by the CommitReplayResponse `success` field, the Replay ID is stored on the server. It’s used the next time the client resubscribes by using the same subscription ID to resume the subscription from where it left off. If the Replay ID commit request isn’t successful, make sure that you sent a valid Replay ID. If your commit request is valid, you can ignore the commit failure. You can rely on future commit requests that the client sends after processing the next batch of events to succeed in case the previous error was transient. The next time the client sends a CommitReplayRequest as part of a new ManagedFetchRequest, if it succeeds, the new Replay ID is stored on the server. ## ManagedSubscribe Keepalive Behavior The ManagedSubscribe call stream stays open under certain conditions, including the number of pending events to be delivered, and the Replay ID commit requests the client sends. The number of pending events is based on the number of events the client requested in ManagedFetchRequests and the number of events delivered in ManagedFetchResponses. This table summarizes how to keep a managed subscription stream open for each condition. | Condition | To keep a managed subscription stream alive | | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | | Pending events to be sent to the client and there are still events in the event bus to be delivered (`pending_num_requested` in ManagedFetchResponse is greater than zero) |
  • The stream stays open.
  • After the first ManagedFetchResponse is received, the client must commit a Replay ID within 30 minutes from when the last ManagedFetchResponse with a CommitReplayResponse was sent. To commit a Replay ID, a client sends a ManagedFetchRequest with the CommitReplayRequest set to the Replay ID value.
| | No more pending events to be sent to the client (`pending_num_requested` in ManagedFetchResponse is equal to zero) |
  • The client must request new events within 60 seconds from when the last ManagedFetchResponse was sent. To request new events, a client sends a ManagedFetchRequest with a `num_requested` field greater than zero.
  • After the first ManagedFetchResponse is received, the client must commit a Replay ID within 30 minutes from when the last ManagedFetchResponse with a CommitReplayResponse was sent. To commit a Replay ID, a client sends a ManagedFetchRequest with the CommitReplayRequest set to the Replay ID value.
| | Pending events to be sent to the client but no new events are available in the event bus to be delivered (`pending_num_requested` in ManagedFetchResponse is greater than zero) | The server keeps the stream open by sending keepalive messages within 270 seconds. A keepalive message indicates that the Pub/Sub API service is healthy and provides a new Replay ID that points to an advanced position in the stream. | ## Keepalive Messages Sent from the Server If there are pending events to be sent to the client (`pending_num_requested` in ManagedFetchResponse is greater than zero) but no new events are available in the event bus, the server keeps the stream alive by sending keepalive messages within 270 seconds. The keepalive message is an empty batch ManagedFetchResponse with the latest `replay_id`. The returned `replay_id` can be the last event consumed in your client or an advanced position in the stream beyond the last consumed event. An advanced position in the stream is possible even if you didn’t receive events because the event bus combines streams from multiple orgs. No org-specific information or personally identifiable information is shared with other orgs. For best performance results, we recommend that you commit the `replay_id` from each empty ManagedFetchResponse by sending a CommitReplayRequest with that `replay_id` as part of a ManagedFetchRequest. That way, when you restart the subscription, it starts from a `replay_id` that’s recent. An old `replay_id` refers to an earlier position in the stream than the `replay_id` that the keepalive message returns. Using the more recent `replay_id` ensures that the server doesn’t take too much time to seek the event in the event bus when the subscription restarts. **See Also** - [Managed Event Subscriptions (Beta)](/docs/platform/pub-sub-api/guide/managed-sub.html) - [Java Quick Start for Managed Event Subscriptions (Beta)](/docs/platform/pub-sub-api/guide/qs-java-managed-quick-start.html) - [Subscribe RPC Method](https://developer.salesforce.com/docs/platform/pub-sub-api/references/methods/subscribe-rpc.md) # Provide the RPC ID When Contacting Support to Troubleshoot an Error The response of each RPC call contains the `rpc-id` field, which identifies the method execution in Salesforce. Salesforce Customer Support uses the `rpc-id` value to aid in troubleshooting issues with an RPC call. If a call fails, the returned exception includes the `rpc-id` field. For more information, see [Handle Errors](https://developer.salesforce.com/docs/platform/pub-sub-api/guide/handling-errors.md). Even if a call is successful and returns a successful response, there can be issues, for example, slow performance in event publishing or delivery. If you determine that the issue doesn’t originate from your code and you contact Salesforce Customer Support for help, provide the `rpc-id` value from the response. **See Also** - [Handle Errors](https://developer.salesforce.com/docs/platform/pub-sub-api/guide/handling-errors.md) - [Generate a Client Trace ID to Troubleshoot Non-Pub/Sub API Errors](https://developer.salesforce.com/docs/platform/pub-sub-api/guide/client-trace-id.md) # Pub/Sub API and Event Allocations Check out allocations for Pub/Sub API, including the event message size, how many events you can publish and request, and how many events can be delivered to subscribers. ## Publish and PublishStream RPC Method Allocations Check out allocations related to event publishing. ### Event Message Size in a PublishRequest We recommend that the total size of a batch of events in a PublishRequest doesn’t exceed 3 MB. The 3-MB recommendation is below the gRPC 4-MB limit and ensures optimal performance. If your PublishRequest exceeds 4 MB, the publish call fails with this gRPC error, and the server closes the stream. ``` Status{code=UNAVAILABLE, description=upstream connect error or disconnect/reset before headers. reset reason: remote reset, cause=null} ``` Each event message in a batch in a PublishRequest has a maximum size of 1 MB. If you publish a batch of events containing an event larger than 1 MB, the PublishResult of that event contains an error, and the event can’t be published. Each event in the batch has its own PublishResult, so it can still be published if there are no errors. ### Number of Events in a Publish Request We recommend that you send no more than 200 events in one publish request for best performance results. ### Platform Event Publishing Allocation Platform events are subject to an hourly publishing allocation. For more information, see [Platform Event Allocations](https://developer.salesforce.com/docs/atlas.en-us.platform_events.meta/platform_events/platform_event_limits.htm) in the _Platform Events Developer Guide_. ## Subscribe and ManagedSubscribe (Beta) RPC Allocations Check out allocations related to event subscription. ### Number of Events Requested in a Subscribe and ManagedSubscribe (Beta) Call The maximum number of events that can be requested across all FetchRequests or ManagedFetchRequests in a Subscribe or ManagedSubscribe call, respectively, is 100. If a client requests more than 100 events, the server considers the number of requested events to be 100. The client sets the number of events to request in the `num_requested` field of the FetchRequest or ManagedFetchRequest. The server adjusts the total number of requested events based on the requested events sent in one or more requests and how many events were already delivered. ### Number of Managed Subscriptions (Beta) The maximum number of unique managed subscriptions that can be created per Salesforce org is 200. ## Event Delivery Allocations for Platform Events and Change Data Capture When you subscribe with the Pub/Sub API, the event delivery allocations for platform events, change data capture events, and real-time event monitoring events apply. For more information, see the documentation for each event type. - _Platform Events Developer Guide_: [Platform Event Allocations](https://developer.salesforce.com/docs/atlas.en-us.platform_events.meta/platform_events/platform_event_limits.htm) - _Change Data Capture Developer Guide_: [Change Data Capture Allocations](https://developer.salesforce.com/docs/atlas.en-us.change_data_capture.meta/change_data_capture/cdc_allocations.htm) - _Salesforce Help_: [Real-Time Event Monitoring Data Streaming](https://help.salesforce.com/s/articleView?id=xcloud.real_time_event_monitoring_streaming.htm&type=5) ## Concurrent RPC Stream Limit for an HTTP/2 Connection A gRPC channel can have a maximum of 1,000 concurrent streams on an underlying HTTP/2 connection. Concurrent streams are RPC calls that are made on the same gRPC channel and that are active at the same time. For example, if a client uses one gRPC channel and makes 1,000 RPC requests on that channel concurrently, such as `GetTopic` and `Subscribe` requests and others, the client hits the concurrent stream limit. After the limit is reached, additional RPCs are queued in the client and must wait for active RPCs to finish before they’re sent. For more information and for a workaround, see [Performance Best Practices](https://grpc.io/docs/guides/performance/) in the gRPC documentation. **note:** Using a large number of concurrent streams on the same channel can slow down overall request throughput. For example, if a client has a high event publishing volume going through the same gRPC channel that is shared with many other RPC requests, the contention on the same gRPC channel can cause lower throughput. We recommend that you understand the tradeoffs of having many concurrent requests and their effect on performance. You can perform your own performance testing based on your individual workloads. # Pub/Sub API and the Expanded Event Bus Using Pub/Sub API, you can interface with the expanded and improved Salesforce event bus by publishing and subscribing to events. The event bus is a multitenant, multicloud event storage and delivery service based on a publish-subscribe model. The event bus is based on a time-ordered event log, which ensures that event messages are stored and delivered in the order that they’re received by Salesforce. Platform events and change data capture events are published to the event bus, where they’re stored for 72 hours. You can retrieve stored event messages from the event bus with Pub/Sub API. Each event message contains the Replay ID field, represented as `replay_id` in the protocol specification. It’s an opaque ID that identifies the event in the stream and enables replaying the stream after a specific replay ID. The expanded event bus service is built outside the main Salesforce CRM stack, which powers Sales Cloud and Service Cloud, the original products of Salesforce. As a result, the API provides cross-cloud integration capabilities between Sales Cloud and Service Cloud and other Salesforce clouds, such as Marketing Cloud, Commerce Cloud, and Tableau Analytics. It also provides enhanced scalability and performance. Developers can focus on building event-driven apps that scale and integrate across various Salesforce clouds. ![Salesforce event bus](https://developer.salesforce.com/docs/media/salesforce-event-bus.png "Salesforce event bus") **See Also** - _Salesforce Architects Blog_: [The New Salesforce Event Bus](https://medium.com/salesforce-architects/the-new-salesforce-event-bus-f82165cb0585) - _Salesforce Engineering Blog_: [How Apache Kafka Inspired Our Platform Events Architecture](https://engineering.salesforce.com/how-apache-kafka-inspired-our-platform-events-architecture-2f351fe4cf63) # Pub/Sub API as a gRPC API Because the Pub/Sub API is a gRPC API, let’s define what a gRPC API is. gRPC is an open source Remote Procedure Call (RPC) framework that enables connecting devices, mobile applications, and browsers to backend services. With gRPC, a client app can call a method on a server as if it were a local object, making it easier for you to create distributed apps and services. gRPC requires defining a service, which specifies the methods that can be called remotely with their parameters and return types. The server implements this interface and runs a gRPC server to handle client calls. The client has a stub that mirrors the methods available on the server. ![Pub/Sub API as a gRPC API](https://developer.salesforce.com/docs/media/pubsub-grpc-stub_.png "Pub/Sub API as a gRPC API") ## Pub/Sub API Endpoint Pub/Sub API uses a secure gRPC channel to connect to the endpoint over SSL. For available endpoints, see [Pub/Sub API Endpoints](https://developer.salesforce.com/docs/platform/pub-sub-api/guide/pub-sub-endpoints.md). ## Protocol Buffer Messages and Binary Apache Avro Encoding for Event Payloads Pub/Sub API uses protocol buffers as the Interface Definition Language (IDL) and as the underlying message interchange format. The Pub/Sub API service is defined in a proto file, with RPC method parameters, such as FetchRequest, and return types, such as FetchResponse, specified as protocol buffer messages. Within those protocol buffer messages, the payload field of published and delivered events is encoded in binary Apache Avro. For example, ProducerEvent is the event that the Publish and PublishStream RPC methods publish as part of PublishRequest. The `payload` field of ProducerEvent is encoded in binary Apache Avro, but the other fields aren't. Similarly, ConsumerEvent is the type of event that is delivered to a Subscribe RPC method as part of FetchResponse. The `payload` field of ConsumerEvent is encoded in binary Apache Avro, but the other fields aren't. For more information about event serialization, see [Event Data Serialization with Apache Avro](https://developer.salesforce.com/docs/platform/pub-sub-api/guide/event-avro-serialization.md). ![ProducerEvent and ConsumerEvent proto file definitions](https://developer.salesforce.com/docs/media/pubsub-proto-events.png "ProducerEvent and ConsumerEvent proto file definitions") This proto file example is based on the Pub/Sub API proto file, but we shortened it in the illustration. The proto file defines the service by listing the methods to publish, subscribe, get the schema, and get the topic information. For the full definition of the Pub/Sub API proto file, see [Pub/Sub API proto file](https://github.com/forcedotcom/pub-sub-api/blob/main/pubsub_api.proto) in GitHub. ![gRPC proto file example](https://developer.salesforce.com/docs/media/pubsub-proto-example.png "gRPC proto file example") This proto file excerpt lists the messages for the Subscribe method. FetchRequest is the request message of the Subscribe method. FetchResponse is the response message of the Subscribe method. For brevity, we omitted other messages in this example. # Pub/Sub API Endpoints Pub/Sub API provides a global endpoint and a region-specific endpoint. The endpoints include the port. Replace the `{port}` placeholder with the port that Pub/Sub API uses: `443` or `7443`. For example, the global endpoint with the `443` port is `api.pubsub.salesforce.com:443`. ## Global Endpoint Use this endpoint if you don't have data in transit privacy requirements for your org. To help reduce API request latency, the Pub/Sub API service attempts to process requests in the same region where they originate. ``` api.pubsub.salesforce.com:{port} ``` ## Europe Endpoint Use this endpoint if you must comply with data in transit privacy requirements for your org. Using this endpoint ensures that Pub/Sub API event processing occurs via the Salesforce service located in Germany. ``` api.deu.pubsub.salesforce.com:{port} ``` Events in the Salesforce event bus aren’t persisted or stored outside the region your org is located in. For more information about the regions where event processing occurs, see [AWS Regions](https://developer.salesforce.com/docs/platform/pub-sub-api/guide/considerations.md#aws-regions). **note:** If your company uses a firewall for network security, make sure that the Pub/Sub API endpoint is allowed in the firewall configuration. Allow the entire domain and port. For example, for the global endpoint, the domain is `api.pubsub.salesforce.com`, and the port is `443` or `7443`. Don’t use hardcoded IP addresses in your allowlist because they can change. Also, allow the HTTPS and TCP protocols. # Pub/Sub API Features Learn about some of the features of Pub/Sub API. | Feature | Pub/Sub API | | ------------------------- | ---------------------------------------------------------------------------------------------------------------- | | Event Encoding | Binary ([Apache Avro](https://avro.apache.org/)) | | Protocol | [gRPC](https://grpc.io/docs/what-is-grpc/) and protocol buffers | | Subscription model | Pull-based—the client requests the number of events to receive as they become available in the event bus. | | Subscription flow control | Client specifies how many events to receive for a Subscribe call based on event processing speed in the client. | | Publish status | Final publishing result returned indicates that the event was successfully published or that the publish failed. | **See Also** - [Pull Subscription and Flow Control](https://developer.salesforce.com/docs/platform/pub-sub-api/guide/flow-control.md) # Publish RPC Method The `Publish` method is a unary RPC, which means that it sends only one request and receives only one response. It synchronously publishes the batch of events in PublishRequest to an event bus topic. After publishing the event messages, the server sends a response to the client. Use the `Publish` method if you want to know the status of a publish operation before publishing the next batch of event messages. ``` rpc Publish (PublishRequest) returns (PublishResponse); ``` The PublishResponse holds a PublishResult for each event published that indicates success or failure of the publish operation. A successful status means that the event was published and is in the event bus. A failed status means that the event failed to publish. ## Publishing Required Fields The `Publish` method doesn’t enforce the presence of required custom event fields in the event payload. If you publish an event without including a required custom field, the publish call succeeds and the received event contains the field with a null value. For system fields, the `CreatedDate` and `CreatedById` fields are required in the event payload. If you publish an event without these system fields, the call fails. ## Order of Events If you publish a batch of events, the order of results in PublishResponse corresponds to the order of the events in the publish request. You can correlate each successful or failed result with the corresponding event in the publish request. ## Retry Publishing Failed Events You can retry publishing one or more failed events. You can match a publish result with the corresponding event in the publish request. If you don’t want to rely on the order of events published and received, you can use the event `id` field and `correlationKey` field of PublishResult. The process is similar to the one described for the `PublishStream` RPC method. For more information, see [PublishStream RPC Method](https://developer.salesforce.com/docs/platform/pub-sub-api/references/methods/publishstream-rpc.html). For more information about the fields in PublishRequest and PublishResponse, see the [Pub/Sub API proto file](https://github.com/forcedotcom/pub-sub-api/blob/main/pubsub_api.proto). # PublishStream RPC Method The `PublishStream` method uses bidirectional streaming. It can send a stream of publish requests while receiving a stream of publish responses from the server. Use PublishStream to achieve a higher publish rate than with Publish. ``` rpc PublishStream (stream PublishRequest) returns (stream PublishResponse); ``` The first PublishRequest of the stream identifies the topic to publish on. If a subsequent PublishRequest provides topic_name, it must match what was provided in the first PublishRequest. Otherwise, the RPC sends an error with an INVALID_ARGUMENT status. The server returns a PublishResponse for each PublishRequest when publishing is complete for the batch. It isn’t necessary for a client to wait for a PublishResponse before sending a new PublishRequest. Multiple publish batches can be queued up. This behavior allows for a higher publish rate because a client can asynchronously publish more events while publish operations are still in flight on the server side. The PublishResponse holds a PublishResult for each event published that indicates success or failure of the publish call. A successful status means that the event was published and is in the event bus. A failed status means that the event failed to publish. For more information about the fields in PublishRequest and PublishResponse, see the [Pub/Sub API proto file](https://github.com/forcedotcom/pub-sub-api/blob/main/pubsub_api.proto). ## Publishing Required Fields The `PublishStream` method doesn’t enforce the presence of required custom event fields in the event payload. If you publish an event without including a required custom field, the publish call succeeds and the received event contains the field with a null value. For system fields, the `CreatedDate` and `CreatedById` fields are required in the event payload. If you publish an event without these system fields, the call fails. ## Order of Events For each batch of events in PublishRequest, the order of results in PublishResponse corresponds to the order of events in the publish request. You can correlate each successful or failed result with the corresponding event in the publish request. When you publish multiple batches of events, each in a PublishRequest, the order of batches published matches the order of batches received. ## Identify an Event with ProducerEvent.id and PublishResponse.correlationKey The event `id` field in a PublishRequest, `ProducerEvent.id`, identifies an event published. It can be either system-generated or user-provided. If you don’t specify an `id` value in the PublishRequest and the publish call succeeds, `PublishResult.correlationKey` is populated with a system-generated UUID. If the publish call fails, `PublishResult.correlationKey` is empty. When publishing an event, you can specify a custom `id` value in `ProducerEvent.id`. If you set a custom `id`, the `PublishResult.correlationKey` field in the returned PublishResponse is the custom `id` value for successful and failed publish calls. The custom `id` value can be any string up to 36 characters. We recommend specifying a unique `id` value, such as a UUID. ## Retry Publishing Failed Events If the publishing fails due to a malformed event payload or a missing `schema_id` field, fix your request and try publishing again. If the cause of the publishing failure is due to a temporary server issue, retry publishing. To be able to retry publishing events, set the ProducerEvent `id` value in PublishRequest in the PublishStream call to a custom `id` value. When the `PublishStream` call fails, the returned PublishResult in PublishResponse contains the `correlationKey` field set to the `id` value that you provided. To identify the event published, match `PublishResult.correlationKey` with `ProducerEvent.id` (the event ID). You can then republish the event. It’s advisable to limit the number of retries so that your publish calls don’t execute too many times in case the publishing keeps failing. ## Keeping the Publishing Stream Alive To hold onto the stream, a client must send a valid publish request with one or more events periodically. The server keeps the stream alive as long as it has been less than 30 minutes since the last PublishRequest that the client sent or since the last PublishResponse that the server sent. When the client is notified of the stream closure, it must make a new `PublishStream` call to resume publishing. ## Handling Errors If an error occurs, the long-lived connection is stopped. Retry the `PublishStream` RPC method call to create a new long-lived connection. For more information, see [Handle Errors](/docs/platform/pub-sub-api/guide/handling-errors.md). **See Also** - [Java Quick Start](/docs/platform/pub-sub-api/guide/qs-java-quick-start-parent.html) # Pull Subscription and Flow Control Unlike push-based subscriptions in which a client waits to receive new events pushed from the server, with a pull subscription, a client requests events from the server. To request events, a client calls the [Subscribe RPC Method](https://developer.salesforce.com/docs/platform/pub-sub-api/references/methods/subscribe-rpc.html) or the [ManagedSubscribe RPC Method (Beta)](/docs/platform/pub-sub-api/references/methods/managedsubscribe-rpc.html) and passes in a FetchRequest or ManagedFetchRequest, respectively. The client sets the number of events that it has capacity to process in the `num_requested` field of FetchRequest or ManagedFetchRequest. This event flow control ensures that the client doesn’t get overwhelmed with more events that it can handle if there’s a spike in event publishing. A client can send one or more FetchRequests as part of the same Subscribe call, and it can receive one or more FetchResponses in return. Similarly, a client can send one or more ManagedFetchRequests as part of the same ManagedSubscribe call, and it can receive one or more ManagedFetchResponses in return. ![Subscribe RPC requests and responses](https://developer.salesforce.com/docs/media/pubsub-client-subscribe-rpc_3.png '{"class": "image-lg", "title": "Subscribe RPC requests and responses"}') **note:** In the remaining part of this page, request refers to FetchRequest or ManagedFetchRequest. Also, response refers to FetchResponse or ManagedFetchResponse. The maximum number of events that can be requested in a Subscribe or ManagedSubscribe call across all requests is 100. If a client requests more than 100 events in a Subscribe or ManagedSubscribe call, the server considers the number of requested events to be 100. The client sets the number of events to request in the `num_requested` field of FetchRequest or ManagedFetchRequest. The server adjusts the total number of requested events based on the requested events sent in one or more requests and how many events were already delivered. If the client requests more events before the server finishes the last requested amount, the server appends the new amount to the current batch of events it still needs to deliver. These sections describe ways to control event flow. ## Single-Event Flow: Requesting and Processing One Event at a Time In this subscription flow, the client can request one event at a time. It waits for the event to be received, processes the event, and then requests a new event. This diagram shows a basic sequence of requesting and processing one event at a time. The sequence is: 1. A client requests an event in a request to Salesforce. 2. When an event is available in the event bus, Salesforce delivers the event to the client. 3. After receiving the event, the client initiates a new request requesting one more event. 4. When a new event is available in the event bus, Salesforce delivers the event to the client. ![Basic flow of one event requested at a time](https://developer.salesforce.com/docs/media/pubsub-flow-one-event_3.png '{"class": "image-lg", "title": "Basic flow of one event requested at a time"}') ## Events Delivered in One or More Responses The server can send the requested events in one or more responses. If there are enough events in the event bus, the requested events can be delivered in one response, or in multiple responses if the total size of events exceeds 3 MB. If there aren’t enough events in the event bus, the client receives a partial number of events in one response. As new events are published to the event bus, they’re sent immediately to the client in a new response as long as there are still pending events to be delivered to the client. The response contains the number of pending events not yet delivered in the `pending_num_requested` field. The client can use this number to decide how many new events to request, if it has the capacity to continue processing events. For example, a client’s processing capacity is 100 events and it requests 100 events. Then the client receives 60 events in a response with a `pending_num_requested` field value of 40. The client can request 60 events again, which corresponds to 100 minus the `pending_num_requested` value of 40. ## Basic Flow: Requesting New Events After All Requested Events Are Delivered In this subscription flow, the client initiates a request and waits until all requested events are delivered before it sends a new request to ask for more events. In this example, the client determines that it can process 100 events and sends a request for 100 events, which corresponds to the maximum number of events that can be requested from Salesforce. The client then waits for all events to be delivered before initiating a new request to ask for more events. The events can be delivered all at once, in one response, or in multiple responses. In this example, the events are delivered in multiple responses. The sequence in this diagram is: 1. A client requests 100 events in a request to Salesforce. 2. As events become available in the event bus, Salesforce delivers the events to the client in multiple responses. 3. After receiving all the events, the client initiates a new request for 100 more events. ![A flow with multiple responses](https://developer.salesforce.com/docs/media/pubsub-flow-multiple-responses_3.png '{"class": "image-lg", "title":"A flow with multiple responses"}') ## Complex Flow: Requesting New Events While Processing Received Events In this subscription flow, the client initiates a request, and while events are delivered, sends a new request to ask for more events. In this example, the client determines how many events to ask for in the next request based on the number of events received so far. The server delivers events as they become available in the event bus in multiple responses. The sequence in this diagram is: 1. A client requests 100 events in a request to Salesforce. 2. The first response returns 60 events. 3. The client sends another request for 60 new events. 4. The requested events from the first request and the second request are delivered in multiple responses. 5. After receiving all the events, the client initiates a new request for 100 more events. ![A complex flow with multiple requests and responses](https://developer.salesforce.com/docs/media/pubsub-flow-complex_3.png '{"class": "image-lg", "title": "A complex flow with multiple requests and responses"}') ## Keeping the Subscription Stream Alive from the Client The Subscribe or ManagedSubscribe call stream stays open as long as there are pending events to be sent to the client. The number of pending events is based on the number of events the client requested and the number of events delivered. If the server sends all requested events to the client, there are no more pending events to be delivered and `pending_num_requested` in response is equal to zero. In this case, to keep the subscription stream open, the client must send a new request within 60 seconds from when the last response was sent. If the client fails to do so, the subscription stream closes and the client must call Subscribe again to open a new stream. If there are pending events to be sent to the client (`pending_num_requested` in FetchResponse or ManagedFetchResponse is greater than zero), but no new events are available in the event bus, the server keeps the stream alive by sending keepalive messages within 270 seconds. For more information, see [Subscribe Keepalive Behavior](https://developer.salesforce.com/docs/platform/pub-sub-api/references/methods/subscribe-rpc.html#subscribe-keepalive-behavior) and [ManagedSubscribe Keepalive Behavior](/docs/platform/pub-sub-api/references/methods/managedsubscribe-rpc.html#managedsubscribe-keepalive-behavior). ## Processing Multiple Subscriptions To process events from multiple topics in separate subscription streams, a client can make multiple Subscribe calls or multiple ManagedSubscribe calls for different managed subscription IDs. The number of events requested and the maximum requested value in a request are per Subscribe or ManagedSubscribe call. If you make multiple calls in one client, event flow is controlled separately for each subscription. For example, you can subscribe to one topic with a low batch size of 10 events and to another topic with a high batch size of 100 events. To stay within your event allocations, you can use methods that reduce event consumption for one topic. - Lower the batch size of requested events. - Send requests less frequently. - Pause a subscription by stopping it and making another Subscribe or ManagedSubscribe call. The received event rate depends on your client implementation and whether you’re processing events from each subscription in parallel or sequentially. # Python Quick Start for Pub/Sub API In this quick start, you learn how to build a Pub/Sub API client in Python. The steps walk you through generating the stub files, authenticating to Salesforce, configuring events, writing code to subscribe to events, and writing code to publish events. **note:** The quick start provides enough instructions and code snippets so that you can build your own client. The code example uses username and password authentication for simplicity, but we recommend that you use OAuth in production apps. For more information, see [Authentication](https://developer.salesforce.com/docs/platform/pub-sub-api/guide/supported-auth.md). You can find full code examples in [https://github.com/forcedotcom/pub-sub-api](https://github.com/forcedotcom/pub-sub-api). The [Go code examples](https://github.com/forcedotcom/pub-sub-api/tree/main/go) use OAuth. The full examples aren’t intended for production use and haven’t undergone thorough functional and performance testing. You can use these examples as a starting point to build your own client. **In this section:** - [Step 1: Generate the Stub Files](https://developer.salesforce.com/docs/platform/pub-sub-api/guide/qs-generate-stub.md) - [Step 2: Build the Python Client](https://developer.salesforce.com/docs/platform/pub-sub-api/guide/qs-build-python-client.md) - [Step 3: Set Up Events](https://developer.salesforce.com/docs/platform/pub-sub-api/guide/qs-set-up-events.md) - [Step 4: Write Code That Subscribes to an Event Channel](https://developer.salesforce.com/docs/platform/pub-sub-api/guide/qs-subscribe.md) - [Step 5: Write Code That Publishes a Platform Event Message](https://developer.salesforce.com/docs/platform/pub-sub-api/guide/qs-publish.md) # Quick Starts Quick starts are provided in Java and Python to show you how to publish and subscribe to events by using Pub/Sub API. **In this section:** - [Java Quick Start](https://developer.salesforce.com/docs/platform/pub-sub-api/guide/qs-java-quick-start-parent.md) - [Python Quick Start](https://developer.salesforce.com/docs/platform/pub-sub-api/guide/qs-python-quick-start.md) # Retry Long-Lived RPC Calls After an Error Occurs When an error occurs, a long-lived RPC call, including `PublishStream`, `Subscribe`, and `ManagedSubscribe (beta)`, stops running and the Pub/Sub API service closes the streaming connection. To create a new streaming connection, you must retry the RPC call. How you retry the long-lived RPC call depends on the error received in the gRPC [StatusRuntimeException](https://grpc.github.io/grpc-java/javadoc/io/grpc/StatusRuntimeException.html), which you can catch in your client. **tip:** Because errors are sometimes transient and can go away after a period of time, we recommend that you limit the number of retry attempts and that you retry the RPC calls with exponential backoff. Increasing the time between calls ensures that you don’t keep receiving the same error and exhaust all your retry attempts. To inspect a Pub/Sub API error code value, get the gRPC [StatusRuntimeException](https://grpc.github.io/grpc-java/javadoc/io/grpc/StatusRuntimeException.html) Trailers section of the exception by calling [getTrailers()](https://grpc.github.io/grpc-java/javadoc/io/grpc/StatusRuntimeException.html#getTrailers()) on the exception. The Pub/Sub API error code provides information about the cause of the failure. For a list of error codes, see [Error Codes](https://developer.salesforce.com/docs/platform/pub-sub-api/guide/handling-errors.md#error-codes). For example, the error code in this example is `sfdc.platform.eventbus.grpc.subscription.fetch.replayid.corrupted`. It means that the specified Replay ID value is invalid. One possible cause is that it’s an old Replay ID outside the 72-hour retention period. ``` === GRPC Exception === io.grpc.StatusRuntimeException: INVALID_ARGUMENT: The Replay ID validation failed. Ensure that the Replay ID is valid. rpcId: 2f6b4cee-3525-49d1-8fdb-0bd3d662062f === Trailers === [Trailer] = content-type [Value] = application/grpc [Trailer] = rpc-id [Value] = 2f6b4cee-3525-49d1-8fdb-0bd3d662062f [Trailer] = error-code [Value] = sfdc.platform.eventbus.grpc.subscription.fetch.replayid.corrupted [Trailer] = type [Value] = Subscribe ``` ## How to Retry the Subscribe RPC Method To retry the [Subscribe RPC Method](/docs/platform/pub-sub-api/references/methods/subscribe-rpc.html) after an error occurs, use a `ReplayPreset` option to specify from which position in the event bus to resubscribe depending on the error received. Here are examples of how to retry the `Subscribe` call based on the error code. - If a temporary server error is returned, retry the `Subscribe` call with the `CUSTOM` `ReplayPreset` enum value. Pass in the Replay ID value that you saved from the last processed event to receive events after the one that the client last processed. The subscription resumes from where it left off, after that event with the specified Replay ID. - If the Replay ID that you passed in with the `CUSTOM ReplayPreset` enum value is corrupted, retry the `Subscribe` call with the `LATEST ReplayPreset` enum value to receive new events only, or use the `EARLIEST` option to receive all events that are stored. Because it can slow down performance when a large volume of events is stored in the event bus, use the `EARLIEST` option sparingly. - If you aren’t interested in the events that were missed during client shutdown, retry the `Subscribe` call by using the `LATEST ReplayPreset` enum value to get new events. For more details about the `Subscribe` call options in the `ReplayPreset` enum value, see [Replaying an Event Stream](/docs/platform/pub-sub-api/references/methods/subscribe-rpc.html#replaying-an-event-stream). ### Example This code snippet shows how to get the error code from the trailers portion of StatusRuntimeException. It then checks for some common error codes and follows a strategy to determine which point in the stream to resubscribe from. This example uses a scheduler to schedule the `Subscribe` RPC call after a delay, if necessary. The implementation of the scheduler isn’t shown in this example. ``` public static String ERROR_REPLAY_ID_VALIDATION_FAILED = "fetch.replayid.validation.failed"; public static String ERROR_REPLAY_ID_INVALID = "fetch.replayid.corrupted"; public static String ERROR_SERVICE_UNAVAILABLE = "service.unavailable"; // ... @Override public void onError(Throwable t) { printStatusRuntimeException("Error during Subscribe", (Exception) t); logger.info("Retries remaining: " + retriesLeft.get()); if (retriesLeft.get() == 0) { logger.info("Exhausted all retries. Closing Subscription."); isActive.set(false); } else { retriesLeft.decrementAndGet(); Metadata trailers = ((StatusRuntimeException)t).getTrailers() != null ? ((StatusRuntimeException)t).getTrailers() : null; String errorCode = (trailers != null && trailers.get(Metadata.Key.of("error-code", Metadata.ASCII_STRING_MARSHALLER)) != null) ? trailers.get(Metadata.Key.of("error-code", Metadata.ASCII_STRING_MARSHALLER)) : null; // Closing the old stream for sanity serverStream.onCompleted(); ReplayPreset retryReplayPreset = ReplayPreset.LATEST; ByteString retryReplayId = null; long retryDelay = getBackoffWaitTime(); // Retry strategies that can be implemented based on the error type. if(errorCode != null && !errorCode.isEmpty()) { if (errorCode.contains(ERROR_SERVICE_UNAVAILABLE)) { logger.info("Service currently unavailable. Trying again with stored Replay."); retryDelay = SERVICE_UNAVAILABLE_WAIT_BEFORE_RETRY_SECONDS * 1000; retryReplayPreset = ReplayPreset.CUSTOM; retryReplayId = getStoredReplay(); } else if (errorCode.contains(ERROR_REPLAY_ID_VALIDATION_FAILED) || errorCode.contains(ERROR_REPLAY_ID_INVALID)) { logger.info("Invalid or no replayId provided in FetchRequest for CUSTOM Replay. Trying again with EARLIEST Replay."); retryReplayPreset = ReplayPreset.EARLIEST; } else { logger.info("Retrying with LATEST Replay."); } } else { logger.info("Unknown error. Retrying with LATEST Replay."); } logger.info(String.format("Retrying in %s ms.", retryDelay)); retryScheduler.schedule(new RetryRequestSender(retryReplayPreset, retryReplayId), retryDelay, TimeUnit.MILLISECONDS); } } ``` ## How to Retry the ManagedSubscribe RPC Method (Beta) The process of retrying the [ManagedSubscribe RPC Method (Beta)](/docs/platform/pub-sub-api/references/methods/managedsubscribe-rpc.html) is simpler than retrying the `Subscribe` RPC method. If a client retries the `ManagedSubscribe` call after sending commit requests to store the Replay ID of a processed event on the server, the subscription resumes after the event with the committed Replay ID. The client doesn’t manage the Replay ID’s storage. Also, the `ManagedSubscribe` RPC method uses event replay options that you set in the ManagedEventSubscription metadata by using Metadata API or Tooling API. These replay options are used if no Replay ID was committed or the committed Replay ID is invalid. - If no Replay ID was committed on the server and you retry the `ManagedSubscribe` call, the subscription restarts from the value specified in the `defaultReplay` field of ManagedEventSubscription. If `defaultReplay` is set to `LATEST`, the subscription receives new events only. If `defaultReplay` is set to `EARLIEST`, the subscription starts from the earliest events stored in the event bus. - If the committed Replay ID is invalid and you retry the `ManagedSubscribe` call, the subscription restarts from the value specified in the `errorRecoveryReplay` field of ManagedEventSubscription. If `errorRecoveryReplay` is set to `LATEST`, the subscription receives new events only. If `errorRecoveryReplay` is set to `EARLIEST`, the subscription starts from the earliest events stored in the event bus. For more information, see [ManagedEventSubscription (Beta)](https://developer.salesforce.com/docs/atlas.en-us.api_meta.meta/api_meta/meta_managedeventsubscription.htm) in the _Metadata API Developer Guide_. ## How to Retry Publishing Failed Events The [PublishStream RPC Method](/docs/platform/pub-sub-api/references/methods/publishstream-rpc.html) uses bidirectional streaming. It can send a stream of publish requests while receiving a stream of publish responses from the server. The server returns a PublishResponse for each PublishRequest when publishing is complete for a batch of events. A PublishResponse contains a list of PublishResult objects, each containing the publishing result for one event. To preserve the order of event batches, wait for the PublishResponse to be received for a published batch of events before sending a new PublishRequest for a new batch. Be aware that this method can slow down the speed of your publishing process. 1. Publish a batch of events in PublishRequest, and then wait for the PublishResponse to be returned before sending a new PublishRequest. 2. In PublishResponse, inspect each PublishResult for each event. 3. If a PublishResult has an error for one or more events, republish the events. 4. Repeat this process to publish a new batch of events, and then inspect new publish results. If the order of event batches isn’t important, you can inspect the publish results in each PublishResponse as it’s received. Then republish failed events whose PublishResult contains an error. However, this method can cause one batch to finish publishing before you republish a failed batch of events. ### Correlating Publish Results with Published Events To determine which PublishResult corresponds to which event published, correlate the failed publish result with the event ID (`ProducerEvent.Id`) by using the `correlationKey` field that’s returned in PublishResult, which is part of PublishResponse. The `correlationKey` is set to the value that you set in `ProducerEvent.id` in PublishRequest when you publish the event. To identify the event published, match `PublishResult.correlationKey` with `ProducerEvent.id` (the event ID). You can then republish the event. It’s advisable to limit the number of retries so that your publish calls don’t execute too many times in case the publishing keeps failing. To view the messages and fields of the pub-sub-api service, see [pubsub_api.proto](https://github.com/forcedotcom/pub-sub-api/blob/main/java/src/main/proto/pubsub_api.proto) in GitHub. For more information about the correlation key, see [Identify an Event with ProducerEvent.id and PublishResponse.correlationKey](/docs/platform/pub-sub-api/references/methods/publishstream-rpc.html#identify-an-event-with-producereventid-and-publishresponsecorrelationkey). ### Example This code snippet shows how to retry publishing failed events. It limits the retries up to a certain number set in the constant `MAX_NUM_OF_RETRIES_PER_BATCH` value. The example uses a private helper method, `validatePublishResponse()`, to get a list of PublishResults that have errors (`PublishResult.error` is true), which represents the failed events. Next, it gets the correlation key from each PublishResult and matches it to a ProducerEvent ID by using a map holding correlation keys and ProducerEvents. Finally, it retries publishing the failed events. ``` while (true) { batchCount++; logger.info("publishing " + NUM_EVENTS_PER_BATCH + " events for batch " + batchCount); // Assume that each event will have a unique ID (i.e., correlationKey) assigned to it. // Store the corresponding ProducerEvent so that we can look up specific events by // correlationKey when a partial publish failure occurs. Map eventsByCorrelationKey = new HashMap<>(NUM_EVENTS_PER_BATCH); for (int i = 0; i < NUM_EVENTS_PER_BATCH; i++) { ProducerEvent producerEvent = generateProducerEvent(i); eventsByCorrelationKey.put(producerEvent.getId(), producerEvent); } PublishRequest publishRequest = PublishRequest.newBuilder().setTopicName(busTopicName) .addAllEvents(eventsByCorrelationKey.values()).build(); // Publish batch of events. requestObserver.onNext(publishRequest); // Get the failed publishing results by using a helper method, validatePublishResponse() List failedEvents = validatePublishResponse( errorStatuses, batchFinishedLatchRef.get(), eventsByCorrelationKey.size(), publishResponses, batchCount); // Loop to retry any failed events for (int i = 1; i < MAX_NUM_OF_RETRIES_PER_BATCH + 1; i++) { if (failedEvents == null || failedEvents.size() == 0) { break; } logger.info("retry attempt " + i + ": retrying " + failedEvents.size() + " event(s) from batch " + batchCount); batchFinishedLatchRef.set(new CountDownLatch(1)); // Use the correlationKey provided in each PublishResult to look up the original // corresponding ProducerEvent List eventsToRetry = Lists.newArrayListWithExpectedSize(failedEvents.size()); for (int j = 0; j < failedEvents.size(); j++) { String correlationKey = failedEvents.get(j).getCorrelationKey(); ProducerEvent producerEvent = eventsByCorrelationKey.get(correlationKey); eventsToRetry.add(j, producerEvent); } // Create a retry PublishRequest that contains all events that must be retried PublishRequest retryPublishRequest = PublishRequest.newBuilder().setTopicName(busTopicName) .addAllEvents(eventsToRetry).build(); // Retry publishing of the failed events requestObserver.onNext(retryPublishRequest); failedEvents = validatePublishResponse(errorStatuses, batchFinishedLatchRef.get(), eventsToRetry.size(), publishResponses, batchCount); } ``` **note:** The process of republishing failed events with the unary [Publish RPC Method](/docs/platform/pub-sub-api/references/methods/publish-rpc.html) is similar to that of the `PublishStream` RPC. You can correlate events with a PublishResult by using the correlation key. The only difference between the two RPCs is that `Publish` RPC has no long-lived connection and the call ends after it executes. # Set a Deadline on the `GetSchema`, `GetTopic`, and `Publish` Unary RPC Methods Set a deadline of 10 minutes or less on the `GetSchema`, `GetTopic`, and `Publish` unary RPC methods so that you can retry a request that takes too long. When you set a deadline, a `DEADLINE_EXCEEDED` status code is returned if you don't receive a response within the allotted time. Without a deadline, it is difficult for you to know if a request should be retried. For more information and sample code, see [the gRPC deadlines documentation](https://grpc.io/docs/guides/deadlines/). # Step 1: Create ManagedEventSubscription with Tooling API Create a managed subscription configuration for `Order_Event__e`. **Prerequisites** - Create the `Order_Event__e` platform event by following the steps in [Step 1: Define a Platform Event](https://developer.salesforce.com/docs/platform/pub-sub-api/guide/qs-java-prereqs.md) in the main Java quick start. - To complete the steps, you can use your favorite REST API tool. We recommend using Postman with the Salesforce API Collection, which contains handy templates for Salesforce API calls, including calls for creating ManagedEventSubscription. For instructions on how to set up Postman, see [Quick Start: Connect Postman to Salesforce](https://trailhead.salesforce.com/content/learn/projects/quick-start-connect-postman-to-salesforce) in Trailhead. 1. Send a POST request to this REST endpoint: `/services/data/v61.0/tooling/sobjects/ManagedEventSubscription` If you’re using Postman, in the Salesforce Platform APIs collection, expand **Event Platform** and **Managed Event Subscriptions**, and then click **Create managed event subscription**. If you don't find the Managed Event Subscriptions folder, refresh the collection. Click **View more actions** (**...**) next to the collection name, click **Pull changes**, and in the new tab, click **Pull Changes**. Use this request body. ``` { "FullName": "My_Managed_Subscription", "Metadata": { "label" : "My Managed Subscription", "topicName" : "/event/Order_Event__e", "defaultReplay": "LATEST", "state" : "RUN", "errorRecoveryReplay" : "LATEST" } } ``` 2. After you create the ManagedEventSubscription in Tooling API, the response returned contains the ID of the created ManagedEventSubscription record, similar to this response. Save the ID. ``` { "id": "18xKj000000000BIAQ", "success": true, "errors": [], "warnings": [], "infos": [] } ``` # Step 1: Define a Platform Event Create a custom platform event named `Order Event`. The steps in this quick start use this platform event. 1. From Setup, in the Quick Find box, enter `Platform Events`, and then select **Platform Events**. 2. On the Platform Events page, click **New Platform Event**. 3. Complete the standard fields, and for the description, enter `Example event for the Pub/Sub API client`. 4. Save your work. 5. To add a field, in the Custom Fields & Relationships related list, click **New**. 6. Follow the custom field wizard to set up these fields. - Field label: `Order Number` (API name: `Order_Number__c`), field type: Text(18) - Field label: `City` (API name: `City__c`), field type: Text(50) - Field label: `Amount` (API name: `Amount__c`), field type: Number(16,2) # Step 1: Generate the Stub Files **Prerequisites** - Python 3.10 or later. Install Python from [https://www.python.org/downloads/](https://www.python.org/downloads/). 1. Install the Python package manager by running this command in the terminal. ``` pip3 install grpcio grpcio-tools avro requests ``` You can use a different package manager. 2. Clone the GitHub repository for Pub/Sub API from [_https://github.com/forcedotcom/pub-sub-api_](https://github.com/forcedotcom/pub-sub-api). The proto file name is `pubsub_api.proto`. 3. Switch to the python directory. ``` cd python ``` 4. Generate the stubs for the Pub/Sub API by running this command from the cloned directory. ``` python3 -m grpc_tools.protoc --proto_path=../ pubsub_api.proto --python_out=. --grpc_python_out=. ``` This command generates two files in your current directory: `pubsub_api_pb2.py` and `pubsub_api_pb2_grpc.py`. It also generates client and server code and protocol buffer code for populating, serializing, and retrieving message types. # Step 2: Build the Client Fork and clone the client from GitHub, and then build the client. **Prerequisites** - Java 11 (see [Java SE 11 Archive Downloads](https://www.oracle.com/java/technologies/javase/jdk11-archive-downloads.html)) - Apache Maven (see [Installing Apache Maven](https://maven.apache.org/install.html)) 1. Fork and then clone the `pub-sub-api` GitHub repository: https://github.com/forcedotcom/pub-sub-api. For more information, see [Fork a repository](https://docs.github.com/en/pull-requests/collaborating-with-pull-requests/working-with-forks/fork-a-repo) in _GitHub Docs_. Forking a repository enables you to save your changes in your forked repository. If you want to skip forking the repo, you can just clone the repo, but you can’t save changes that you make in the `pub-sub-api` repo. 2. In a terminal window, go to the `java` folder in your cloned `pub-sub-api` folder. 3. Build the Java client package and generate required sources from the proto file. From the `java` folder, enter: `mvn clean install` # Step 2: Build the Python Client 1. Create a Python file in any editor. For example, create a file named `PubSubAPIClient.py`. 2. To import these modules, add these import statements at the top of the file. ``` import grpc import requests import threading import io import pubsub_api_pb2 as pb2 import pubsub_api_pb2_grpc as pb2_grpc import avro.schema import avro.io import time import certifi import json ``` 3. Set a semaphore at the beginning of the program. ``` semaphore = threading.Semaphore(1) ``` Because of Python gRPC’s design, the program shuts down immediately if no response comes back in the milliseconds between calling an RPC and the end of the program. When you set a semaphore, the client keeps running indefinitely. In this example, you define the semaphore globally. Because this sample starts one subscription via one Subscribe RPC call, the global semaphore works. However, the global semaphore isn’t appropriate for multiple subscriptions because the semaphore is shared. If you change this sample code to make multiple Subscribe RPC calls, make sure that each Subscribe call uses a separate semaphore and the semaphore isn’t shared among subscriptions. 4. Create a global variable to store the replay ID. ``` latest_replay_id = None ``` 5. Set up the gRPC channel, and generate the stub. The gRPC channel points to the Pub/Sub API endpoint `api.pubsub.salesforce.com:7443`. In this quick start, we use port `7443`, but you can alternatively use port `443`. If your company uses a firewall for network security, make sure that the Pub/Sub API endpoint is allowed in the firewall configuration. ``` with open(certifi.where(), 'rb') as f: creds = grpc.ssl_channel_credentials(f.read()) with grpc.secure_channel('api.pubsub.salesforce.com:7443', creds) as channel: # All of the code in the rest of the tutorial will go inside this block. # Make sure that the indentation of the new code you add starts from this comment’s block. ``` 6. Retrieve your session token. Replace the username, password, and API login URL with values for your Salesforce org. Replace `MyDomainName` with your org’s My Domain name. For more information, see [My Domain URL Formats ](https://help.salesforce.com/s/articleView?id=xcloud.domain_name_app_url_changes.htm&type=5) in _Salesforce Help_. If you’re using a production instance, the API login URL is `https://MyDomainName.my.salesforce.com/services/Soap/u/59.0/`. If it’s a sandbox, the URL is `https://MyDomainName--SandboxName.sandbox.my.salesforce.com/services/Soap/u/59.0/`. Send a POST request formatted like this to the login URL. ``` username = '{your username}' password = '{your password}' url = '{the appropriate login URL}' headers = {'content-type': 'text/xml', 'SOAPAction': 'login'} xml = f""" """ res = requests.post(url, data=xml, headers=headers, verify=False) #Optionally, print the content field returned print(res.content) ``` **note:** - When you copy the code snippet, ensure that it aligns with the indented block of code above it. In Python, code blocks are delimited by their indentation. - If you haven’t set up a range of trusted IP addresses for your org, append a security token to your password. For more information, see [Reset Your Security Token](https://help.salesforce.com/s/articleView?id=xcloud.user_security_token.htm&type=5) and [Set Trusted IP Ranges for Your Org](https://help.salesforce.com/s/articleView?id=xcloud.security_networkaccess.htm&type=5). - The code example uses username and password authentication for simplicity, but we recommend that you use OAuth in production apps. For more information, see [Supported Authentication](https://developer.salesforce.com/docs/platform/pub-sub-api/guide/supported-auth.md). 7. Run this client by entering this command on the command line. ``` python3 PubSubAPIClient.py ``` When the request returns, you have XML-formatted data in the response content field `res.content`. It contains a session ID wrapped within the `` tags. It also contains the server URL wrapped within the `` tags and the org ID within the `` tags under ``. Note those values because you use them in the next step. **note:** After getting the session information, you can delete or comment out the login code you added in the previous step. To comment out a line in Python, prepend it with `#`. 8. Store the authorization information, including session ID, instance URL, and org ID, in a tuple called `authmetadata`. Each element in this tuple is also a tuple. You use this information when subscribing to a channel or publishing events. 1. Replace the `sessionid` placeholder value with the session ID that you got from the previous step. 2. Replace the `instanceurl` placeholder value with the first part of the server URL that you got from the previous step, without the path portion. For example, `https://MyDomainName.my.salesforce.com`. 3. Replace the `tenantid` placeholder value with the org ID that you got from the previous step. For more information about the headers, see [Include Authorization Headers in RPC Method Calls](https://developer.salesforce.com/docs/platform/pub-sub-api/references/methods/rpc-method-headers.html). ``` sessionid = '{the session ID you just got from the XML}' instanceurl = '{the first part of the server URL ending in .com}' tenantid = '{organization ID}' authmetadata = (('accesstoken', sessionid), ('instanceurl', instanceurl), ('tenantid', tenantid)) ``` 9. Generate your stub object like this. `PubSubStub` comes from the `pubsub_api_pb2_grpc.py` file, which you generated in the previous step. ``` stub = pb2_grpc.PubSubStub(channel) ``` # Step 2: Start the Managed Subscription Start a managed subscription by using the `ManagedSubscribe.java` class, which calls the `ManagedSubscribe` RPC method (beta). 1. Set up the Pub/Sub API Java client by following the steps in [Step 2: Build the Client](https://developer.salesforce.com/docs/platform/pub-sub-api/guide/qs-java-build-client.md) in the main Java quick start. 2. Configure the required client parameters, such as your Salesforce org login URL and credentials, in `arguments.yaml` by following the steps in [Step 3: Configure Client Parameters](https://developer.salesforce.com/docs/platform/pub-sub-api/guide/qs-java-parameters.md) in the main Java quick start. 3. In `arguments.yaml`, specify the managed subscription ID. In this case, you use the ID `MANAGED_SUB_ID=18xKj000000000BIAQ`. 4. To run the `ManagedSubscribe` RPC example, enter `./run.sh genericpubsub.ManagedSubscribe`. **important:** When you create a ManagedEventSubscription record, there can be a delay before the new configuration takes effect in Pub/Sub API. If the subscription returns an error saying that the managed subscription configuration isn’t found, wait several minutes and then restart it. Because no events are published yet, the client doesn’t receive any events yet. In the next step, you publish some events. # Step 3: Configure Client Parameters The Java client example contains parameters that are passed to the client. Enter values for the required parameters, such as the Pub/Sub API endpoint and your Salesforce login credentials. Optional parameters, such as the event topic, have default values. Provide values only if you want to change the defaults. 1. In your cloned repo, navigate to `src/main/resources` and open the `arguments.yaml` file. 2. Enter the required parameters that are used to establish the streaming connection. - PUBSUB_HOST: Specify the Pub/Sub API endpoint `api.pubsub.salesforce.com`. - PUBSUB_PORT: Specify the Pub/Sub API port to be used, `7443` or `443`. - LOGIN_URL: Specify the login URL of the Salesforce org to use to run the examples, starting with `https://`. For login URL formats, see [My Domain Login and Application URL Formats](https://help.salesforce.com/s/articleView?id=xcloud.domain_name_url_formats.htm&type=5) in Salesforce Help. - Provide values for either USERNAME and PASSWORD or ACCESS_TOKEN and TENANT_ID. - USERNAME and PASSWORD: To authenticate via username and password, specify the Salesforce username and password. - ACCESS_TOKEN and TENANT_ID: To authenticate via a session token, specify the sessionToken and tenant ID of the Salesforce org. **Optional Parameters** Change the optional parameters if you want to use values other than the defaults. Here’s a description of the optional parameters. - TOPIC: This parameter is the topic for publishing and subscribing to platform events and for subscribing to change data capture events. The default value is `/event/Order_Event__e`. You can specify the topic of one event or, to subscribe to multiple events, specify the topic of a channel. To learn more about channels, see [Group Platform Events into One Stream with a Custom Channel](https://developer.salesforce.com/docs/atlas.en-us.platform_events.meta/platform_events/platform_events_subscribe_custom_channels.htm) in the _Platform Events Developer Guide_ and [Subscription Channels](https://developer.salesforce.com/docs/atlas.en-us.change_data_capture.meta/change_data_capture/cdc_subscribe_channels.htm) in the _Change Data Capture Developer Guide_. - NUMBER_OF_EVENTS_TO_PUBLISH: This parameter is the number of events to publish when using the PublishStream RPC. The default value is 5. - SINGLE_PUBLISH_REQUEST: This parameter indicates whether to add events to a single PublishRequest (true) or in different PublishRequests (false). The default value is false. - NUMBER_OF_EVENTS_IN_FETCHREQUEST: This parameter is the number of events that the Subscribe RPC requests from the server in each FetchRequest. The default value is 5. The example fetches at most 5 events in each Subscribe request. If you pass in more than 5, it sends multiple Subscribe requests with at most 5 events requested in each FetchRequest. For more information about requesting events, see [Pull Subscription and Flow Control](https://developer.salesforce.com/docs/platform/pub-sub-api/guide/flow-control.html). - REPLAY_PRESET: This parameter specifies where to start the subscription. To fetch past events that are stored in the event bus, use REPLAY_PRESET with REPLAY_ID. For more information, see the [Readme page](https://github.com/forcedotcom/pub-sub-api/tree/main/java) in the pub-sub-api GitHub repository and [Replaying an Event Stream](https://developer.salesforce.com/docs/platform/pub-sub-api/references/methods/subscribe-rpc.html#replaying-an-event-stream) in the Pub/Sub API documentation. # Step 3: Publish Events and Verify the Managed Subscription Publish events by using the `PublishStream` RPC. Then verify that the subscription resumes after the last received event after restarting it and calling `ManagedSubscribe` RPC with the same managed subscription ID. **note:** The `ManagedSubscribe` RPC method is part of a Beta Service. Customer may opt to try such Beta Service in its sole discretion. Any use of the Beta Service is subject to the applicable Beta Services Terms provided at [Agreements and Terms](https://www.salesforce.com/company/legal/agreements/). 1. In `arguments.yaml`, supply these arguments. - `SINGLE_PUBLISH_REQUEST: true`: This argument specifies that all events are sent in one publish request. - `TOPIC: /event/Order_Event__e`: This argument specifies the topic for event publishing. 2. Modify the publishing code to specify the event fields and values to publish. In `java/src/main/java/utility/CommonContext.java`, modify the `public List createEventMessages(Schema schema, final int numEvents)` method. Replace the `` placeholder value of the `CreatedById` field with a valid user ID in your Salesforce org. To get the user ID, see [Find the Salesforce ID for a User or Profile](https://help.salesforce.com/s/articleView?id=000381643&type=1). ``` .set("CreatedById", "") ``` 3. In a second terminal window, go to the top-level `java` folder in your `pub-sub-api` folder. 4. To build the Java client, enter `mvn clean install`. 5. To run the `PublishStream` RPC example, enter `./run.sh genericpubsub.PublishStream`. The example publishes five events as set in the `arguments.yaml` default configuration. It prints out logging information to the terminal. 6. Return to the terminal for the `ManagedSubscribe` client, and verify that the logging output shows the five events received. Also, the log output shows that a commit request is sent after the batch of events is received to commit the Replay ID. The output looks similar to this output. ``` 2024-06-03 09:12:30,942 [main] java.lang.Class - Using grpcHost api.pubsub.salesforce.com and grpcPort 7443 2024-06-03 09:12:31,524 [main] java.lang.Class - Starting managed subscription with developer name My_Managed_Subscription 2024-06-03 09:12:31,738 [grpc-nio-worker-ELG-1-2] utility.APISessionCredentials - Client Trace Id for current request: cd5faf97-8fa1-4efe-a770-d6166067b922 2024-06-03 09:12:36,531 [main] java.lang.Class - Subscription Active. Received a total of 0 events. 2024-06-03 09:12:41,535 [main] java.lang.Class - Subscription Active. Received a total of 0 events. 2024-06-03 09:12:44,520 [grpc-default-executor-1] java.lang.Class - ManagedFetchResponse batch of 5 events pending requested: 0 2024-06-03 09:12:44,520 [grpc-default-executor-1] java.lang.Class - RPC ID: e5219940-7133-b6a4-97a6-9beabbc7573e 2024-06-03 09:12:44,520 [grpc-default-executor-1] java.lang.Class - processEvent - EventID: f23132b9-7df6-446b-8fe6-2b7cee811376 SchemaId: _CADE3qaegX3ECEW0rlbYA 2024-06-03 09:12:44,524 [grpc-default-executor-1] utility.APISessionCredentials - Client Trace Id for current request: 0c1cbc1c-1189-40fe-8b9b-cc64c333e530 2024-06-03 09:12:44,685 [grpc-default-executor-1] java.lang.Class - Received event: { "CreatedDate": 1717431164101, "CreatedById": "005SG000009B3S1YAK", "Order_Number__c": "99", "City__c": "Los Angeles", "Amount__c": 35.0 } 2024-06-03 09:12:44,685 [grpc-default-executor-1] java.lang.Class - processEvent - EventID: 981d92b0-179f-40f4-be72-fc93ae8a3660 SchemaId: _CADE3qaegX3ECEW0rlbYA 2024-06-03 09:12:44,685 [grpc-default-executor-1] java.lang.Class - Received event: { "CreatedDate": 1717431164101, "CreatedById": "005SG000009B3S1YAK", "Order_Number__c": "100", "City__c": "New York", "Amount__c": 20.0 } 2024-06-03 09:12:44,685 [grpc-default-executor-1] java.lang.Class - processEvent - EventID: 94b4b89a-9158-4d4b-beaa-63e3555c4cbd SchemaId: _CADE3qaegX3ECEW0rlbYA 2024-06-03 09:12:44,685 [grpc-default-executor-1] java.lang.Class - Received event: { "CreatedDate": 1717431164101, "CreatedById": "005SG000009B3S1YAK", "Order_Number__c": "101", "City__c": "San Francisco", "Amount__c": 2.0 } 2024-06-03 09:12:44,685 [grpc-default-executor-1] java.lang.Class - processEvent - EventID: dc37a809-542b-4696-b6f5-bede7a7733d0 SchemaId: _CADE3qaegX3ECEW0rlbYA 2024-06-03 09:12:44,685 [grpc-default-executor-1] java.lang.Class - Received event: { "CreatedDate": 1717431164101, "CreatedById": "005SG000009B3S1YAK", "Order_Number__c": "102", "City__c": "San Jose", "Amount__c": 123.0 } 2024-06-03 09:12:44,685 [grpc-default-executor-1] java.lang.Class - processEvent - EventID: 102bedd7-df2c-4daf-b7c9-e25acde4be47 SchemaId: _CADE3qaegX3ECEW0rlbYA 2024-06-03 09:12:44,685 [grpc-default-executor-1] java.lang.Class - Received event: { "CreatedDate": 1717431164101, "CreatedById": "005SG000009B3S1YAK", "Order_Number__c": "103", "City__c": "Boston", "Amount__c": 180.0 } 2024-06-03 09:12:44,685 [grpc-default-executor-1] java.lang.Class - Processed batch of 5 event(s) 2024-06-03 09:12:44,686 [grpc-default-executor-1] java.lang.Class - Sending CommitRequest with CommitReplayRequest ID: bd557942-4fd1-4699-be18-70db3054b33c 2024-06-03 09:12:44,686 [grpc-default-executor-1] java.lang.Class - Fetching more events: 5 2024-06-03 09:12:44,686 [main] java.lang.Class - Subscription Active. Received a total of 5 events. 2024-06-03 09:12:44,795 [grpc-default-executor-1] java.lang.Class - ManagedFetchResponse batch of 0 events pending requested: 5 2024-06-03 09:12:44,796 [grpc-default-executor-1] java.lang.Class - RPC ID: e5219940-7133-b6a4-97a6-9beabbc7573e 2024-06-03 09:12:44,796 [grpc-default-executor-1] java.lang.Class - Successfully committed replay with CommitRequestId: bd557942-4fd1-4699-be18-70db3054b33c with process time: 1717431164709 2024-06-03 09:12:44,796 [main] java.lang.Class - Subscription Active. Received a total of 5 events. ``` 7. To disconnect the `ManagedSubscribe` client, press Ctrl+C in the terminal. 8. Publish more events while the subscriber is disconnected. a. To differentiate the events that are published next from the previous events, modify the Order Number field values so that they’re higher than the previous numbers. In `java/src/main/java/utility/CommonContext.java`, in the `createEventMessages(Schema schema, final int numEvents)` method, replace the `orderNumbers` array with this statement. ``` String[] orderNumbers = {"104","105","106","107","108"}; ``` b. Because you modified the publishing code, rebuild the Pub/Sub API Java client. From the top-level java folder, enter `mvn clean install`. c. In the `PublishStream` client terminal, run the client by entering `./run.sh genericpubsub.PublishStream`. d. Verify that all five new events are published. 9. Return to the `ManagedSubscribe` client terminal. Restart the managed subscription by entering `./run.sh genericpubsub.ManagedSubscribe`. 10. The subscription resumes after the last received event, starting with the missed events. The `ManagedSubscribe` client output shows all the remaining events received with order numbers `104` through `108`. Also, the log output shows that a commit request is sent after the batch of events is received to commit the Replay ID. ``` 2024-06-03 09:14:09,931 [main] java.lang.Class - Using grpcHost api.pubsub.salesforce.com and grpcPort 7443 2024-06-03 09:14:10,487 [main] java.lang.Class - Starting managed subscription with developer name My_Managed_Subscription 2024-06-03 09:14:10,663 [grpc-nio-worker-ELG-1-2] utility.APISessionCredentials - Client Trace Id for current request: 22fdf915-3846-49b6-9943-09aa95b50903 2024-06-03 09:14:11,452 [grpc-default-executor-1] java.lang.Class - ManagedFetchResponse batch of 5 events pending requested: 0 2024-06-03 09:14:11,453 [grpc-default-executor-1] java.lang.Class - RPC ID: 579112d8-c727-b880-94a1-1eceb6f72882 2024-06-03 09:14:11,453 [grpc-default-executor-1] java.lang.Class - processEvent - EventID: 160810f4-f73f-4071-ad7f-a869b3e46dd8 SchemaId: _CADE3qaegX3ECEW0rlbYA 2024-06-03 09:14:11,461 [grpc-default-executor-1] utility.APISessionCredentials - Client Trace Id for current request: e734fa7b-dd28-4f80-bfc8-94c06df82f25 2024-06-03 09:14:11,576 [grpc-default-executor-1] java.lang.Class - Received event: { "CreatedDate": 1717431244798, "CreatedById": "005SG000009B3S1YAK", "Order_Number__c": "104", "City__c": "Los Angeles", "Amount__c": 35.0 } 2024-06-03 09:14:11,576 [grpc-default-executor-1] java.lang.Class - processEvent - EventID: b13516f1-7c40-4b7f-a57f-cdb2cc936579 SchemaId: _CADE3qaegX3ECEW0rlbYA 2024-06-03 09:14:11,576 [grpc-default-executor-1] java.lang.Class - Received event: { "CreatedDate": 1717431244798, "CreatedById": "005SG000009B3S1YAK", "Order_Number__c": "105", "City__c": "New York", "Amount__c": 20.0 } 2024-06-03 09:14:11,576 [grpc-default-executor-1] java.lang.Class - processEvent - EventID: 24a4be7c-441c-4c42-886d-09e5de87c0c3 SchemaId: _CADE3qaegX3ECEW0rlbYA 2024-06-03 09:14:11,576 [grpc-default-executor-1] java.lang.Class - Received event: { "CreatedDate": 1717431244798, "CreatedById": "005SG000009B3S1YAK", "Order_Number__c": "106", "City__c": "San Francisco", "Amount__c": 2.0 } 2024-06-03 09:14:11,576 [grpc-default-executor-1] java.lang.Class - processEvent - EventID: 302f4a46-f970-4341-a1c5-63d36070cd36 SchemaId: _CADE3qaegX3ECEW0rlbYA 2024-06-03 09:14:11,576 [grpc-default-executor-1] java.lang.Class - Received event: { "CreatedDate": 1717431244798, "CreatedById": "005SG000009B3S1YAK", "Order_Number__c": "107", "City__c": "San Jose", "Amount__c": 123.0 } 2024-06-03 09:14:11,576 [grpc-default-executor-1] java.lang.Class - processEvent - EventID: a5d081d5-5678-4fdc-b3c9-3a52b61fc04a SchemaId: _CADE3qaegX3ECEW0rlbYA 2024-06-03 09:14:11,576 [grpc-default-executor-1] java.lang.Class - Received event: { "CreatedDate": 1717431244798, "CreatedById": "005SG000009B3S1YAK", "Order_Number__c": "108", "City__c": "Boston", "Amount__c": 180.0 } 2024-06-03 09:14:11,576 [grpc-default-executor-1] java.lang.Class - Processed batch of 5 event(s) 2024-06-03 09:14:11,577 [grpc-default-executor-1] java.lang.Class - Sending CommitRequest with CommitReplayRequest ID: 7cf61ac5-cb17-49ae-975e-4f236f194945 2024-06-03 09:14:11,577 [grpc-default-executor-1] java.lang.Class - Fetching more events: 5 2024-06-03 09:14:11,578 [main] java.lang.Class - Subscription Active. Received a total of 5 events. 2024-06-03 09:14:11,638 [grpc-default-executor-1] java.lang.Class - ManagedFetchResponse batch of 0 events pending requested: 5 2024-06-03 09:14:11,638 [grpc-default-executor-1] java.lang.Class - RPC ID: 579112d8-c727-b880-94a1-1eceb6f72882 2024-06-03 09:14:11,638 [grpc-default-executor-1] java.lang.Class - Successfully committed replay with CommitRequestId: 7cf61ac5-cb17-49ae-975e-4f236f194945 with process time: 1717431251600 2024-06-03 09:14:11,638 [main] java.lang.Class - Subscription Active. Received a total of 5 events. ``` Thanks to managed subscriptions (beta), you can track the events processed in your subscriptions. You don’t worry about how to catch up on missed events after a subscriber client is disconnected. # Step 3: Set Up Events To subscribe to change data capture events on the standard channel, select the objects that you want to receive events for. 1. In Setup, search for `Change Data Capture`, and then select **Change Data Capture**. 2. On the Change Data Capture page, select the object. For example, for this quick start, select the Opportunity object. 3. Save your work. To subscribe to a custom channel, ensure that the custom channel is created first. After the channel is created, the entities are selected as part of the custom channel creation. See the [_Change Data Capture Developer Guide_](https://developer.salesforce.com/docs/atlas.en-us.change_data_capture.meta/change_data_capture/cdc_intro.htm). To subscribe to a custom platform event, define a custom platform event on the Platform Events page in Setup. To subscribe to a standard platform event, including real-time event monitoring events, you can view the available events in the [Standard Platform Event Object List](https://developer.salesforce.com/docs/atlas.en-us.platform_events.meta/platform_events/platform_events_objects_list.htm) in the [_Platform Events Developer Guide_](https://developer.salesforce.com/docs/atlas.en-us.platform_events.meta/platform_events/platform_events_intro.htm). # Step 4: Publish and Subscribe to Platform Events Use the Java examples for the `Subscribe` RPC and `PublishStream` RPC to try out publishing and receiving events. **Prerequisites** - If you’re using a Developer Edition org and you set up a namespace prefix, modify the Java code that builds the event message to prepend the namespace prefix to the platform event fields. In `java/src/main/java/utility/CommonContext.java`, modify the `createEventMessage` and `createEventMessages` methods to include the namespace prefix in the field names. 1. In a Terminal window, navigate to the top-level `java` folder. 2. To run the Subscribe RPC example, enter: `./run.sh genericpubsub.Subscribe` 3. The Subscribe example prints out logging info on the terminal, similar to: ``` 2024-04-09 11:56:35,443 [main] java.lang.Class - Using grpcHost api.pubsub.salesforce.com and grpcPort 7443 2024-04-09 11:56:36,610 [main] utility.APISessionCredentials - Client Trace Id for current request: 161d4754-cd94-446e-8c41-77626805dd7c 2024-04-09 11:56:37,126 [main] java.lang.Class - Subscription started for topic: /event/Order_Event__e. 2024-04-09 11:56:37,128 [main] utility.APISessionCredentials - Client Trace Id for current request: aeecdf2c-9522-4d1d-a4ba-359f137cf53e 2024-04-09 11:56:42,140 [main] java.lang.Class - Subscription Active. Received a total of 0 events. ``` Because no events were published, the count of received events is zero. 4. Before you can publish events, modify the code to specify the event fields and values to publish. In this quick start, you modify the code to supply the user ID for the `CreatedById` field in the event message. You use the sample values for the other event fields. In a text editor or an IDE, such as Visual Studio Code, open the `java/src/main/java/utility/CommonContext.java` from your `pub-sub-api` folder. 5. There are several `createEventMessage` methods and one `createEventMessages` method that build the event messages. This method is used in `Publish.java`. ``` // Used by Publish.java public GenericRecord createEventMessage(Schema schema) ``` These methods are used in `PublishStream.java`. In this quick start, we use `PublishStream.java` with the default argument value of `SINGLE_PUBLISH_REQUEST` of `true`, so the `createEventMessages` method is used. ``` // Used by PublishStream.java when SINGLE_PUBLISH_REQUEST is false public GenericRecord createEventMessage(Schema schema, final int counter) // Used by PublishStream.java when SINGLE_PUBLISH_REQUEST is true public List createEventMessages(Schema schema, final int numEvents) ``` Provide a valid `CreatedById` value for the published event message in this method. Replace the `` placeholder value of the `CreatedById` field with a valid user ID in your Salesforce org. To get the user ID, see [Find the Salesforce ID for a User or Profile](https://help.salesforce.com/s/articleView?id=000381643&type=1). ``` public List createEventMessages(Schema schema, final int numEvents) { String[] orderNumbers = {"99","100","101","102","103"}; String[] cities = {"Los Angeles", "New York", "San Francisco", "San Jose", "Boston"}; Double[] amounts = {35.0, 20.0, 2.0, 123.0, 180.0}; // Update CreatedById with the appropriate User Id from your org. List events = new ArrayList<>(); for (int i=0; i") .set("Order_Number__c", orderNumbers[i % 5]) .set("City__c", cities[i % 5]) .set("Amount__c", amounts[i % 5]).build()); } return events; } ``` 6. Optional. If you want to supply other event fields and values, make additional changes. 7. To publish all events in one publish request using the `createEventMessages` method, in `java/src/main/resources/arguments.yaml`, set `SINGLE_PUBLISH_REQUEST` to true: `SINGLE_PUBLISH_REQUEST: true` 8. In a new terminal window, navigate to the top-level `java` folder in your `pub-sub-api` folder. 9. To build the Java client, enter: `mvn clean install` 10. To run the `PublishStream` RPC example, enter: `./run.sh genericpubsub.PublishStream` The example publishes five events as set in the `arguments.yaml` file’s default configuration. It prints out logging information to the terminal. 11. Return to the terminal for the Subscribe example. The output shows the events received. ``` 2024-04-09 11:56:51,302 [pool-3-thread-1] java.lang.Class - Received event with payload: { "CreatedDate": 1712689010574, "CreatedById": "0055f000005mc66AAA", "Order_Number__c": "99", "City__c": "Los Angeles", "Amount__c": 35.0 } with schema name: Order_Event__e 2024-04-09 11:56:51,302 [pool-3-thread-2] java.lang.Class - Received event with payload: { "CreatedDate": 1712689010574, "CreatedById": "0055f000005mc66AAA", "Order_Number__c": "100", "City__c": "New York", "Amount__c": 20.0 } with schema name: Order_Event__e 2024-04-09 11:56:51,302 [pool-3-thread-3] java.lang.Class - Received event with payload: { "CreatedDate": 1712689010574, "CreatedById": "0055f000005mc66AAA", "Order_Number__c": "101", "City__c": "San Francisco", "Amount__c": 2.0 } with schema name: Order_Event__e 2024-04-09 11:56:51,303 [pool-3-thread-2] java.lang.Class - Received event with payload: { "CreatedDate": 1712689010574, "CreatedById": "0055f000005mc66AAA", "Order_Number__c": "102", "City__c": "San Jose", "Amount__c": 123.0 } with schema name: Order_Event__e 2024-04-09 11:56:51,303 [pool-3-thread-1] java.lang.Class - Received event with payload: { "CreatedDate": 1712689010574, "CreatedById": "0055f000005mc66AAA", "Order_Number__c": "103", "City__c": "Boston", "Amount__c": 180.0 } with schema name: Order_Event__e ``` ## Learn More To learn more about the example code, see [Implementation](https://github.com/forcedotcom/pub-sub-api/tree/main/java#implementation) and [Limitations](https://github.com/forcedotcom/pub-sub-api/tree/main/java#limitations) in the Readme page of the `pub-sub-api` GitHub repository. The client code for the Pub/Sub API RPC methods is in the [genericpubsub](https://github.com/forcedotcom/pub-sub-api/tree/main/java/src/main/java/genericpubsub) folder. To check out the implementation for each RPC method, open the corresponding Java file in the `genericpubsub` folder. For the `PublishStream` example, open [PublishStream.java](https://github.com/forcedotcom/pub-sub-api/blob/main/java/src/main/java/genericpubsub/PublishStream.java). And for the `Subscribe` example, open [Subscribe.java](https://github.com/forcedotcom/pub-sub-api/blob/main/java/src/main/java/genericpubsub/Subscribe.java). For more information about the `PublishStream` and `Subscribe` RPC methods, see [PublishStream RPC Method](https://developer.salesforce.com/docs/platform/pub-sub-api/references/methods/publishstream-rpc.html) and [Subscribe RPC Method](https://developer.salesforce.com/docs/platform/pub-sub-api/references/methods/subscribe-rpc.html). To run other RPC methods, use this command format: `./run.sh .` # Step 4: Write Code That Subscribes to an Event Channel Follow instructions in this step to build a subscriber client. **note:** When you copy the code snippets in this page, ensure that the indentation starts from the comment block below the `with grpc.secure_channel` statement that you added in [Step 2: Build the Python Client](https://developer.salesforce.com/docs/platform/pub-sub-api/guide/qs-build-python-client.md). In Python, code blocks are delimited by their indentation. 1. Get the topic name that you want to subscribe to. The topic format is: - For a custom platform event: `/event/EventName__e` - For a standard platform event: `/event/EventName` - For a change data capture channel that captures events for all selected entities: `/data/ChangeEvents` - For a change data capture single-entity channel for a standard object: `/data/{StandardObjectName}ChangeEvent` - For a change data capture single-entity channel for a custom object: `/data/{CustomObjectName}__ChangeEvent` - For a change data capture custom channel: `/data/CustomChannelName__chn` - For a custom platform event channel: `/event/CustomChannelName__chn` 2. Create a generator function to make a FetchRequest stream. After the semaphore is released in the client, the `fetchReqStream` function acquires a lock and sends a FetchRequest. In this FetchRequest, `num_requested` is the number of events that the client requests from the server. The specified `num_requested` of events can be sent in one or more FetchResponses to the client, with each FetchResponse containing a batch of events. In this case, we set it to 1. ```python def fetchReqStream(topic): while True: semaphore.acquire() yield pb2.FetchRequest( topic_name = topic, replay_preset = pb2.ReplayPreset.LATEST, num_requested = 1) ``` 3. Create a decoding function to decode the payloads of received event messages. ```python def decode(schema, payload): schema = avro.schema.parse(schema) buf = io.BytesIO(payload) decoder = avro.io.BinaryDecoder(buf) reader = avro.io.DatumReader(schema) ret = reader.read(decoder) return ret ``` 4. Make the subscribe call, and handle received event messages. Decode the payloads of the events with your decoding function. Store the latest replay ID received. You can use the replay ID later to restart a subscription after the last consumed event, if necessary. For more information, see [Replaying an Event Stream](https://developer.salesforce.com/docs/platform/pub-sub-api/references/methods/subscribe-rpc.html#replaying-an-event-stream). ```python mysubtopic = "/data/OpportunityChangeEvent" print('Subscribing to ' + mysubtopic) substream = stub.Subscribe(fetchReqStream(mysubtopic), metadata=authmetadata) for event in substream: if event.events: semaphore.release() print("Number of events received: ", len(event.events)) payloadbytes = event.events[0].event.payload schemaid = event.events[0].event.schema_id schema = stub.GetSchema( pb2.SchemaRequest(schema_id=schemaid), metadata=authmetadata).schema_json decoded = decode(schema, payloadbytes) print("Got an event!", json.dumps(decoded)) else: print("[", time.strftime('%b %d, %Y %l:%M%p %Z'), "] The subscription is active.") latest_replay_id = event.latest_replay_id ``` **note:** In this example, the `event` variable represents a FetchResponse. The code example processes one event at a time. When a FetchResponse containing an event is received, the semaphore is released, which results in the `fetchReqStream` function locking the semaphore and sending a new FetchRequest. The new FetchRequest requests one event in `num_requested`. This flow control works only when requesting and processing one event at a time. For more information about flow control, see [Pull Subscription and Flow Control](https://developer.salesforce.com/docs/platform/pub-sub-api/guide/flow-control.md). For more information about semaphores, see [Semaphore Objects](https://docs.python.org/3/library/threading.html#semaphore-objects) in the Python documentation. If you run your code at this point, you don’t receive any event messages unless you or Salesforce publishes an event message. For change data capture events, make sure that Change Data Capture is tracking the object by selecting the object in Setup, on the Change Data Capture page. For this example, select the Opportunity object. Then make a change to an opportunity so that Salesforce generates an event message. Salesforce publishes most standard platform events, including real-time event monitoring events, in response to an action in Salesforce. You can publish only the standard events that support the `create()` call. For more information, see [Standard Platform Event Object List](https://developer.salesforce.com/docs/atlas.en-us.platform_events.meta/platform_events/platform_events_objects_list.htm) in the [_Platform Events Developer Guide_](https://developer.salesforce.com/docs/atlas.en-us.platform_events.meta/platform_events/platform_events_intro.htm). ## Received Event Example If you subscribe to the `/data/OpportunityChangeEvent` topic and make a change to an opportunity, you receive a change event similar to this event message. This event message is for an opportunity whose `Amount` field was changed and the `Type` field was cleared (set to null). **note:** `ChangeEventHeader` contains bitmap fields that you must decode before reading their contents. These fields are: `nulledFields`, `changedFields`, and `diffFields`. For more information, see [Event Deserialization Considerations](https://developer.salesforce.com/docs/platform/pub-sub-api/guide/event-deserialization-considerations.md). ```json { "ChangeEventHeader": { "entityName": "Opportunity", "recordIds": ["006R0000002FyVgIAK"], "changeType": "UPDATE", "changeOrigin": "com/salesforce/api/soap/55.0;client=SfdcInternalAPI/", "transactionKey": "00006230-d3c7-f430-989e-f686e0b451bf", "sequenceNumber": 1, "commitTimestamp": 1656432588000, "commitNumber": 77385917926, "commitUser": "005R0000000Iu3SIAS", "nulledFields": ["0x0800"], "diffFields": [], "changedFields": ["0x21000840"] }, "AccountId": null, "IsPrivate": null, "Name": null, "Description": null, "StageName": null, "Amount": 110000, "Probability": null, "ExpectedRevenue": null, "TotalOpportunityQuantity": null, "CloseDate": null, "Type": null, "NextStep": null, "LeadSource": null, "IsClosed": null, "IsWon": null, "ForecastCategory": null, "ForecastCategoryName": null, "CampaignId": null, "HasOpportunityLineItem": null, "Pricebook2Id": null, "OwnerId": null, "CreatedDate": null, "CreatedById": null, "LastModifiedDate": 1656432588000, "LastModifiedById": null, "LastStageChangeDate": null, "ContactId": null, "ContractId": null, "LastAmountChangedHistoryId": "008R0000003RnLUIA0", "LastCloseDateChangedHistoryId": null } ``` You can also publish a custom platform event. The next step shows you how to do that using the Pub/Sub API in Python. **See Also** - [Subscription Channels](https://developer.salesforce.com/docs/atlas.en-us.change_data_capture.meta/change_data_capture/cdc_subscribe_channels.htm) - [Group Platform Events into One Stream with a Custom Channel](https://developer.salesforce.com/docs/atlas.en-us.platform_events.meta/platform_events/platform_events_subscribe_custom_channels.htm) # Step 5: Subscribe to Change Events Subscribe to change events and perform bitmap field expansion. By setting a flag in `arguments.yaml`, the `Subscribe` example expands the `changedFields` bitmap fields in change events into its component fields. Change event headers are described in [ChangeEventHeader Fields](https://developer.salesforce.com/docs/atlas.en-us.change_data_capture.meta/change_data_capture/cdc_event_fields_header.htm) in the _Change Data Capture Developer Guide_. To learn more about bitmap field expansion, see [Event Deserialization Considerations](https://developer.salesforce.com/docs/platform/pub-sub-api/guide/event-deserialization-considerations.md). **Prerequisites** - Because this example subscribes to changes for Opportunity, select Opportunity for change data capture in Setup. For more information, see [_Select Objects for Change Notifications in the User Interface_](https://developer.salesforce.com/docs/atlas.en-us.change_data_capture.meta/change_data_capture/cdc_select_objects.htm) in the _Change Data Capture Developer Guide_. 1. In your `pub-sub-api` folder, go to `java/src/main/resources` and open the `arguments.yaml` file. - Provide the change event channel: `TOPIC: /data/OpportunityChangeEvent` - Enable bitmap field expansion for ChangeEventHeader: `PROCESS_CHANGE_EVENT_HEADER_FIELDS: true`. 2. In a Terminal window, navigate to the `java` folder. 3. To run the example, from the top-level `java` folder, enter: `./run.sh genericpubsub.Subscribe` The example subscribes to the `/data/OpportunityChangeEvent` channel and captures any new or changed Opportunity records. 4. In Salesforce, create an opportunity record. The change event is received and printed. For example: ``` { "ChangeEventHeader": { "entityName": "Opportunity", "recordIds": [ "006ZM000001dwBYYAY" ], "changeType": "CREATE", "changeOrigin": "com/salesforce/api/soap/61.0;client=SfdcInternalAPI/", "transactionKey": "00007d13-afe8-458d-370b-72f604213dd3", "sequenceNumber": 1, "commitTimestamp": 1722536398000, "commitNumber": 77395491487, "commitUser": "005ZM000000M6o1YAC", "nulledFields": [], "diffFields": [], "changedFields": [] }, "AccountId": "001ZM000001QUlLYAW", "IsPrivate": null, "Name": "My Opportunity", "Description": null, "StageName": "Prospecting", "Amount": 10000, "Probability": 10, "ExpectedRevenue": null, "TotalOpportunityQuantity": null, "CloseDate": 1725235200000, "Type": null, "NextStep": null, "LeadSource": null, "IsClosed": false, "IsWon": false, "ForecastCategory": "Pipeline", "ForecastCategoryName": "Pipeline", "CampaignId": null, "HasOpportunityLineItem": false, "Pricebook2Id": null, "OwnerId": "005ZM000000M6o1YAC", "CreatedDate": 1722536397000, "CreatedById": "005ZM000000M6o1YAC", "LastModifiedDate": 1722536397000, "LastModifiedById": "005ZM000000M6o1YAC", "LastStageChangeDate": null, "ContactId": null, "ContractId": null, "LastAmountChangedHistoryId": null, "LastCloseDateChangedHistoryId": null } ``` 5. Update the opportunity record. For example, select a `Type` of `New Customer`. The change event for the update is received and printed. The output includes the Changed Fields section, which contains the fields that were expanded from the `changedFields` bitmap field. It lists the two fields that were changed: the `Type` field that you changed, and the `LastModifiedDate` field that the system changed. For example: ``` { "ChangeEventHeader": { "entityName": "Opportunity", "recordIds": [ "006ZM000001dwBYYAY" ], "changeType": "UPDATE", "changeOrigin": "com/salesforce/api/soap/61.0;client=SfdcInternalAPI/", "transactionKey": "00007cd5-988e-e3db-3adc-158958207197", "sequenceNumber": 1, "commitTimestamp": 1722536439000, "commitNumber": 77395586403, "commitUser": "005ZM000000M6o1YAC", "nulledFields": [], "diffFields": [], "changedFields": [ "0x01000800" ] }, "AccountId": null, "IsPrivate": null, "Name": null, "Description": null, "StageName": null, "Amount": null, "Probability": null, "ExpectedRevenue": null, "TotalOpportunityQuantity": null, "CloseDate": null, "Type": "New Customer", "NextStep": null, "LeadSource": null, "IsClosed": null, "IsWon": null, "ForecastCategory": null, "ForecastCategoryName": null, "CampaignId": null, "HasOpportunityLineItem": null, "Pricebook2Id": null, "OwnerId": null, "CreatedDate": null, "CreatedById": null, "LastModifiedDate": 1722536439000, "LastModifiedById": null, "LastStageChangeDate": null, "ContactId": null, "ContractId": null, "LastAmountChangedHistoryId": null, "LastCloseDateChangedHistoryId": null } 2024-08-01 11:20:50,888 [grpc-default-executor-3] java.lang.Class - ============================ 2024-08-01 11:20:50,889 [grpc-default-executor-3] java.lang.Class - ChangedFields 2024-08-01 11:20:50,889 [grpc-default-executor-3] java.lang.Class - ============================ 2024-08-01 11:20:50,889 [grpc-default-executor-3] java.lang.Class - Type 2024-08-01 11:20:50,889 [grpc-default-executor-3] java.lang.Class - LastModifiedDate 2024-08-01 11:20:50,889 [grpc-default-executor-3] java.lang.Class - ============================ ``` ## Learn More The client code for the Pub/Sub API RPC methods is in the [genericpubsub](https://github.com/forcedotcom/pub-sub-api/tree/main/java/src/main/java/genericpubsub) folder. To check out the implementation for each RPC method, open the corresponding Java file in the `genericpubsub` folder. For the `Subscribe` example, open [Subscribe.java](https://github.com/forcedotcom/pub-sub-api/blob/main/java/src/main/java/genericpubsub/Subscribe.java). See [Subscribe RPC Method](https://developer.salesforce.com/docs/platform/pub-sub-api/references/methods/subscribe-rpc.html). **See Also** - _Change Data Capture Developer Guide_: [ChangeEventHeader Fields](https://developer.salesforce.com/docs/atlas.en-us.change_data_capture.meta/change_data_capture/cdc_event_fields_header.htm) - [Event Deserialization Considerations](https://developer.salesforce.com/docs/platform/pub-sub-api/guide/event-deserialization-considerations.md) # Step 5: Write Code That Publishes a Platform Event Message Follow instructions in this step to build a publisher client. **note:** When you copy the code snippets in this page, ensure that the indentation starts from the comment block below the `with grpc.secure_channel` statement that you added in [Step 2: Build the Python Client](https://developer.salesforce.com/docs/platform/pub-sub-api/guide/qs-build-python-client.md). In Python, code blocks are delimited by their indentation. Before you publish a custom platform event message, ensure that the platform event is defined in your org. You can view defined platform events in Setup on the Platform Events page. For example, this image shows the definition of `Order_Event__e`. This event has two fields: `Order_Number__c` of type Text and `Has_Shipped__c` of type Checkbox. ![The Order Event detail page](https://developer.salesforce.com/docs/media/pubsub-order-event-definition.png "The Order Event detail page") 1. Create a separate Python file, `publish.py`, for publishing so that you can run the publisher and subscriber clients independently. Include common code from the subscriber client that you created earlier. Include the import statements, code for creating the channel, the authentication code to build authmetadata, and the stub. Also, add this import statement: `from datetime import datetime, timedelta` 2. Get the topic name for the event that you want to publish. The topic format is `/event/EventName__e`. For example, for `Order_Event__e` the topic is `/event/Order_Event__e`. You use it in the next step. 3. Replace the `mypubtopic` placeholder value with your topic. Get the schema ID and schema for the event. To get the schema ID, call the `GetTopic` method and pass the topic name. Next, pass the schema ID to the `GetSchema` method, which returns the schema. Typically, you store the schema and reuse it if the schema ID hasn’t changed, but this example doesn’t do that for simplicity. ```python mypubtopic = '{your publish topic}' schemaid = stub.GetTopic(pb2.TopicRequest(topic_name=mypubtopic), metadata=authmetadata).schema_id # Typically, you store the schema and reuse it if the schema ID hasn't changed, # but this example doesn't do that for simplicity. schema = stub.GetSchema(pb2.SchemaRequest(schema_id=schemaid), metadata=authmetadata).schema_json ``` 4. Create a function to encode the information that you want to send by using the schema. ```python def encode(schema, payload): schema = avro.schema.parse(schema) buf = io.BytesIO() encoder = avro.io.BinaryEncoder(buf) writer = avro.io.DatumWriter(schema) writer.write(payload, encoder) return buf.getvalue() ``` 5. Create a function that creates a `PublishRequest`. Construct the payload by adding the event fields and values in the payload variable. Populate the values of the required system fields: `CreatedDate` and `CreatedById`. The `CreatedById` value isn’t validated. For `{event field}: {field value}`, list the event fields and values. For example, for Order_Event\_\_e, you can add: ```python "Order_Number__c": "100", "Has_Shipped__c": True ``` The `req` variable contains the encoded payload, which is returned by the encode function. It also contains the schema ID. ```python def makePublishRequest(schemaid): dt = datetime.now() + timedelta(days=5) payload = { "CreatedDate": int(datetime.now().timestamp()), "CreatedById": '005R...', #Your user ID "{replace event field}": "{replace field value}" } req = { "schema_id": schemaid, "payload": encode(schema, payload) } return [req] ``` 6. Make the publish call, and handle any acknowledgements that you get back. ```python publishresponse = stub.Publish(pb2.PublishRequest( topic_name=mypubtopic, events= makePublishRequest(schemaid)), metadata=authmetadata) ``` 7. Modify your subscriber client, `PubSubAPIClient.py`, to susbribe to the platform event. ```python mysubtopic = "/event/Order_Event__e" ``` 8. Run the client by entering this command on the command line. ``` python3 publish.py ``` If the publish request is successful, you receive a PublishResponse message containing the replay ID. The subscriber client receives and displays the event message. If the publish request wasn’t successful, you get an error similar to: ``` results { error { code: PUBLISH msg: "com.salesforce.eventbus.exceptions.PublishException: Unsupported topic [/event/Tracker_Event__e]. Standard Volume event type is not supported." } } schema_id: "AKTsT5i0mDe_UF8qnC8Aig" ``` # Subscribe RPC Method The `Subscribe` method uses bidirectional streaming. It’s pull-based, which means that it requests new events. This subscription model is in contrast to push-based subscription, in which the subscriber is a listener that waits for events to be sent. The Subscribe method offers flow control, which enables you to specify how many events to receive in a call based on the client’s event processing speed. `rpc Subscribe (stream FetchRequest) returns (stream FetchResponse);` A client can request the number of events it can process by setting the `num_requested` field in FetchRequest. The maximum value for `num_requested` is 100. The server returns the events in one or more FetchResponses. Each FetchResponse contains the number of events that are pending delivery in the `pending_num_requested` field. If a client requests more events before the server finishes the last requested amount, the server appends the new amount to the current batch of events it still needs to fetch and deliver. For more information about the pull subscription behavior, see [Pull Subscription and Flow Control](https://developer.salesforce.com/docs/platform/pub-sub-api/guide/flow-control.html). For more information about the fields in FetchRequest and FetchResponse, see the [Pub/Sub API proto file](https://github.com/forcedotcom/pub-sub-api/blob/main/pubsub_api.proto). ## Subscribe Keepalive Behavior The Subscribe call stream stays open as long as there are pending events to be sent to the client. The number of pending events is based on the number of events the client requested in FetchRequests and the number of events delivered in FetchResponses. This table summarizes how to keep a managed subscription stream open for each condition. | Condition | To keep a subscription stream alive | | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | Pending events to be sent to the client and there are still events in the event bus to be delivered (`pending_num_requested` in ManagedFetchResponse is greater than zero) | The stream stays open. | | No more pending events to be sent to the client (`pending_num_requested` in ManagedFetchResponse is equal to zero) | The client must send a new FetchRequest within 60 seconds from when the last FetchResponse was sent. | | Pending events to be sent to the client but no new events are available in the event bus to be delivered (`pending_num_requested` in ManagedFetchResponse is greater than zero) | The server keeps the stream open by sending keepalive messages within 270 seconds. A keepalive message indicates that the Pub/Sub API service is healthy and provides a new Replay ID that points to an advanced position in the stream. | ## Keepalive Messages Sent from the Server If there are pending events to be sent to the client (`pending_num_requested` in FetchResponse is greater than zero) but no new events are available in the event bus, the server keeps the stream alive by sending keepalive messages within 270 seconds. The keepalive message is an empty batch FetchResponse with the latest `replay_id`. The returned `replay_id` can be the last event consumed in your client or an advanced position in the stream beyond the last consumed event. An advanced position in the stream is possible even if you didn’t receive events because the event bus combines streams from multiple orgs. No org-specific information or personally identifiable information is shared with other orgs. For best performance results, we recommend that you save the `replay_id` from each empty FetchResponse and use it when resubscribing. That way, when you restart the subscription, you receive the next unprocessed events and not duplicate events. Saving the `replay_id` from each empty FetchResponse ensures that the `replay_id` is recent. An old `replay_id` refers to an earlier position in the stream than the `replay_id` that the keepalive message returns. Using the more recent `replay_id` ensures that the server doesn’t take too much time to seek the event in the event bus when the subscription restarts. ## Handling Errors If an error occurs, the long-lived connection is stopped. Retry the `Subscribe` RPC method call to create a new long-lived connection. For more information, see [Handle Errors](/docs/platform/pub-sub-api/guide/handling-errors.html). ## Replaying an Event Stream Platform events and change data capture events are retained in the event bus for 3 days. A client can subscribe at any position in the stream by providing a replay option in the first FetchRequest. Any subsequent FetchRequests with a new replay option are ignored. A client must call the `Subscribe` RPC again to restart the subscription at a new position in the stream. The replay option consists of a combination of `replay_preset` and `replay_id` values in the first FetchRequest received from a client. This table describes the replay options and when to use each. | `replay_preset` | Behavior | When to Use | | --------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `LATEST` | Default if no replay preset is specified. Subscribe from the tip of the stream. | Use when you’re interested only in new event messages and don’t need the earlier event messages stored in the event bus. | | `CUSTOM` | Resubscribe after a specific `replay_id`. To use this option, also set the `replay_id` to the replay ID of the last keepalive message or the last processed event message, whichever was received last. Store replay ID values as bytes because replay IDs are opaque and not guaranteed to always be numbers. | Use to catch up on missed events after a certain event message, for example, after a connection failure. | | `EARLIEST` | Subscribe from the earliest retained events. | Use to catch up on missed events and retrieve all stored events. Use this option after a client has been disconnected for more than 3 days and the last saved replay ID is no longer valid. Use `EARLIEST` sparingly because it can slow performance when retrieving a large number of events. | **tip:** Instead of using the `replay_preset` options to retrieve missed events from Salesforce, you can use the [ManagedSubscribe RPC Method (Beta)](https://developer.salesforce.com/docs/platform/pub-sub-api/references/methods/managedsubscribe-rpc.md). With the `ManagedSubscribe` method, you can track the Replay ID of the last processed event by committing the Replay ID on the server. You don’t worry about managing a replay ID store on the client. Then when resubscribing, the subscription resumes after the stored Replay ID. The first FetchRequest of the stream identifies the topic to subscribe to. If a subsequent FetchRequest provides `topic_name`, it must match what was provided in the first FetchRequest. Otherwise, the RPC sends an error with an `INVALID_ARGUMENT` status. **See Also** - [Java Quick Start for Publishing and Subscribing to Events](/docs/platform/pub-sub-api/guide/qs-java-quick-start.html) - [Allocations](/docs/platform/pub-sub-api/guide/allocations.html) - [Pull Subscription and Flow Control](/docs/platform/pub-sub-api/guide/flow-control.html) - [ManagedSubscribe RPC Method](https://developer.salesforce.com/docs/platform/pub-sub-api/references/methods/managedsubscribe-rpc.md) # Use Pub/Sub API Learn about the features of Pub/Sub API, including pull subscription, error handling, considerations, and allocations for Pub/Sub API and events.