SFDCServiceImpl implements SFDCService { public void setupPushListener(String channelName, ClientSessionChannel.MessageListener messageListener) throws Exception { if (!isConnected()) { return; } final BayeuxClient client = makeClient(); if (client == null) { throw new Exception("Login failed !"); } client.getChannel(Channel.META_HANDSHAKE).addListener (new ClientSessionChannel.MessageListener() { @Override public void onMessage(ClientSessionChannel channel, Message message) { logger.debug("[CHANNEL:META_HANDSHAKE]: " + message); boolean success = message.isSuccessful(); if (!success) { String error = (String) message.get("error"); if (error != null) { logger.error("Error during HANDSHAKE: " + error); } Exception exception = (Exception) message.get("exception"); if (exception != null) { logger.error("Exception during HANDSHAKE: ", exception); } } } }); client.getChannel(Channel.META_CONNECT).addListener( new ClientSessionChannel.MessageListener() { public void onMessage(ClientSessionChannel channel, Message message) { logger.debug("[CHANNEL:META_CONNECT]: " + message); boolean success = message.isSuccessful(); if (!success) { String error = (String) message.get("error"); if (error != null) { logger.error("Error during CONNECT: " + error); } } } }); client.getChannel(Channel.META_SUBSCRIBE).addListener( new ClientSessionChannel.MessageListener() { public void onMessage(ClientSessionChannel channel, Message message) { logger.debug("[CHANNEL:META_SUBSCRIBE]: " + message); boolean success = message.isSuccessful(); if (!success) { String error = (String) message.get("error"); if (error != null) { logger.error("Error during SUBSCRIBE: " + error); } } } }); client.handshake(); logger.debug("Waiting for handshake"); boolean handshaken = client.waitFor(10 * 1000, BayeuxClient.State.CONNECTED); if (!handshaken) { logger.error("Failed to handshake: " + client); } logger.info("Subscribing for channel: " + channelName); client.getChannel(channelName).subscribe(messageListener); }  void setPersistenceService(PersistenceService persistenceService); void setDefaultSFDCConfiguration(SFDCConfiguration defaultSFDCConfiguration); SFDCSession getSFDCSession(); @Override SFDCConfiguration loadConfiguration(); @Override boolean saveConfiguration(SFDCConfiguration sfdcConfiguration); void start(); void stop(); Set<String> getRecentLeadIds(); Map<String, Object> getSObject(String sobjectName, String objectId); Map<String, Object> getSObjectDescribe(String sobjectName); Map<String, Object> getLead(String leadId); Set<String> getLeadMandatoryFields(); boolean deleteLead(String leadId); Set<String> findLeadIdsByIdentifierValue(String identifierFieldValue); @Override String createOrUpdateLead(Profile profile); @Override boolean updateProfileFromLead(Profile profile); @Override Map<String, Object> query(String query); @Override Map<String, Object> getLimits(); void setupPushListener(String channelName, ClientSessionChannel.MessageListener messageListener); boolean login(SFDCConfiguration sfdcConfiguration); void logout(); boolean isConfigured(); boolean isConnected();  }
@Test public void testStreaming() throws Exception { if (!checkCanRunTests()) return; System.out.println("Running streaming client example...."); sfdcServiceImpl.setupPushListener(sfdcConfiguration.getSfdcChannel(), new ClientSessionChannel.MessageListener() { @Override public void onMessage(ClientSessionChannel clientSessionChannel, Message message) { System.out.println("Received message for channel" + sfdcConfiguration.getSfdcChannel() + ":"+ message); } }); System.out.println("Waiting 10 seconds for streamed data from your organization ..."); int i=0; while (i < 10) { Thread.sleep(1000); i++; } }