Merge pull request #86 from Asmod4n/master

Added EventStream example

examples/server_sent_events.rb

@@ -0,0 +1,115 @@
+#!/usr/bin/env ruby
+# Run with: bundle exec examples/server_sent_events.rb
+# Test with: curl -vNH 'Accept: text/event-stream' -H 'Last-Event-ID: 1' -H 'Cache-Control: no-cache' http://localhost:63310
+
+require 'bundler/setup'
+require 'time'
+require 'reel'
+
+class ServerSentEvents < Reel::Server
+  include Celluloid::Logger
+
+  def initialize(ip = '127.0.0.1', port = 63310)
+    @connections = []
+    @history = []
+    @lastMessageId = 0
+    async.ping
+    async.ring
+    super(ip, port, &method(:on_connection))
+  end
+
+  def broadcast(event, data)
+    #only keep the last 5000 Events
+    if @history.size >= 6000
+      @history.slice!(0, @history.size - 1000)
+    end
+    @lastMessageId += 1
+    @history << {id: @lastMessageId, event: event, data: data}
+    info "Sending Event: #{event} Data: #{data} to #{@connections.count} Clients"
+    @connections.each do |socket|
+      async.send_sse(socket, data, event, @lastMessageId)
+    end
+    true
+  end
+
+  private
+  #event and id are optional
+  def send_sse(socket, data, event = nil, id = nil)
+    begin
+      socket.id id if id
+      socket.event event if event
+      socket.data data
+    rescue IOError, Errno::ECONNRESET, Errno::EPIPE
+      @connections.delete(socket)
+    end
+  end
+
+  def send_ping
+    @connections.each do |socket|
+      begin
+        #Lines that start with a Colon are Comments and will be ignored
+        socket << ":\n"
+      rescue IOError, Errno::ECONNRESET, Errno::EPIPE
+        @connections.delete(socket)
+      end
+    end
+  end
+
+  def ping
+    #apache 2.2 closes connections after five seconds when nothing is send
+    every(5) do
+      send_ping
+    end
+  end
+
+  def ring
+    every(2) do
+      broadcast(:time, Time.now.httpdate)
+    end
+  end
+
+  def handle_request(request)
+    query = {}
+    (request.query_string || '').split('&').each do |kv|
+      key, value = kv.split('=')
+      if key && value
+        key, value = CGI.unescape(key), CGI.unescape(value)
+        query[key] = value
+      end
+    end
+    body = Reel::EventStream.new do |socket|
+      @connections << socket
+      socket.retry 5000
+      #after a Connection reset resend newer Messages to the Client
+      if @history.count > 0 && id = (request.headers['Last-Event-ID'] || query['lastEventId'])
+        begin
+          id = Integer(id)
+          if history = @history.select {|h| h[:id] >= id}.map {|a| "id: %d\nevent: %s\ndata: %s" % [a[:id], a[:event], a[:data]]}.join("\n\n")
+            socket << "%s\n\n" % [history]
+          else
+            socket << "id\n\n"
+          end
+        rescue ArgumentError, IOError, Errno::ECONNRESET, Errno::EPIPE
+          @connections.delete(socket)
+          request.close
+        end
+      else
+        socket << "id\n\n"
+      end
+    end
+    #X-Accel-Buffering is nginx(?) specific. Setting this to "no" will allow unbuffered responses suitable for Comet and HTTP streaming applications
+    request.respond Reel::StreamResponse.new(:ok, {
+      'Content-Type' => 'text/event-stream; charset=utf-8',
+      'Cache-Control' => 'no-cache',
+      'X-Accel-Buffering' => 'no',
+      'Access-Control-Allow-Origin' => '*'}, body)
+  end
+
+  def on_connection(connection)
+    connection.each_request do |request|
+      handle_request(request)
+    end
+  end
+end
+
+ServerSentEvents.run