lib/angelo/base.rb in angelo-0.1.14 vs lib/angelo/base.rb in angelo-0.1.15

@@ -4,17 +4,20 @@
     include ParamsParser
     include Celluloid::Logger
 
     extend Forwardable
     def_delegators :@responder, :content_type, :headers, :redirect, :request
+    def_delegators :@klass, :websockets, :sses, :sse_event, :sse_message
 
     @@addr = DEFAULT_ADDR
     @@port = DEFAULT_PORT
 
     @@ping_time = DEFAULT_PING_TIME
     @@log_level = DEFAULT_LOG_LEVEL
 
+    @@report_errors = false
+
     if ARGV.any? and not Kernel.const_defined?('Minitest')
       require 'optparse'
       OptionParser.new { |op|
         op.on('-p port',   'set the port (default is 4567)')      { |val| @@port = Integer(val) }
         op.on('-o addr',   "set the host (default is #{@@addr})") { |val| @@addr = val }
@@ -85,27 +88,37 @@
           routes[m][path] = Responder.new &block
         end
       end
 
       def websocket path, &block
-        routes[:websocket][path] = WebsocketResponder.new &block
+        routes[:websocket][path] = Responder::Websocket.new &block
       end
 
+      def eventsource path, &block
+        routes[:get][path] = Responder::Eventsource.new &block
+      end
+
       def on_pong &block
-        WebsocketResponder.on_pong = block
+        Responder::Websocket.on_pong = block
       end
 
       def task name, &block
         Angelo::Server.define_task name, &block
       end
 
       def websockets
-        @websockets ||= Stash.new server
+        @websockets ||= Stash::Websocket.new server
         @websockets.reject! &:closed?
         @websockets
       end
 
+      def sses
+        @sses ||= Stash::SSE.new server
+        @sses.reject! &:closed?
+        @sses
+      end
+
       def content_type type
         Responder.content_type type
       end
 
       def run addr = @@addr, port = @@port
@@ -124,12 +137,27 @@
           lp = File.join(public_dir, path)
           File.file?(lp) ? lp : nil
         end
       end
 
+      def sse_event event_name, data
+        data = data.to_json if Hash === data
+        SSE_EVENT_TEMPLATE % [event_name.to_s, data]
+      end
+
+      def sse_message data
+        data = data.to_json if Hash === data
+        SSE_DATA_TEMPLATE % data
+      end
+
     end
 
+    def initialize responder
+      @responder = responder
+      @klass = self.class
+    end
+
     def async meth, *args
       self.class.server.async.__send__ meth, *args
     end
 
     def future meth, *args
@@ -142,12 +170,10 @@
                   when POST; parse_post_body
                   when PUT;  parse_post_body
                   end
     end
 
-    def websockets; self.class.websockets; end
-
     def request_headers
       @request_headers ||= Hash.new do |hash, key|
         if Symbol === key
           k = key.to_s.upcase
           k.gsub! UNDERSCORE, DASH
@@ -173,10 +199,23 @@
           ws.socket << ::WebSocket::Message.ping.to_data
         end
       end
     end
 
+    task :handle_event_source do |socket, block|
+      begin
+        block[socket]
+      rescue Reel::SocketError, IOError, SystemCallError => e
+        # probably closed on client
+        warn e.message if @@report_errors
+        socket.close unless socket.closed?
+      rescue => e
+        error e.inspect
+        socket.close unless socket.closed?
+      end
+    end
+
     def halt status = 400, body = ''
       throw :halt, HALT_STRUCT.new(status, body)
     end
 
     def send_file local_file, opts = {}
@@ -197,9 +236,23 @@
       # Content-Length
       #
       headers CONTENT_LENGTH_HEADER_KEY => File.size(lp)
 
       halt 200, File.read(lp)
+    end
+
+    def eventsource &block
+      headers SSE_HEADER
+      async :handle_event_source, responder.connection.detach.socket, block
+      halt 200, :sse
+    end
+
+    def report_errors?
+      @@report_errors
+    end
+
+    def sleep time
+      Celluloid.sleep time
     end
 
   end
 
 end