Unable to broadcast to single connection using Atmosphere runtime

Question

I am using Atmosphere runtime 0.6 Snapshot. Tomcat 7 is logging correctly that I am using the Http11 Nio connector and there is no warning that BlockingIO will be used.

I am trying to send messages to three kinds of channels.

1. Global Broadcaster - broadcast to all suspended resources. (All)
2. Broadcast to a particular resource (say, Partner)
3. Broadcast to current resource (Self)

When a login action occurs, what all do I have to store in session in order to achieve this kind of broadcasting?

Some details of my code are as follows:

1. My Handler implements AtmosphereHandler

2. In the constructor, I instantiate the globalBroadcaster as follows:

   globalBroadcaster = new DefaultBroadcaster();

3. On login,

   resource.getAtmosphereConfig().getServletContext().setAttribute(name, selfBroadcaster); where name is the user name from request parameter and selfBroadcaster is a new instance of DefaultBroadcaster.

4. Here is the code for sendMessageToPartner,

private synchronized void sendMessageToPartner(Broadcaster selfBroadcaster, AtmosphereResource<HttpServletRequest, HttpServletResponse> resource,String name, String message) {
// this gives the partner's name
String partner= (String) resource.getAtmosphereConfig().getServletContext().getAttribute(name + PARTNER_NAME_TOKEN);
// get partner's broadcaster
Broadcaster outsiderBroadcaster = (Broadcaster) resource
.getAtmosphereConfig().getServletContext()
.getAttribute(partner);
if (outsiderBroadcaster == null) {
sendMessage(selfBroadcaster, "Invalid user " + partner);
return;
}
// broadcast to partner
outsiderBroadcaster.broadcast(" **" + message);

I hope I have given all the required information. I can provide more information if required.

The problem is, the global message gets sent. When message to partner is sent, sometimes it gets blocked, the message is not received in the client at all. This happens consistently after 3-4 messages.

Is there some threading problem? What am I doing wrong?

I hope somebody helps me out with this.

Answer 1

Ok, I figured out how this can be achieved with Atmosphere runtime. First, I upgraded to 0.7 SNAPSHOT, but I think the same logic would work with 0.6 as well.

So, to create a broadcaster for a single user:

In GET request,

  // Use one Broadcaster per AtmosphereResource             
try {       
atmoResource.setBroadcaster(BroadcasterFactory.getDefault().get());     

} catch (Throwable t) {
                throw new IOException(t);
            }

            // Create a Broadcaster based on this session id.
            selfBroadcaster = atmoResource.getBroadcaster();
            // add to the selfBroadcaster
            selfBroadcaster.addAtmosphereResource(atmoResource);

            atmoResource.suspend();   

When login action is invoked,

//Get this broadcaster from session and add it to BroadcasterFactory.

Broadcaster selfBroadcaster = (Broadcaster) session.getAttribute(sessionId);

BroadcasterFactory.getDefault().add(selfBroadcaster, name);

Now the global broadcaster. The logic here is, you create a broadcaster from the first resource and then add each resource as they log in.

Broadcaster globalBroadcaster;

globalBroadcaster = BroadcasterFactory.getDefault().lookup(DefaultBroadcaster.class, GLOBAL_TOKEN, false);
                if (globalBroadcaster == null) {
                  globalBroadcaster = selfBroadcaster;

                    } else {
                        BroadcasterFactory.getDefault().remove(
                                globalBroadcaster, GLOBAL_TOKEN);
                        AtmosphereResource r = (AtmosphereResource) session
                                .getAttribute("atmoResource");
                        globalBroadcaster.addAtmosphereResource(r);

                    }
                    BroadcasterFactory.getDefault().add(globalBroadcaster,
                            GLOBAL_TOKEN);

Finally, you can broadcast to Single connection or Globally to all connections as follows:

// Single Connection/Session
Broadcaster singleBroadcaster= BroadcasterFactory.getDefault().lookup(
                            DefaultBroadcaster.class, name);
singleBroadcaster.broadcast("Only for you");

// Global 
Broadcaster globalBroadcaster = BroadcasterFactory.getDefault().lookup(DefaultBroadcaster.class,GLOBAL_TOKEN, false);
globalBroadcaster.broadcast("Global message to all");

To send message to partner, just lookup the broadcaster for the partner and do the same as above for single connection.

Hope this helps someone who tries to achieve the same. There may be better ways of doing this. I think I will have to use this approach until someone suggests a better solution.