Durable Message Replay from Change Data Capture

Question

Given the following posts from Salesforce:

• Message Durability
• Replay Events Sample: Code Walkthrough
• StreamingReplayClientExtensions/java/CometDReplayExtension.java

The way I'm understanding the behavior here is that during the connection to the /meta/subscribe channel, if I wish to replay prior durable events, I can use the following options:

• -2: All events from the beginning of what is available/stored
• -1: Everything going forward after subscription
• Replay#: A previous 'replay' value from a prior message as starting point

I'm trying to pass the map value { "replay": { "/data/ChangeEvents": -2 } along when { "channel": "/meta/subscribe" } connection is going out but it is not replaying the 3 days prior changes as noted in the documentation for change data capture.

Is this the correct way to retrieve change data capture events for the retention period? I'm assuming I should see all events replayed after making the connection but nothing old is coming through, only new changes.

Answer 1

I don't know if you are still looking for an answer for this but I was able to make it work. Here is how I did it:

First step was to add the replay extension to the bayeux client (please note that I am using C#)

var bayeuxClient = new BayeuxClient(endpoint, new [] {transport});
bayeuxClient.AddExtension(new ReplayExtension());

Next I added a listener to the meta/handshake channel. This is where I will do my subscriptions.

bayeuxClient.GetChannel("/meta/handshake").AddListener(new HandshakeListener(bayeuxClient));

The HandshakeListener implements the IMessageListener interface, so it has an OnMessage method where we can perform the channel subscriptions.

public void OnMessage(IClientSessionChannel channel, IMessage message)
{
    _bayeuxClient.GetChannel(channelName, replayId).Subscribe(new Listener());
}

You just need to find a way to store the replayId per channel so you can access it in the HandshakeListener's OnMessage method. I personally created a class that inherits from BayeuxClient and I added a Dictionary<string, long> which contains the channel name and its associated replayId. I then pass the instance of the class to the HandshakeListener who in turn passes it to all the channel listeners. When a message is received in a listener, I update the dictionary with the new replayId.

I still need to look into that because I feel there is an easier way to do this but it works. I tried hardcoding -2 and I received all the changes I had made that day, not just the new ones. I didn't try going back 3 days but I noticed the documentation is saying that this limit is for high-volume events as opposed to 24 hours for standard-volume events.

Hope this helps!