Pub/Sub duplicate messages

[canhduong28]

Hi,

I'm writing a small application which allows users to publish/subscribe multi channels at a time and I chose Tornado-Redis to implement this feature. The scenario is that one message would be sent to multi channels, but if a user who subscribed all these channels, then he/she should get the message one time only. Right now, users keep receiving duplicate messages from the Pub/Sub handler.

[Solutions]: I have an idea that I will override the SockJSSubscriber class to use a bloom filter to avoid duplicating messages sent over this object. If you guys have any better idea, just let me know. I would highly appreciate that.

Thanks, -Canh

[leporo]

Hi Canh,

Your idea seem fine for me.

Here are my thoughts:

• Please make sure there is no way to avoid publishing the same message to multiple channels.
• The best place to add the filter is right before broadcast method call:

            broadcasters = list(self.subscribers[msg.channel].keys())
            if broadcasters:
                broadcasters[0].broadcast(broadcasters, str(msg.body))

• The broadcasters variable in fact contains the list of client connection handlers subscribed to redis channel (I'll rename it to subscribers to avoid confusion);
• I think it's better to assign an unique ID to each SockJSConnection (subscriber) instance. I guess you'd like to use both message and subscriber ID to filter the list of subscribers before broadcating the message;
• I think you could use bloom filter rotation to keep the false positives rate low. I mean you may create a new bloom filter each N messages and use two most recent ones.

Thanks for your idea.

It would be great if you share your implementation when you have it :)

[canhduong28]

Hi Vlad,

Thank for you suggestions, I highly appreciate that.

However, I came up with a solution that I will override the broadcast method of SockJSConnection class.

Implementation code would look like that:

class SubscriptionHandler(sockjs.tornado.SockJSConnection):

def __init__(self, *args, **kwargs):
    super(SubscriptionHandler, self).__init__(*args, **kwargs)
    self.bloom_filter = pyreBloom.pyreBloom(str(uuid.uuid4()), 10000, 0.01)

def broadcast(self, clients, message):
    msg_id = utils.get_msgId(message)
    if not self.bloom_filter.contains(msg_id):
        self.session.broadcast(clients, message)
        self.bloom_filter.add(msg_id)

The bloom filter serves only when the session is activated, it will be deleted when connection is closed. So this implementation is quite basic. But it worked for me and I'm happy with that.

Thank you all and happy coding :)

[leporo]

Thank you for sharing your code. I highly appreciate that.

Do you mind if I share my thoughts about it?

• This way you allocate a ~28K bitmap redis key for each client connection. I think it would be more efficient to use the same message filter for all client connections and use the message + client ID pair for filtering;
• You use the first connection instance in subscriber list for filtering. For two clients: client1 (subscribed to channel A) and client2 (subscribed to channels A and B), assuming client2 connects AFTER client1, client2 will be receiving duplicates of messages sent to both channels.

Thanks again! It was fun :)

[canhduong28]

Thank you for sharing your thoughts, I highly appreciate that.

1. I agreed that using message + client ID pair will be more efficient. I will implement this approach to improve the process.
2. I already knew that. This is why I chose the SockJSConnection to override instead of SockJSSubscriber :)

It was really fun for me too.