Learn More About EMP Connector

Let’s take a closer look at the components of EMP Connector.

Authenticating

The LoginExample class logs in to production by default using the passed-in user-credential information.

After initial authentication, LoginExample reauthenticates the user if the authentication becomes invalid, such as when a Salesforce session is invalidated or an access token is revoked. LoginExample listens to 401::Authentication invalid error messages that Streaming API sends when the authentication is no longer valid. The class reauthenticates after a 401 error is received. The token provider performs the reauthentication and is set using the EmpConnector.setBearerTokenProvider() method.

1BearerTokenProvider tokenProvider = new BearerTokenProvider(() -> {
2    try {
3        return login(argv[0], argv[1]);
4    } catch (Exception e) {
5        e.printStackTrace(System.err);
6        System.exit(1);
7        throw new RuntimeException(e);
8    }
9});
10
11BayeuxParameters params = tokenProvider.login();
12// . . . 
13connector.setBearerTokenProvider(tokenProvider);

For OAuth authentication, the BearerTokenExample uses the BayeuxParameters constructor to override the methods in the BayeuxParameters class and provides the token and URL values.

1BayeuxParameters params = new BayeuxParameters() {
2
3    @Override
4    public String bearerToken() {
5        return "<token>";
6    }
7
8    @Override
9    public URL host() {
10        try {
11            return new URL("<URL>");
12        } catch (MalformedURLException e) {
13            throw new IllegalArgumentException(
14                String.format("Unable to create url: %s", argv[0]), e);
15        }
16    }
17};

BearerTokenExample doesn’t support reauthentication, but you can add this support. Reauthentication is implemented only in LoginExample and DevLoginExample.

Note

Listening to Events

To listen to events, the connector uses the Java event in a lambda expression. This statement prints the event message to the output for each received event notification. Place this statement before the statement that subscribes to the channel.

1Consumer<Map<String, Object>> consumer = event -> System.out.println(
2        String.format("Received:\n%s", JSON.toString(event)));

Subscribing to a Channel

The EmpConnector class is the main class that exposes the functionality of starting a connection and subscribing. The class contains functions to create a connection, subscribe to a channel, cancel a subscription, and stop a connection.

1// Instantiate the EMP connector
2EmpConnector connector = new EmpConnector(params);
3
4connector.setBearerTokenProvider(tokenProvider);
5
6// Wait for handshake with Streaming API
7connector.start().get(5, TimeUnit.SECONDS);
8
9// Subscribe to a channel
10// Block and wait for the subscription to succeed for 5 seconds
11TopicSubscription subscription = connector.subscribe("<Channel_Name>", 
12    replayFrom, consumer).get(5, TimeUnit.SECONDS);

To end a subscription, call these functions.

1// Cancel a subscription
2subscription.cancel();
3
4// Stop the connector
5connector.stop();

Debug Logging

To aid in debugging, the LoggingListener class logs Bayeux messages to the console. BearerTokenExample and DevLoginExample use logging but not LoginExample. DevLoginExample is part of the EMP Connector GitHub project, but is not covered in this walkthrough. For more information, see the EMP Connector Readme page.