JMS, XMPP, ... (via AbstractBroadcasterProxy) : Broadcast to a specific resource

[jimmyjim]

Hi

I'm using atmosphere to send messages to some gwt clients from the server (tomcat7).

So i call the method broadcaster.broadcast( String message, AtmosphereResource resourceA );

    Using atmosphere with defaultBroadcaster : everything is ok.

    But if i switch to jmsBroadcaster (activeMq ), the message is delivered to all ressources of the broadcaster and not only the resourceA.

Has someone already test this method with jmsBroadcaster ?

Here is my conf.

<!-- ********************************************************************************************************************* -->
    <!-- SERVLETS Atmosphere -->
    <!-- ********************************************************************************************************************* -->

    <servlet>
        <description>AtmosphereServlet</description>
        <servlet-name>AtmosphereServlet</servlet-name>
        <servlet-class>org.atmosphere.cpr.AtmosphereServlet</servlet-class>
        <!-- If you want to use Servlet 3.0 -->
        <async-supported>true</async-supported>
        <init-param>
            <!-- prevent deadlocks -->
            <param-name>org.atmosphere.disableOnStateEvent</param-name>
            <param-value>true</param-value>
        </init-param>

        <!-- By default, the Atmosphere Framework doesn't create sessions and isn't using them. If your application uses HttpSession, you need to tell Atmosphere to track the session as well by enabling it, 
            in web/application.xml: -->
        <!-- This is specially important if you are using WebSocket because of the difference of WebServer implementation, the framework will makes sure it works uniformly with all of them. -->
        <init-param>
            <param-name>org.atmosphere.cpr.sessionSupport</param-name>
            <param-value>true</param-value>
        </init-param>

        <!-- By default, Atmosphere is using the DefaultBroadcaster, which only broadcast events in memory. Enabling one of the Broadcaster described below allow servers/nodes to communicate events between them, 
            e.g when a broadcast operation occurs in on server, the event will also be distributed to other servers. The available Broadcasters are: -->

        <!--JMSBroadcaster : allow events to be broadcasted using JMS implementation like ActiveMQ. -->
        <!--XMPPBroadcaster : Use the XMPP protocol to broadcast events across multiple server. For example, a GMail account can be used to distribute events (because the GMail chat supports XMPP) -->
        <!--HazelcastBroadcaster : allow events to be broadcasted using the Hazelcast framework. -->
        <!--RedisBroadcaster : allow events to be broadcasted using the Redis pubsub API. -->
        <!--JGroupsBroadcaster : allow events to be broadcasted using the JGroups framework. -->
        -->
        <init-param>
             <param-name>org.atmosphere.cpr.broadcasterClassName</param-name>
             <param-value>org.atmosphere.plugin.jms.JMSBroadcaster</param-value>
        </init-param>

        <init-param>     
            <param-name>org.atmosphere.plugin.jms.JMSBroadcaster.JNDINamespace</param-name>
            <param-value>java:comp/env/</param-value>
        </init-param>

        <init-param>
            <param-name>org.atmosphere.plugin.jms.JMSBroadcaster.JNDIConnectionFactoryName</param-name>
            <param-value>jms/atmosphereFactory</param-value>
        </init-param>

        <init-param>
            <param-name>org.atmosphere.plugin.jms.JMSBroadcaster.JNDITopic</param-name>
            <param-value>jms/atmosphereTopic</param-value>
        </init-param>

        <load-on-startup>1</load-on-startup>
    </servlet>

[jimmyjim]

Hi

to workaround,
Is it a good alternative to create a temp broadcaster with the resource and remove it like

                   final Broadcaster tempBroadcaster = BroadcasterFactory.getDefault().lookup( UUID.randomUUID(), true );

                    tempBroadcaster.setBroadcasterLifeCyclePolicy( BroadcasterLifeCyclePolicy.EMPTY_DESTROY );
                    tempBroadcaster.addAtmosphereResource( resource );
                    tempBroadcaster.broadcast(message);
                    tempBroadcaster.addBroadcasterListener( new BroadcasterListener() {

                      @Override
                      public void onPreDestroy( Broadcaster b ) {
                      }

                      @Override
                      public void onPostCreate( Broadcaster b ) {
                      }

                      @Override
                      public void onComplete( Broadcaster b ) {
                        System.out.println( "remove temp broadcaster " );
                        tempBroadcaster.removeAtmosphereResource( resource );
                      }
                    } );

What do you think about ?

[jfarcand]

No, that will create too many instance of Broadcaster. Instead, write a BroadcasterFilter and set the AtmosphereResource you want to discard before calling broadcast.

The fix will be in 1.1, most probably during the week of Jan 23

[jimmyjim]

Hi,

Sorry I've already tested using a PerRequestBroadcastFilter. Unfortunately it does'nt work either.

 Broadcaster broadcasterAlerte = getBroadCaster();
      PerRequestBroadcastFilter filter = new PerRequestBroadcastFilter() {

      @Override
      public BroadcastAction filter( Object originalMessage, Object message )
      {
      return new BroadcastAction(message );
      }

      @Override
      public BroadcastAction filter( AtmosphereResource atmosphereResource,
      Object originalMessage, Object message ) {

        if (message!=null && message.toString().equals( "hello" ) ) {
          System.out.println( "continue :" + atmosphereResource );
          return new BroadcastAction( ACTION.CONTINUE, message );
          }
          else {
          System.out.println( "abort :" + atmosphereResource );
          return new BroadcastAction( ACTION.ABORT, message );   

      }
      } };
      broadcasterAlerte.getBroadcasterConfig().addFilter(filter);
      broadcasterAlerte.broadcast("hello");
      broadcasterAlerte.getBroadcasterConfig().removeFilter( filter);

The output is :
continue : hello

But the resource doesn't receive the message.

Thanks a lot. Have a nice day.
No problem i'm waiting for the version 1.1

[jfarcand]

Ok, I won't fix that one as it requires to add some protocol on top of the communication channel used. The workaround is to broadcast the list of UUIDs to all node in the cluster, and use a BroadcastFilter to filter the one that needs to be invoked. I will add this issue to the FAQ.

[abin103]

How do PrivateMessag send ？

[jfarcand]

@abin103 You need to implements it yourself, as I described it many time on the mailing list. It not that compicated, I did it for Atmosphere Satellite What you need is add some kind of protocol and replicate the state of Atmosphere's uuid amongs the cluster.