CometD Issues with Publishing Data to a Channel

Question

I am new to cometd, I have planned to send a message to the server and get the message in my browser using cometd If i send a message it is successfully send to the server but couldn't get it in the browser

 private void testService() {
String channelName = "/service/out";


log.info("Channel Name = " + channelName);
        log.info("bayeuxServer : " + (bayeuxServer == null ? "Is Null" : "Is Not Null"));



System.out.println("CHANNELS : " + bayeuxServer.getChannels().toString());
        System.out.println("Subscribers on /service/in = "+bayeuxServer.getChannel("/service/in").getSubscribers().toString());
        System.out.println("Subscribers on /service/out = "+bayeuxServer.getChannel("/service/out").getSubscribers().toString());

// convert to cometd format
        Map<String, Object> data = new HashMap<String, Object>(4);
        data.put("serverMsg", getDetails());
ServerChannel channel = bayeuxServer.getChannel(channelName);
subscribers = channel.getSubscribers().size();
        log.info("Subscribers = " + subscribers);
        log.info("channel = " + channel);
channel.publish(sender, data, null);


System.out.println("Session subscriptions :" + sender.getServerSession().getSubscriptions());
        System.out.println("Listeners on /service/out = "+bayeuxServer.getChannel("/service/out").getListeners().toString());
        System.out.println("Subscribers on /service/out = "+bayeuxServer.getChannel("/service/out").getSubscribers().toString());

}

But this one is not working

  @Subscription("/service/out")
    public void echo(Message message)
    {
        System.out.println("Echo service published " + message);
    }

Logs:

1

7:48:18,775 INFO  ClientHelloService [] Bayeux server =org.cometd.server.BayeuxServerImpl@1436088
17:48:18,775 INFO  ClientHelloService [] Message = Hello world
17:48:18,775 INFO  ClientHelloService [] remote Client Id = 3renjdwk25ercglzli36tudpl
17:48:18,776 INFO  ClientHelloService [] Local session = L:_21w17u5f3mvvluvp71c27yaqvt
17:48:18,776 INFO  ClientHelloService [] session = 3renjdwk25ercglzli36tudpl - last connect 1 ms ago
17:48:18,776 INFO  ClientHelloService [] Channel Name = /service/out
17:48:18,776 INFO  ClientHelloService [] bayeuxServer : Is Not Null
CHANNELS : [/service/out, /meta/subscribe, /service, /service/*, /meta, /meta/handshake, /meta/disconnect, /service/in, /meta/connect, /meta/unsubscribe]
Subscribers on /service/in = []
Subscribers on /service/out = []
17:48:18,777 INFO  ClientHelloService [] msg = Hello world
17:48:18,777 INFO  ClientHelloService [] Subscribers = 0
17:48:18,777 INFO  ClientHelloService [] channel = /service/out
17:48:18,777 DEBUG 21192840 [] <  {data={serverMsg=Hello world}, channel=/service/out}
17:48:18,777 INFO  ClientHelloService [] publish the channel
Session subscriptions :[]
Listeners on /service/out = []
Subscribers on /service/out = []

Application.js

var sendChannel = '/service/in';            // Message from jsp
var receiveChannel = '/service/*';       // Message from server
/*var cometdServerURL = 'http://127.0.0.1:8080/cometd';*/

require(['dojox/cometd', 'dojo/dom', 'dojo/domReady!'], function(cometd, dom)
{       //  // configuration object
  //  cometd.websocketEnabled = true;
   // Open connection to CometD server
    cometd.configure({
        url: location.protocol + '//' + location.host + config.contextPath + '/cometd',
        logLevel: 'debug'
    });

cometd.addListener('/meta/*', function(message)
    {
        if (message.successful)
        {
            dom.byId('status').innerHTML += '<div>CometD handshake successful</div>';

            cometd.subscribe(receiveChannel, function(message) {
                dom.byId('results').innerHTML +=' Message  from server  ' + message.data;
                dom.byId('results').innerHTML +=' Subscription to ' + receiveChannel;
            });
        }
        else if(_connectionBroken()){
            dom.byId('status').innerHTML += '<div>CometD Connection Broken</div>';
        }
        else
        {
            dom.byId('status').innerHTML += '<div>CometD handshake failed</div>';
        }
    });
dom.byId('greeter').onclick = function()
    {
        var text = dom.byId('msg').value;
        cometd.publish(sendChannel, 'Hello world');
        dom.byId('msg').value = "" ;
        dom.byId('results').innerHTML +='Message send to server' ;
    };
    cometd.handshake();
});

Answer 1

There are several mistakes in your code, addressed below.

First, you don't want in your javascript to add a listener to /meta/* to handle subscriptions in the if (message.successful) branch. That branch will be executed for any meta message response, for example also for responses to subscriptions (that are sent over /meta/subscribe), executing the code multiple times when that is not the intention.

Change the listener to listen to /meta/handshake channel and perform the subscription in that listener.

Likewise you want to execute the if (_connectionBroken()) branch in a /meta/connect listener.

Please refer to the primer to build a proper skeleton of your application. Also, follow the tutorials to better understand the roles of your listeners.

Second, it is not recommended that you subscribe, from the client, to service channels. Have a read at the CometD concepts to understand the difference between a service channel and a broadcast channel, and about the difference between adding a listener and subscribing.

Third, when you have service channel, publish() is a local activity, so no message will be delivered to remote clients. The right API to use in this case is ServerSession.deliver(...), if you really want to use service channels.

Finally, your use case is covered by the tutorials, so I recommend you follow those and your application will work.

Hope that helped.

Answer 2

var sendChannel = '/service/in';            // Message from jsp
var receiveChannel = '/service/out';       // Message from server
/*var cometdServerURL = 'http://127.0.0.1:8080/cometd';*/

require(['dojox/cometd', 'dojo/dom', 'dojo/domReady!','dojo/_base/unload'], function(cometd, dom)
{       //  // configuration object

   // Open connection to CometD server
    cometd.configure({
        url: location.protocol + '//' + location.host + config.contextPath + '/cometd',
        logLevel: 'debug'
    });
   cometd.addListener('/meta/connect', function(message) {
        var wasConnected;
        if(cometd.isDisconnected()) {
            dom.byId('status').innerHTML +=' Disconnected from the server  = ' +  message;
        }  else{
           /* dom.byId('status').innerHTML +=' Disconnected from the server  = ' +  message.data;*/
        }
    });

 // listener for handshake
    cometd.addListener('/meta/handshake', function(message)
    {
        if (message.successful)
        {
            dom.byId('status').innerHTML += '<div>CometD handshake successful</div>';
            cometd.batch(function()
            {
            cometd.subscribe(receiveChannel, function(message) {
                dom.byId('results').innerHTML +=' Message  from server  ' + message.data;
                dom.byId('results').innerHTML +=' Subscription to ' + receiveChannel;
            });
               // cometd.publish('/service/in', { name: 'World' });
            });
        }
        else if(_connectionBroken()){
            dom.byId('status').innerHTML += '<div>CometD Connection Broken</div>';
        }
        else
        {
            dom.byId('status').innerHTML += '<div>CometD handshake failed</div>';
        }
    });

    dom.byId('greeter').onclick = function()
    {
        var text = dom.byId('msg').value;
        cometd.publish(sendChannel, 'Hello world');
        dom.byId('msg').value = "" ;
        dom.byId('results').innerHTML +='Message send to server' ;
    };
    cometd.handshake();

  });

Now also couldnt publish to the channels

@Service
public class ClientHelloService {

    static Logger log = Logger.getLogger(
            ClientHelloService.class.getName());
    String details;
    @Inject
    private BayeuxServer bayeuxServer;
    @Session
    private LocalSession sender;
    @Session
    private ClientSession bayeuxClient;
    @Session
    ServerSession session;
    int subscribers;
       // represent callback
    @Listener("/service/in")
    public void processClientHello(ServerSession session, ServerMessage message)
    {
        log.info("Bayeux server =" + bayeuxServer);
        log.info("Message = " + message.getData());
        log.info("remote Client Id = " + session.getId());
        log.info("Local session = " + sender);
        log.info("session = " + getSession());
       // log.info("sender = " + sender);
        details = (String) message.getData();
       // session.deliver(sender,"/service/out",getDetails() , null);
        testService();
    }

    private void testService() {
        // Create the channel name  using the symbol

        String channelName = "/service/out";
        log.info("Channel Name = " + channelName);
        log.info("bayeuxServer : " + (bayeuxServer == null ? "Is Null" : "Is Not Null"));
        // Initialize the channel, making it persistant
        // new sendChannel
        bayeuxServer.createIfAbsent(channelName, new ConfigurableServerChannel.Initializer() {
            public void configureChannel(ConfigurableServerChannel channel) {
                log.info("Configurable channel " + channel);
                //  channel exists even if it has no subscribers
                channel.setPersistent(true);
                channel.setLazy(true);
            }
        });
        System.out.println("CHANNELS : " + bayeuxServer.getChannels().toString());
        System.out.println("Subscribers on /service/out = "+bayeuxServer.getChannel("/service/out").getSubscribers().toString());

        // convert to cometd format
        Map<String, Object> data = new HashMap<String, Object>(4);
        data.put("serverMsg", getDetails());
        log.info("msg = " + data.get("serverMsg"));

        // Publish the channel  to all
        ServerChannel channel = bayeuxServer.getChannel(channelName);
        subscribers = channel.getSubscribers().size();
        log.info("Subscribers = " + subscribers);
        log.info("channel = " + channel);
        // publish the message
        try {
            channel.publish(getSession(), data, null);
            log.info("publish the channel");
        }
        catch (Exception e){
            System.out.println(" Exception = " + e);
        }
        System.out.println("Session subscriptions :" + sender.getServerSession().getSubscriptions());
        System.out.println("Listeners on /service/out = "+bayeuxServer.getChannel("/service/out").getListeners().toString());
        System.out.println("Subscribers on /service/out = "+bayeuxServer.getChannel("/service/out").getSubscribers().toString());

        publishData(data);
    }

    public ServerSession getSession() {
        return session;
    }

    public void setSession(ServerSession session) {
        this.session = session;
    }

    private void publishData(Map<String, Object> data) {
        System.out.println("Published data = " + data);
    }
    /* @org.cometd.annotation.Subscription("/service/out")*/
   @Subscription("/service/out")
    public void echo(Message message)
    {
        System.out.println("Echo service published " + message);
    }

    public String getDetails() {
        return details;
    }

    public void setDetails(String details) {
        this.details = details;
    }
}