Server-Sent Event with Last-Event-ID

Author: nono

lib/board-linuxfr.rb

@@ -6,20 +6,24 @@
 class BoardLinuxfr < Goliath::API
   plugin RedisPlugin
 
-  def initialize
-    Redis.new
-  end
-
   def response(env)
-    env.logger.debug "New client"
-    env['subscription'] = status[:channel].subscribe do |msg|
-      env.stream_send("data: #{msg}\n\n")
-    end
+    env.logger.debug "New client: #{env['PATH_INFO']}"
+    send_msg = ->(msg) {
+      env.logger.debug " -> #{msg}"
+      env.stream_send("data: #{msg}\nid: #{msg[:id]}\n\n")
+    }
+    event_id     = env['HTTP_LAST_EVENT_ID']
+    chan_name    = env['PATH_INFO'].delete('/b/')
+    env['cache'] = status[:cache][chan_name]
+    env['chan']  = status[:channels][chan_name]
+    env['sid']   = env['chan'].subscribe &send_msg
+    env['timer'] = EM.add_periodic_timer(15) { env.stream_send "::\n\n" }
+    EM.next_tick { env['cache'].from(event_id, &send_msg) } if event_id
     streaming_response(200, {'Content-Type' => 'text/event-stream'})
   end
 
-  def on_close
-    return unless env['subscription']
-    status[:channel].unsubscribe(env['subscription'])
+  def on_close(env)
+    env['chan'].unsubscribe env['sid'] if env['sid']
+    env['timer'].cancel                if env['timer']
   end
 end

lib/board-linuxfr/cache.rb

@@ -0,0 +1,38 @@
+require "goliath/api"
+
+
+class BoardLinuxfr < Goliath::API
+  class Cache
+    CAPACITY = 10
+
+    class Entry < Array
+      def push(item)
+        super item
+        shift if size > CAPACITY
+      end
+
+      def from(id, &blk)
+        found = false
+        each do |item|
+          blk.call item if found
+          found ||= item[:id] == id
+        end
+      end
+    end
+
+    def initialize
+      @keys = Entry.new
+      @vals = Entry.new
+    end
+
+    def [](key)
+      n = @keys.index(key)
+      unless n
+        @keys.push key
+        @vals.push Entry.new
+        n = @keys.length - 1
+      end
+      @vals[n]
+    end
+  end
+end

lib/board-linuxfr/redis_plugin.rb

@@ -2,15 +2,18 @@
 require "em-synchrony"
 require "redis"
 require "redis/connection/synchrony"
+require "board-linuxfr/cache"
 
 
 class BoardLinuxfr < Goliath::API
   class RedisPlugin
+
     def initialize(port, config, status, logger)
       logger.info "Initializing the Redis plugin"
-      @logger  = logger
-      @channel = status[:channel] = EM::Channel.new
-      @redis   = Redis.new
+      @logger = logger
+      @chans  = status[:channels] = Hash.new { |h,k| h[k] = EM::Channel.new }
+      @cache  = status[:cache]    = Cache.new
+      @redis  = Redis.new
     end
 
     def run
@@ -22,8 +25,10 @@ def run
 
           on.pmessage do |pattern, chan, msg|
             _, chan, id, kind = *chan.split('/')
-            @channel.push(chan: chan, id: id, kind: kind, msg: msg)
-            @logger.info "New message: #{chan} #{id} #{kind} #{msg}"
+            @logger.info "New message: [#{chan}] #{id}. #{kind}> #{msg}"
+            [@chans, @cache].each do |storage|
+              storage[chan].push(id: id, kind: kind, msg: msg)
+            end
           end
 
           on.punsubscribe do |pattern, total|