public void setupPushListener()

in extensions/salesforce-connector/services/src/main/java/org/apache/unomi/sfdc/services/internal/SFDCServiceImpl.java [599:676]
62 lines of code
11 McCabe index (conditional complexity)

    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.info("[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.info("[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);

    }