Merge remote-tracking branch 'upstream/main'

Author: Craig Meyer

.gitignore

@@ -14,4 +14,8 @@ $RECYCLE.BIN/
 # Dependency directory for Go examples
 go/vendor/
 
-*/target/
+*/target/
+
+# Python compiled files
+*.pyo
+*.pyc

README.md

@@ -3,7 +3,7 @@
 - [About Pub/Sub API](#about-pubsub-api)
 - [gRPC](#grpc)
 - [Documentation and Blog Posts](#documentation-and-blog-post)
-- [Code Samples](#code-samples)
+- [Code Samples](#code-samples-from-salesforce)
 
 ## About Pub/Sub API
 Welcome to 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](https://grpc.io/docs/what-is-grpc/introduction/) and HTTP/2, Pub/Sub API enables efficient delivery of binary event messages in the Apache Avro format.
@@ -23,24 +23,32 @@ officially supported languages have well-supported Avro libraries:
 |C++|[Apache Avro C++](https://avro.apache.org/docs/current/api/cpp/html/index.html)|
 |Dart|[avro-dart](https://github.com/sqs/avro-dart) (last updated 2012)|
 |Go|[goavro](https://github.com/linkedin/goavro)|
-|Java|[Apache Avro Java](https://avro.apache.org/docs/1.10.2/gettingstartedjava.html)|
+|Java|[Apache Avro Java](https://avro.apache.org/docs/current/getting-started-java/)|
 |Kotlin|[avro4k](https://github.com/avro-kotlin/avro4k)|
 |Node|[avro-js](https://www.npmjs.com/package/avro-js)|
 |Objective C|[ObjectiveAvro](https://github.com/jlawton/ObjectiveAvro) (but read [this](https://stackoverflow.com/questions/57216446/data-serialisation-in-objective-c-avro-alternative))|
 |PHP|[avro-php](https://github.com/wikimedia/avro-php)|
-|Python|[Apache Avro Python](https://avro.apache.org/docs/current/gettingstartedpython.html)|
+|Python|[Apache Avro Python](https://avro.apache.org/docs/current/getting-started-python/)|
 |Ruby|[AvroTurf](https://github.com/dasch/avro_turf)|
 
 ## Documentation, Blog Post and Videos
 - [Pub/Sub API Developer Guide](https://developer.salesforce.com/docs/platform/pub-sub-api/overview)
 - [Salesforce Architects Blog Post](https://medium.com/salesforce-architects/announcing-pub-sub-api-generally-available-3980c9eaf0b7)
 - [Introducing the New gRPC-based Pub Sub API YouTube Developer Quick Takes](https://youtu.be/g9P87_loVVA)
 
-## Code Samples
+## Code Samples from Salesforce
+Salesforce provides these samples for demonstration purposes. They aren't meant to be used in production code. Before you use these samples in production, make sure you perform thorough functional and performance testing.
+- [Java Quick Start in the Developer Guide](https://developer.salesforce.com/docs/platform/pub-sub-api/guide/qs-java-quick-start.html)
 - [Python Quick Start in the Developer Guide](https://developer.salesforce.com/docs/platform/pub-sub-api/guide/qs-python-quick-start.html)
 - [Python Code Examples](python/)
 - [Go Code Examples](go/)
 - [Java Code Examples](java/)
+
+## Code Samples from the Developer Community
+These examples are developed by the community. They aren't supported by Salesforce. Use at your own discretion.
 - [E-Bikes Sample Application](https://github.com/trailheadapps/ebikes-lwc)
 - [Pub/Sub API Node Client](https://github.com/pozil/pub-sub-api-node-client)
-  
+- [.NET Code Examples](https://github.com/Meyce/pub-sub-api/tree/main/.Net)
+- [Ruby Pub/Sub API Example](https://github.com/RenoFi/salesforce-pub-sub-rb)
+
+If you have a code sample for Pub/Sub API that you would like to add a link to in this section, submit a PR with the modified readme page. We don't guarantee that we can link to all samples. Priority will be given to samples implemented in a programming language that is not represented in this repository's samples.

java/README.md

@@ -1,7 +1,7 @@
 # Pub/Sub API Java Examples
 
 ## Overview
-This directory contains some simple Java Examples that can be used with the Pub/Sub API. These examples range from generic Publish and Subscribe, processing CustomEventHeaders in change events and also a specific example of updating the Salesforce Account standard object. It is important to note that these examples are not performance tested nor are they production ready. They are meant to be used as a learning resource or a starting point to understand the flows of each of the Remote Procedure Calls (RPCs) of Pub/Sub API. There are some limitations to these examples as well mentioned below.
+This directory contains some Java examples that can be used with the Pub/Sub API. These examples range from generic Publish, Subscribe, ManagedSubscribe (beta), processing CustomEventHeaders in change events, and a specific example of updating the Salesforce Account standard object. It is important to note that these examples are not performance tested nor are they production ready. They are meant to be used as a learning resource or a starting point to understand the flows of each of the Remote Procedure Calls (RPCs) of Pub/Sub API. There are some limitations to these examples as well mentioned below.
 
 ## Project Structure
 In the `src/main` directory of the project, you will find several sub-directories as follows:
@@ -20,14 +20,14 @@ In the `src/main` directory of the project, you will find several sub-directorie
 3. Run `mvn clean install` from the `java` directory to build the project and generate required sources from the proto file.
 4. The `arguments.yaml` file in the `src/main/resources` sub-directory contains a list of required and optional configurations needed to run the examples. The file contains detailed comments on how to set the configurations.
 5. Get the username, password, and login URL of the Salesforce org you wish to use.
-6. For the examples in `genericpubsub` package, a custom **_CarMaintenance_** [platform event](https://developer.salesforce.com/docs/atlas.en-us.platform_events.meta/platform_events/platform_events_define_ui.htm) has to be created in the Salesforce org. Ensure your CarMaintenance platform event matches the following structure:
+6. For the examples in `genericpubsub` package, a custom **_Order Event_** [platform event](https://developer.salesforce.com/docs/atlas.en-us.platform_events.meta/platform_events/platform_events_define_ui.htm) has to be created in the Salesforce org. Ensure your `Order Event` platform event matches the following structure:
    - Standard Fields
-       - Label: `CarMaintenance`
-       - Plural Label: `CarMaintenances`
+       - Label: `Order Event`
+       - Plural Label: `Order Events`
    - Custom Fields
-       - `Cost` (Number)
-       - `Mileage` (Number)
-       - `WorkDescription` (Text, 200)
+       - `Order Number` (Text, 18)
+       - `City` (Text, 50)
+       - `Amount` (Number, (16,2))
 7. For the examples in the `accountupdateapp` package, another custom **_NewAccount_** [platform event](https://developer.salesforce.com/docs/atlas.en-us.platform_events.meta/platform_events/platform_events_define_ui.htm) has to be created in the Salesforce org. [More info here](src/main/java/accountupdateapp/README.md).
 
 ### Execution
@@ -37,14 +37,20 @@ In the `src/main` directory of the project, you will find several sub-directorie
        * `PUBSUB_PORT`: Specify the Pub/Sub API port to be used (usually 7443).
        * `LOGIN_URL`: Specify the login url of the Salesforce org being used to run the examples.
        * `USERNAME` & `PASSWORD`: For authentication via username and password, you will need to specify the username and password of the Salesforce org. 
-       * `ACCESS_TOKEN` & `TENANT_ID`: For authentication via session token an tenant ID, you will need to specify the sessionToken and tenant ID of the Salesforce org.
+       * `ACCESS_TOKEN` & `TENANT_ID`: For authentication via session token and tenant ID, you will need to specify the sessionToken and tenant ID of the Salesforce org.
+       *  When using managed event subscriptions (beta), one of these configurations is required.
+          * `MANAGED_SUB_DEVELOPER_NAME`: Specify the developer name of ManagedEventSubscription. This parameter is used in ManagedSubscribe.java.
+          * `MANAGED_SUB_ID`: Specify the ID of the ManagedEventSubscription Tooling API record. This parameter is used in ManagedSubscribe.java.
+
    2. Optional Parameters:
        * `TOPIC`: Specify the topic for which you wish to publish/subscribe. 
        * `NUMBER_OF_EVENTS_TO_PUBLISH`: Specify the number of events to publish while using the PublishStream RPC.
-       * `NUMBER_OF_EVENTS_TO_SUBSCRIBE`: Specify the number of events to subscribe while using the Subscribe RPC.
+       * `SINGLE_PUBLISH_REQUEST`: Specify if you want to publish the events in a single or multiple PublishRequests.
+       * `NUMBER_OF_EVENTS_IN_FETCHREQUEST`: Specify the number of events that the Subscribe RPC requests from the server in each FetchRequest. 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 FetchRequest each. 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) in the Pub/Sub API documentation.
+       * `PROCESS_CHANGE_EVENT_HEADER_FIELDS`: Specify whether the Subscribe or ManagedSubscribe client should process the change data capture event bitmap fields in `ChangeEventHeader`. In this sample, only the `changedFields` field is expanded. To expand the `diffFields` and `nulledFields` header fields, modify the sample code. See [Event Deserialization Considerations](https://developer.salesforce.com/docs/platform/pub-sub-api/guide/event-deserialization-considerations.html).
        * `REPLAY_PRESET`: Specify the ReplayPreset for subscribe examples.
          * If a subscription has to be started using the CUSTOM replay preset, the `REPLAY_ID` parameter is mandatory. 
-         * The `REPLAY_ID` has to be specified in the following format: `[<byte_values_separated_by_commas>]`. Please enter the values as is within the square brackets and without any quotes. 
+         * The `REPLAY_ID` is a byte array and must be specified in this format: `[<byte_values_separated_by_commas>]`. Please enter the values as is within the square brackets and without any quotes. 
          * Example: `[0, 1, 2, 3, 4, -5, 6, 7, -8]`
 
 2. After setting up the configurations, any example can be executed using the `./run.sh` file available at the parent directory.
@@ -53,10 +59,12 @@ In the `src/main` directory of the project, you will find several sub-directorie
 
 ## Implementation
 - This repo can be used as a reference point for clients looking to create a Java app to integrate with Pub/Sub API. Note that the project structure and the examples included in this repo are intended for demo purposes only and clients are free to implement their own Java apps in any way they see fit.
-- The Subscribe RPC examples demonstrates a basic flow control strategy where a new `fetchRequest` is sent only after the requested number of events in the previous `fetchRequest(s)` are received. Custom flow control strategies can be implemented as needed. More info on flow control available [here](https://developer.salesforce.com/docs/platform/pub-sub-api/guide/flow-control.html).
+- The Generic Subscribe and ManagedSubscribe (beta) RPC examples create a long-lived subscription. After all requested events are received, Subscribe sends a new `FetchRequest` and ManagedSubscribe sends a new `ManagedFetchRequest` to keep the subscription alive and the client listening to new events.
+- The Generic Subscribe and ManagedSubscribe (beta) RPC examples demonstrate a basic flow control strategy where a new `FetchRequest` or `ManagedFetchRequest` is sent only after the requested number of events in the previous requests are received. The ManagedSubscribe RPC example also shows how to commit a Replay ID by sending commit requests. Custom flow control strategies can be implemented as needed. More info on flow control available [here](https://developer.salesforce.com/docs/platform/pub-sub-api/guide/flow-control.html).
+- The Generic Subscribe RPC example demonstrates error handling. After an exception occurs, it attempts to resubscribe after the last received event by implementing Binary Exponential Backoff. The example processes events and sends the retry requests asynchronously. If the error is an invalid replay ID,  it tries to resubscribe since the earliest stored event in the event bus. See the `onError()` method in `Subscribe.java`.
 
 # Limitations
 1. No guarantees that streams will remain open with `PublishStream` examples - Pub/Sub API has idle timeouts and will close idle streams. If a stream is closed while running these examples, you will most likely need to stop and restart.
 2. No support for republishing on error - If an error occurs while publishing the relevant examples will surface the error but will not attempt to republish the event.
 3. No security guarantees - Teams using these examples for reference will need to do their own security audits to ensure the dependencies introduced here can be safely used.
-4. No performance testing - These examples have not been perf tested.
+4. No performance testing - These examples have not been perf tested.

java/pom.xml

@@ -13,7 +13,9 @@
         <maven.compiler.source>11</maven.compiler.source>
         <maven.compiler.target>11</maven.compiler.target>
         <avro.version>1.11.0</avro.version>
-        <grpc.version>1.35.1</grpc.version>
+        <grpc.version>1.64.0</grpc.version>
+        <protoc.version>3.25.3</protoc.version>
+        <protocJava.version>1.64.0</protocJava.version>
     </properties>
 
     <dependencies>
@@ -83,9 +85,9 @@
                 <artifactId>protobuf-maven-plugin</artifactId>
                 <version>0.5.0</version>
                 <configuration>
-                    <protocArtifact>com.google.protobuf:protoc:3.3.0:exe:${os.detected.classifier}</protocArtifact>
+                    <protocArtifact>com.google.protobuf:protoc:${protoc.version}:exe:${os.detected.classifier}</protocArtifact>
                     <pluginId>grpc-java</pluginId>
-                    <pluginArtifact>io.grpc:protoc-gen-grpc-java:1.4.0:exe:${os.detected.classifier}</pluginArtifact>
+                    <pluginArtifact>io.grpc:protoc-gen-grpc-java:${protocJava.version}:exe:${os.detected.classifier}</pluginArtifact>
                 </configuration>
                 <executions>
                     <execution>
@@ -98,19 +100,20 @@
             </plugin>
             <plugin>
                 <groupId>org.apache.maven.plugins</groupId>
-                <artifactId>maven-assembly-plugin</artifactId>
+                <artifactId>maven-shade-plugin</artifactId>
                 <version>3.3.0</version>
-                <configuration>
-                    <descriptorRefs>
-                        <descriptorRef>jar-with-dependencies</descriptorRef>
-                    </descriptorRefs>
-                </configuration>
                 <executions>
                     <execution>
                         <phase>package</phase>
                         <goals>
-                            <goal>single</goal>
+                            <goal>shade</goal>
                         </goals>
+                        <configuration>
+                            <createDependencyReducedPom>false</createDependencyReducedPom>
+                            <transformers>
+                                <transformer implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer" />
+                            </transformers>
+                        </configuration>
                     </execution>
                 </executions>
             </plugin>

java/run.sh

@@ -10,4 +10,4 @@ if [ "x$EXAMPLE" = "x" ]; then
   echo "Please specify one of the example class names from the package com.salesforce.eventbusclient.example"
   exit -1
 fi
-java -cp target/pubsub-java-1.0-SNAPSHOT-jar-with-dependencies.jar $EXAMPLE $@
+java -cp target/pubsub-java-1.0-SNAPSHOT.jar $EXAMPLE $@

java/src/main/java/accountupdateapp/AccountListener.java

@@ -72,10 +72,8 @@ public void onNext(FetchResponse fetchResponse) {
                         logger.info(e.toString());
                     }
                 }
-                if (subscriber.getReceivedEvents().get() < subscriber.getTotalEventsRequested()) {
+                if (fetchResponse.getPendingNumRequested() == 0) {
                     subscriber.fetchMore(subscriber.getBatchSize());
-                } else {
-                    subscriber.receivedAllEvents.set(true);
                 }
             }
 
@@ -96,7 +94,6 @@ public void onCompleted() {
     // Helper function to start the app.
     public void startApp() throws InterruptedException {
         subscriber.startSubscription();
-        subscriber.waitForEvents();
     }
 
     // Helper function to stop the app.

java/src/main/java/accountupdateapp/AccountUpdateAppUtil.java

@@ -46,7 +46,7 @@ public class AccountUpdateAppUtil {
      */
     protected static GenericRecord createNewAccountRecord(Schema schema, String accountRecordId) {
         return new GenericRecordBuilder(schema).set("CreatedDate", System.currentTimeMillis() / 1000)
-                .set("CreatedById", "005xx000001Svwo").set("AccountRecordId__c", accountRecordId).build();
+                .set("CreatedById", "<User_Id>").set("AccountRecordId__c", accountRecordId).build();
     }
 
     /**
@@ -126,6 +126,7 @@ public static void updateAccountRecord(ExampleConfigurations subParams, String a
 
         if (res > 299) {
             logger.info("Unable to update Account Record.");
+            logger.info(response.getContentAsString());
         } else {
             logger.info("Successfully updated Account Record. Updated AccountNumber: " + accountNumber);
         }

java/src/main/java/accountupdateapp/AccountUpdater.java

@@ -65,10 +65,8 @@ public void onNext(FetchResponse fetchResponse) {
                         logger.info(e.toString());
                     }
                 }
-                if (subscriber.getReceivedEvents().get() < subscriber.getTotalEventsRequested()) {
+                if (fetchResponse.getPendingNumRequested() == 0) {
                     subscriber.fetchMore(subscriber.getBatchSize());
-                } else {
-                    subscriber.receivedAllEvents.set(true);
                 }
             }
 
@@ -89,7 +87,6 @@ public void onCompleted() {
     // Helper function to start the app.
     public void startApp() throws InterruptedException {
         subscriber.startSubscription();
-        subscriber.waitForEvents();
     }
 
     public void stopApp() {

java/src/main/java/accountupdateapp/README.md

@@ -9,7 +9,7 @@ This example subscribes to change events corresponding to the creation of [Accou
     * Search for the `Account` object and click on the right arrow in the middle of the screen to select the entity.
     * Click on the `Save` button to update the changes.
 2. The `NewAccount` custom platform needs to be created with the following fields:
-    - Standard Fields
+    - Platform Event Name
         - Label: `NewAccount`
         - Plural Label: `NewAccounts`
     - Custom Fields
@@ -35,5 +35,11 @@ This example subscribes to change events corresponding to the creation of [Accou
 ```
 
 ## Notes:
-* Subscribers in both the `AccountUpdater` and `AccountListener` subscribe with the ReplayPreset set to LATEST. Therefore, only events generated once the examples have started running will be processed
+* Please use the `my domain` URL for your org for running these examples. You can find the my domain URL through Developer Console.
+  * Open Developer Console
+  * Click on the Debug menu and select Open Execute Anonymous Window.
+  * Key in the following in the window: `System.debug(System.url.getOrgDomainUrl());` and execute the same.
+  * Once done, in the Logs tab below open the logs recently executed code. 
+  * In the logs, get the `my domain` URL from the USER_DEBUG event.  
+* Subscribers in both the `AccountUpdater` and `AccountListener` subscribe with the ReplayPreset set to LATEST. Therefore, only events generated once the examples have started running will be processed.
 * The `AccountUpdater` logs the `AccountNumber` that has been added to the `Account` record which can be used to verify if the update is correct.