Do you need a Push Notification Manager? – Redis PubSub to the rescue

Happy New Year!

And its indeed an awesome start to a surely prosperous year — my stunts with Redis PubSub worked ! We have been looking at getting a push notification manager in place to communicate between applications! Here is what we wanted to do:

• Applications should be able to subscribe for notifications.
• Notification wildcards should be supported!
• Applications should be able to publish notifications to subscribers.
• These should push notifications!
• Guaranteed delivery of notifications i.e. Applications should get pending messages
• Sequential delivery of notifications — even for the backlog.

Well, initially I went searching for various solutions – almost all pointed to EventMachine, Websockets & Cramp. When I came across Redis pubsub, it fit like a glove but it has the following drawbacks:

• Its a lightweight non-persistent PubSub.
• Missing or unavailable subscribers don’t get notifications.

I came across this mail thread on the Redis Google Group and realized that I would have to implement this. First a little background on Redis pubsub. For those not familiar with Redis, I would strongly recommend getting familiar with it :). Redis is a nosql based memory datastore which supports persistence asynchronously – in short, best of both worlds – speed and consistency. After installing the redis-server, I added ‘redis’ to the Gemfile and I was on my way.

Redis Pubsub supports subscriptions in 2 commands: subscribe and psubscribe, the latter supports wildcards – that is exactly what I wanted. A subscription block is simple:

uri = URI.parse(ENV["redis://127.0.0.1:6379"])
REDIS = Redis.new(:host => uri.host, :port => uri.port, :password => uri.password)

REDIS.psubscribe('my.first.event', 'basic.events.*', 'event.*' ) do |on|
 on.psubscribe do |event, total|
    Rails.logger.info("Subscribed to ##{event} (#{total} subscriptions)")
 end

 on.pmessage do |pattern, event, message|
    # Message processing here
 end

 on.punsubscribe do |event, total|
   Rails.logger.info("Unsubscribed for ##{event} (#{total} subscriptions)")
 end
end

The block parameter ‘on’ is a Subscription object, which registers the callbacks for ‘psubscribe’, ‘pmessage’ and ‘punsubscribe’. Some caveates:

• If you use wildcards, remember to not subscribe to the same event more than once. You will receive the event twice. Eg. If you subscribe to ‘event.one’ and ‘event.*’, the callback will be invoked twice!
• psubscribe is a blocking call. Remember NOT to use the REDIS connection for issuing any other commands otherwise it will throw an exception.
• Since its a blocking call, its recommended to run this in a separate process (like DelayedJob) or a Thread at least, otherwise your Rails server will get blocked!

Now, my problem was that I also need to persist the events during a publish and process pending messages. To achieve this, I created a custom Redis Client that handles a backlog show below:

require 'redis'
require 'multi_json'

class PubSubRedis < Redis

  def initialize(options = {})
    @timestamp = options[:timestamp].to_i || 0 # 0 means -- no backlog needed
    super
  end

  # Add each event to a Sorted Set with the timestamp as the score
  def publish(channel, message)
    timestamp = Time.now.to_i
    zadd(channel, timestamp, MultiJson.encode([channel, message]))
    super(channel, MultiJson.encode(message))
  end

  # returns the backlog of pending messages [ event, payload ] pairs
  # We do a union of sorted sets because we need to support wild-card channels.
  def backlog(channels, &block)
    return if @timestamp == 0

    # Collect the entire set of events with wild-card support.
    events = channels.collect {|e| keys(e)}.flatten

    return if not events or events.empty? # no events to process

    destination = "backlog-#{Time.now.to_i}"
    zunionstore(destination, events)
    # We want events only after the timestamp so add the (. This ensures that
    # an event with this timestamp will not be sent.
    # TODO: We may have a condition where, multiple events for the same timestamp
    # may be recorded but will be missed out because of the (.
    messages = zrangebyscore(destination, "(#{@timestamp.to_s}", "+inf")

    messages.each do |message|
      event, payload = MultiJson.decode(message)
      block.call(event, payload)
    end

    # cleanup
    del(destination)

  end
end

The only thing left to do now is tie up loose ends. Its necessary to store the timestamp of the events received – what better store than Redis? I passed the timestamp as a variable to the custom Redis Client. We process the backlog first before invoking subscribe. Here is the snippet:

# config/initializer/redis.rb
LAST_TIMESTAMP='notification_timestamp_42037933' # some random key

uri = URI.parse('redis://127.0.0.1:6379')
# Publishing channel, since the REDIS_SUB will block!
REDIS_PUB = PubSubRedis.new(:host => uri.host, :port => uri.port, :password => uri.password)
REDIS_SUB = PubSubRedis.new(:host => uri.host, :port => uri.port, :password => uri.password,
               :timestamp => REDIS.get(LAST_TIMESTAMP).to_i )

# File where you are processing the backlog (DelayedJob or a separate thread)
REDIS_SUB.backlog('my.first.event', 'basic.events.*', 'event.*') do |event, payload|
  # backlog notification processing
end

# callback processing loop shown earlier.
REDIS_SUB.psubscribe('my.first.event', 'basic.events.*', 'event.*' ) do |on|
 ...
 on.pmessage do |pattern, event, message|
   REDIS_PUB.set(LAST_TIMESTAMP, Time.now.to_i)

   # callback processing
 end
 ...
end

Voila! My next post is going to be how I set this up on Heroku using Redistogo. Stay tuned.

I hope you found this article valuable. Feel free to ask questions and give feedback in the comments section of this post. Thanks!