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.

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.

Note

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.
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<GenericRecord> createEventMessages(Schema schema, final int numEvents) method. Replace the <User_Id> 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.

1.set("CreatedById", "<User_Id>")

In a second terminal window, go to the top-level java folder in your pub-sub-api folder.
To build the Java client, enter mvn clean install.
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.
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.

12024-06-03 09:12:30,942 [main] java.lang.Class - Using grpcHost api.pubsub.salesforce.com and grpcPort 7443
22024-06-03 09:12:31,524 [main] java.lang.Class - Starting managed subscription
3with developer name My_Managed_Subscription
42024-06-03 09:12:31,738 [grpc-nio-worker-ELG-1-2] utility.APISessionCredentials -
5Client Trace Id for current request: cd5faf97-8fa1-4efe-a770-d6166067b922
62024-06-03 09:12:36,531 [main] java.lang.Class - Subscription Active. Received a total of 0 events.
72024-06-03 09:12:41,535 [main] java.lang.Class - Subscription Active. Received a total of 0 events.
82024-06-03 09:12:44,520 [grpc-default-executor-1] java.lang.Class - ManagedFetchResponse batch
9of 5 events pending requested: 0
102024-06-03 09:12:44,520 [grpc-default-executor-1] java.lang.Class - RPC ID: e5219940-7133-b6a4-97a6-9beabbc7573e
112024-06-03 09:12:44,520 [grpc-default-executor-1] java.lang.Class - processEvent -
12EventID: f23132b9-7df6-446b-8fe6-2b7cee811376 SchemaId: _CADE3qaegX3ECEW0rlbYA
132024-06-03 09:12:44,524 [grpc-default-executor-1] utility.APISessionCredentials -
14Client Trace Id for current request: 0c1cbc1c-1189-40fe-8b9b-cc64c333e530
152024-06-03 09:12:44,685 [grpc-default-executor-1] java.lang.Class - Received event:
16{
17  "CreatedDate": 1717431164101,
18  "CreatedById": "005SG000009B3S1YAK",
19  "Order_Number__c": "99",
20  "City__c": "Los Angeles",
21  "Amount__c": 35.0
22}
232024-06-03 09:12:44,685 [grpc-default-executor-1] java.lang.Class - processEvent -
24EventID: 981d92b0-179f-40f4-be72-fc93ae8a3660 SchemaId: _CADE3qaegX3ECEW0rlbYA
252024-06-03 09:12:44,685 [grpc-default-executor-1] java.lang.Class - Received event:
26{
27  "CreatedDate": 1717431164101,
28  "CreatedById": "005SG000009B3S1YAK",
29  "Order_Number__c": "100",
30  "City__c": "New York",
31  "Amount__c": 20.0
32}
332024-06-03 09:12:44,685 [grpc-default-executor-1] java.lang.Class - processEvent -
34EventID: 94b4b89a-9158-4d4b-beaa-63e3555c4cbd SchemaId: _CADE3qaegX3ECEW0rlbYA
352024-06-03 09:12:44,685 [grpc-default-executor-1] java.lang.Class - Received event:
36{
37  "CreatedDate": 1717431164101,
38  "CreatedById": "005SG000009B3S1YAK",
39  "Order_Number__c": "101",
40  "City__c": "San Francisco",
41  "Amount__c": 2.0
42}
432024-06-03 09:12:44,685 [grpc-default-executor-1] java.lang.Class - processEvent -
44EventID: dc37a809-542b-4696-b6f5-bede7a7733d0 SchemaId: _CADE3qaegX3ECEW0rlbYA
452024-06-03 09:12:44,685 [grpc-default-executor-1] java.lang.Class - Received event:
46{
47  "CreatedDate": 1717431164101,
48  "CreatedById": "005SG000009B3S1YAK",
49  "Order_Number__c": "102",
50  "City__c": "San Jose",
51  "Amount__c": 123.0
52}
532024-06-03 09:12:44,685 [grpc-default-executor-1] java.lang.Class - processEvent -
54EventID: 102bedd7-df2c-4daf-b7c9-e25acde4be47 SchemaId: _CADE3qaegX3ECEW0rlbYA
552024-06-03 09:12:44,685 [grpc-default-executor-1] java.lang.Class - Received event:
56{
57  "CreatedDate": 1717431164101,
58  "CreatedById": "005SG000009B3S1YAK",
59  "Order_Number__c": "103",
60  "City__c": "Boston",
61  "Amount__c": 180.0
62}
632024-06-03 09:12:44,685 [grpc-default-executor-1] java.lang.Class - Processed batch of 5 event(s)
642024-06-03 09:12:44,686 [grpc-default-executor-1] java.lang.Class - Sending CommitRequest with
65CommitReplayRequest ID: bd557942-4fd1-4699-be18-70db3054b33c
662024-06-03 09:12:44,686 [grpc-default-executor-1] java.lang.Class - Fetching more events: 5
672024-06-03 09:12:44,686 [main] java.lang.Class - Subscription Active. Received a total of 5 events.
682024-06-03 09:12:44,795 [grpc-default-executor-1] java.lang.Class - ManagedFetchResponse batch
69of 0 events pending requested: 5
702024-06-03 09:12:44,796 [grpc-default-executor-1] java.lang.Class - RPC ID: e5219940-7133-b6a4-97a6-9beabbc7573e
712024-06-03 09:12:44,796 [grpc-default-executor-1] java.lang.Class - Successfully committed replay
72with CommitRequestId: bd557942-4fd1-4699-be18-70db3054b33c with process time: 1717431164709
732024-06-03 09:12:44,796 [main] java.lang.Class - Subscription Active. Received a total of 5 events.

To disconnect the ManagedSubscribe client, press Ctrl+C in the terminal.
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.
1String[] 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.
Return to the ManagedSubscribe client terminal. Restart the managed subscription by entering ./run.sh genericpubsub.ManagedSubscribe.
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.

12024-06-03 09:14:09,931 [main] java.lang.Class - Using grpcHost api.pubsub.salesforce.com and grpcPort 7443
22024-06-03 09:14:10,487 [main] java.lang.Class - Starting managed subscription with developer name My_Managed_Subscription
32024-06-03 09:14:10,663 [grpc-nio-worker-ELG-1-2] utility.APISessionCredentials -
4Client Trace Id for current request: 22fdf915-3846-49b6-9943-09aa95b50903
52024-06-03 09:14:11,452 [grpc-default-executor-1] java.lang.Class - ManagedFetchResponse batch
6of 5 events pending requested: 0
72024-06-03 09:14:11,453 [grpc-default-executor-1] java.lang.Class - RPC ID: 579112d8-c727-b880-94a1-1eceb6f72882
82024-06-03 09:14:11,453 [grpc-default-executor-1] java.lang.Class - processEvent -
9EventID: 160810f4-f73f-4071-ad7f-a869b3e46dd8 SchemaId: _CADE3qaegX3ECEW0rlbYA
102024-06-03 09:14:11,461 [grpc-default-executor-1] utility.APISessionCredentials -
11Client Trace Id for current request: e734fa7b-dd28-4f80-bfc8-94c06df82f25
122024-06-03 09:14:11,576 [grpc-default-executor-1] java.lang.Class - Received event:
13{
14  "CreatedDate": 1717431244798,
15  "CreatedById": "005SG000009B3S1YAK",
16  "Order_Number__c": "104",
17  "City__c": "Los Angeles",
18  "Amount__c": 35.0
19}
202024-06-03 09:14:11,576 [grpc-default-executor-1] java.lang.Class - processEvent -
21EventID: b13516f1-7c40-4b7f-a57f-cdb2cc936579 SchemaId: _CADE3qaegX3ECEW0rlbYA
222024-06-03 09:14:11,576 [grpc-default-executor-1] java.lang.Class - Received event:
23{
24  "CreatedDate": 1717431244798,
25  "CreatedById": "005SG000009B3S1YAK",
26  "Order_Number__c": "105",
27  "City__c": "New York",
28  "Amount__c": 20.0
29}
302024-06-03 09:14:11,576 [grpc-default-executor-1] java.lang.Class - processEvent -
31EventID: 24a4be7c-441c-4c42-886d-09e5de87c0c3 SchemaId: _CADE3qaegX3ECEW0rlbYA
322024-06-03 09:14:11,576 [grpc-default-executor-1] java.lang.Class - Received event:
33{
34  "CreatedDate": 1717431244798,
35  "CreatedById": "005SG000009B3S1YAK",
36  "Order_Number__c": "106",
37  "City__c": "San Francisco",
38  "Amount__c": 2.0
39}
402024-06-03 09:14:11,576 [grpc-default-executor-1] java.lang.Class - processEvent -
41EventID: 302f4a46-f970-4341-a1c5-63d36070cd36 SchemaId: _CADE3qaegX3ECEW0rlbYA
422024-06-03 09:14:11,576 [grpc-default-executor-1] java.lang.Class - Received event:
43{
44  "CreatedDate": 1717431244798,
45  "CreatedById": "005SG000009B3S1YAK",
46  "Order_Number__c": "107",
47  "City__c": "San Jose",
48  "Amount__c": 123.0
49}
502024-06-03 09:14:11,576 [grpc-default-executor-1] java.lang.Class - processEvent -
51EventID: a5d081d5-5678-4fdc-b3c9-3a52b61fc04a SchemaId: _CADE3qaegX3ECEW0rlbYA
522024-06-03 09:14:11,576 [grpc-default-executor-1] java.lang.Class - Received event:
53{
54  "CreatedDate": 1717431244798,
55  "CreatedById": "005SG000009B3S1YAK",
56  "Order_Number__c": "108",
57  "City__c": "Boston",
58  "Amount__c": 180.0
59}
602024-06-03 09:14:11,576 [grpc-default-executor-1] java.lang.Class - Processed batch of 5 event(s)
612024-06-03 09:14:11,577 [grpc-default-executor-1] java.lang.Class - Sending CommitRequest with
62CommitReplayRequest ID: 7cf61ac5-cb17-49ae-975e-4f236f194945
632024-06-03 09:14:11,577 [grpc-default-executor-1] java.lang.Class - Fetching more events: 5
642024-06-03 09:14:11,578 [main] java.lang.Class - Subscription Active. Received a total of 5 events.
652024-06-03 09:14:11,638 [grpc-default-executor-1] java.lang.Class - ManagedFetchResponse batch
66of 0 events pending requested: 5
672024-06-03 09:14:11,638 [grpc-default-executor-1] java.lang.Class - RPC ID: 579112d8-c727-b880-94a1-1eceb6f72882
682024-06-03 09:14:11,638 [grpc-default-executor-1] java.lang.Class - Successfully committed replay
69with CommitRequestId: 7cf61ac5-cb17-49ae-975e-4f236f194945 with process time: 1717431251600
702024-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.