Clarification of JedisPubSub and multiple subscriptions

[pablochacin]

Hi

I've started implementing something with Jedis pubsub and there's something that that puzzles me:
How could I add more channel/patterns to a jedis connection on which I already did a subscribe/psubscribe?
For example, supose I subscribe to channel news.sports and later want to subscribe, over the same connection
to news.wheather. Is that possible? What's the proper way to do so? As I understand the connection is blocked
during the subscription. Should I do it from another thread? it's that safe?

The JedisPubSub class defines methods like onSubscribe and onPUnsubscribe, what made me believe that
they were a way to notify of such subscriptions. If not, what's the intention of these methods?

many thanks in advance

[Jonathan Leibiusky]

You shouldn't do it from a different thread.

You could send a message to the channel saying it to subscribe to another channel.

But I think the pubsub API should be reimplemented to be thread safe, so it is easier to do this kind of stuff.

[Pablo]

Jonathan

What really puzzles me is this paragraph from documentation:

 You can call subscribe or psubscribe on an existing JedisPubSub instance to change your subscriptions.

My understanding is that if I have

 class Listener extends JedisPubSub {
 }

 JedisPubSub listener = new Listener();

If in a thread T1 I do this:

jedis.subscribe(listener,channel);

Thread T1 will block. However, according to the documentation, I could do in another Thread T2

listener.subscribe(anotherchannel);

And then listener will start receiving notifications for the new method.

Am I right or completely lost?

 
Thanks

[Jonathan Leibiusky]

You can do that. But as it is not thread safe you could run into strange situations and errors. I would avoid doing that, unless you use some king of synchronisation between threads, which is painful usually.

I think that PubSub API in Jedis needs some work.

[Pablo]

On 08/21/2012 03:59 PM, Jonathan Leibiusky wrote:
> I think that PubSub API in Jedis needs some work.

Agree.

Thanks

[pablochacin]

I manage to make this to work (still ironing out some synchronization details), but now I need to find a way to gracefully shutdown the thread that issues the subscribe command.

I implemented a class that runs in its own thread:

public class SubscriptionHandler extends JedisPubSub implements Runnable {

This class has a simple run method on which it subscribes to an initial channel (see explanation below) as follows:

public void run(){
   
    try{
      connection.subscribe("control.channel");
    } catch(JedisException e){
            //Handle exceptions
    }
}

To subscribe to additional channels, I added a method:

public void subscribeToChannel(String channel){
      super.subscribe(channel);
}

The control channel can be used to send control commands to this thread (like, for example, to stop subscriptions), which are then processed in the onMessage method

 public void onMessage(String channel, String message){

     if(channel.equals("control.channel")){
         if(command.equals("stop")){
             unsubscribe();   //terminate all subscriptions
        }
        connection.quit();
    }
  }

My concrete question is, how can I make the subscribe in the run method to return and allow the thread to finish?
I  tried doing a quit on the connection, as shown above, but it generates an exception:

java.lang.ClassCastException: java.util.ArrayList cannot be cast to [B
    at redis.clients.jedis.

Connection.getStatusCodeReply(Connection.java:162)
    at redis.clients.jedis.Jedis.quit(Jedis.java:77)


Thanks in advance

[Jonathan Leibiusky]

You should unsubscribe(). By not specifying a channel, redis will unsubscribe from all and send you an UNSUBSCRIBED messages for each. Once you have unsubscribed from all of them (or actually you can just check for the control channel, you can safely disconnect.

[pablochacin]

Thanks, that did the trick.

Thanks

[ravi kumar]

Hi,

 Can you please post example which is working...  Thank you in advance..


Ravikumar Kotta
8050942484

[peter kristensen]

Hi.

 

I know this is an old thread, but since I found myself in the same situation just now, I figured that it wouldn't hurt to give you my solution.

private class SubscriberThread extends Thread{

    private Jedis sub = new Jedis("localhost");

    @Override
    public void run() {
        JedisPubSub pubSub = new JedisPubSub() {

            {
                sub.psubscribe(this, "__keyevent@0__:*");
            }

            @Override

            public void onMessage(String channel, String message) {

                logger.log("pubSubOnMessage(" + channel + ", " + message + ")");
                try {
                    handleMessage(message);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }

            @Override
            public void onPMessage(String pattern, String channel, String message) {
                logger.log("pubSubOnPMessage(" + pattern + ", " + channel + ", " + message + ")");
            }
        };
        sub.subscribe(pubSub, ipAndPort, "global");
    }
}

As you see, I wanted to use both "SUBSCRIBE" and "PSUBSCRIBE". So what I did was to put the second call (psubscribe) inside an initializer.

I hope this can help someone.