Create EmpConnector connection
The first thing is to create a configuration that on startup will connect to the EmpConnector and start listing for incoming events on a specific topic.
1. Set all your Bayeux Parameters (bayeuxParameters)
2. Create EmpConnector connection (empConnector)
3. Start listening to topic (startAndPublishAsyncEventToExchange)
4. Adding listeners to the topic (startAndPublishAsyncEventToExchange)
package com.app.core.events;
import com.app.core.service.IncomingRequestHandler;
import com.app.utils.AppPlanUtils;
import com.salesforce.emp.connector.BayeuxParameters;
import com.salesforce.emp.connector.EmpConnector;
import com.salesforce.emp.connector.TopicSubscription;
import com.salesforce.emp.connector.example.LoggingListener;
import org.cometd.bayeux.Channel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Lazy;
import java.net.URL;
import java.util.Map;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.function.Consumer;
import static com.salesforce.emp.connector.LoginHelper.login;
@Configuration
@ConfigurationProperties
@Lazy
public class SalesforceEventConfig {
private static Logger itsLogger = LoggerFactory.getLogger(SalesforceEventConfig.class);
@Value("${salesforce.username}")
private String username;
@Value("${salesforce.password}")
private String password;
@Value("${salesforce.token}")
private String token;
@Value("${salesforce.baseurl}")
private String url;
@Value("${salesforce.version:39.0}")
private String version;
@Autowired
protected IncomingRequestHandler itsReqHandler;
private static final Integer WAIT_TIME = 2;
private static final Integer START_UP_WAIT_TIME = 5;
private static final String EVENT_NAME = "App_Events__e";
@Bean
public BayeuxParameters bayeuxParameters() throws Exception {
String theUrlEnv = AppUtils.getConfigVar("SALESFORCE_URL");
String passwordAndToken = password + token;
if (theUrlEnv != null) {
url = getNextToken(0, theUrlEnv);
username = getNextToken(url.length() + version.length() + 2, theUrlEnv);
passwordAndToken = theUrlEnv.substring(url.length() + version.length() + username.length() + 3);
itsLogger.info("Found SALESFORCE_URL to parse. url={}, version={}, username={}", url, version, username);
url = url.split("/services/Soap/u/")[0];
}
BayeuxParameters bayeuxParams = getBayeuxParamWithSpecifiedAPIVersion(version);
BayeuxParameters bayeuxParameters = login(new URL(url),username, passwordAndToken, bayeuxParams);
return bayeuxParameters;
}
@Bean
public EmpConnector empConnector(BayeuxParameters bayeuxParameters){
itsLogger.debug("BayeuxParameters url {} version {}", bayeuxParameters.host(), bayeuxParameters.version());
EmpConnector empConnector = new EmpConnector(bayeuxParameters);
itsLogger.debug("EmpConnector isConnected {} isHandshook {}", empConnector.isConnected(), empConnector.isHandshook());
return empConnector;
}
private static BayeuxParameters getBayeuxParamWithSpecifiedAPIVersion(String apiVersion) {
BayeuxParameters params = new BayeuxParameters() {
@Override
public String version() {
return apiVersion;
}
@Override
public String bearerToken() {
return null;
}
};
return params;
}
private String getNextToken(int startIdx, String origStr) {
int theIdx = origStr.indexOf("|", startIdx);
if (theIdx > 0) {
return origStr.substring(startIdx, theIdx);
} else {
return origStr.substring(startIdx);
}
}
@Bean
public TopicSubscription startAndPublishAsyncEventToExchange(EmpConnector empConnector) {
TopicSubscription subscription = null;
try {
long replayFrom = EmpConnector.REPLAY_FROM_TIP;
itsLogger.debug("Setup event consumer with replyFrom {}", replayFrom);
SalesforceEventPayload eventPayload = new SalesforceEventPayload();
Consumer<Map<String, Object>> consumer = event -> {
eventPayload.setPayload(event);
itsReqHandler.handleRequest(eventPayload.getPayload());
};
LoggingListener loggingListener = new LoggingListener(true, true);
itsLogger.debug("Adding event listeners");
empConnector.addListener(Channel.META_CONNECT, loggingListener)
.addListener(Channel.META_HANDSHAKE, loggingListener)
.addListener(Channel.META_DISCONNECT, loggingListener)
.addListener(Channel.META_SUBSCRIBE, loggingListener)
.addListener(Channel.META_UNSUBSCRIBE, loggingListener)
.addListener(Channel.META_DISCONNECT, loggingListener)
.addListener(Channel.SERVICE, loggingListener);
itsLogger.debug("Starting Event Bus");
empConnector.start().get(START_UP_WAIT_TIME, TimeUnit.SECONDS);
itsLogger.debug("Subscribing to event {}", EVENT_NAME);
subscription = empConnector.subscribe("/event/" + EVENT_NAME, replayFrom, consumer).get(WAIT_TIME, TimeUnit.SECONDS);
itsLogger.debug(String.format("Subscribed: %s", subscription));
} catch (InterruptedException e) {
e.printStackTrace();
} catch (ExecutionException e) {
e.printStackTrace();
} catch (TimeoutException e) {
e.printStackTrace();
}
return subscription;
}
}
Basic POJO to serialize events received for a specific topic
Note: as the ‘payload’ tag has some extra tags I remove them before serializing the payload to POJO.
package com.app.core.events;
import com.app.db.model.UserServiceRequestEvent;
import com.fasterxml.jackson.core.JsonParseException;
import com.fasterxml.jackson.databind.DeserializationFeature;
import com.fasterxml.jackson.databind.JsonMappingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.commons.lang3.StringEscapeUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.io.Serializable;
import java.util.Map;
public class SalesforceEventPayload implements Serializable{
private static final String SALESFORCE_EVENT_KEY = "Event_Data__c=";
private static Logger itsLogger = LoggerFactory.getLogger(SalesforceEventPayload.class);
public SalesforceEventPayload() {}
private UserServiceRequestEvent payload;
public UserServiceRequestEvent getPayload() {
return payload;
}
public void setPayload(Map<String, Object> fieldMappings) {
ObjectMapper theMapper = new ObjectMapper();
theMapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
theMapper.configure(DeserializationFeature.FAIL_ON_IGNORED_PROPERTIES, false);
theMapper.configure(DeserializationFeature.USE_JAVA_ARRAY_FOR_JSON_ARRAY, true);
theMapper.configure(DeserializationFeature.ACCEPT_EMPTY_STRING_AS_NULL_OBJECT, true);
theMapper.configure(DeserializationFeature.ACCEPT_EMPTY_ARRAY_AS_NULL_OBJECT, true);
if (fieldMappings.containsKey("payload")){
String eventPayload = fieldMappings.get("payload").toString();
String splitEventString = eventPayload.toString().split(SALESFORCE_EVENT_KEY)[1];
String removeLastTwo = splitEventString.substring(1, splitEventString.length() - 2);
String unescapeJson = StringEscapeUtils.unescapeJson(removeLastTwo);
try{
payload = theMapper.readValue(unescapeJson, UserServiceRequestEvent.class);
} catch (JsonParseException e) {
itsLogger.error("Event json parse exception {}", e.getMessage());
} catch (JsonMappingException e) {
itsLogger.error("Event json mapping exception {}", e.getMessage());
} catch (IOException e) {
itsLogger.error("Event io exception {}", e.getMessage());
}
}
}
}