Merge pull request #32 from mgartner/no-data-callback

Call-back for when no data is received for 90 seconds

lib/twitter/json_stream.rb

@@ -20,6 +20,8 @@ class JSONStream < EventMachine::Connection
     RECONNECT_MAX   = 320
     RETRIES_MAX     = 10
 
+    NO_DATA_TIMEOUT = 90
+
     DEFAULT_OPTIONS = {
       :method         => 'GET',
       :path           => '/',
@@ -44,6 +46,7 @@ class JSONStream < EventMachine::Connection
     attr_accessor :nf_last_reconnect
     attr_accessor :af_last_reconnect
     attr_accessor :reconnect_retries
+    attr_accessor :last_data_received_at
     attr_accessor :proxy
 
     def self.connect options = {}
@@ -72,6 +75,7 @@ def initialize options = {}
       @immediate_reconnect = false
       @on_inited_callback = options.delete(:on_inited)
       @proxy = URI.parse(options[:proxy]) if options[:proxy]
+      @last_data_received_at = nil
     end
 
     def each_item &block
@@ -86,6 +90,10 @@ def on_reconnect &block
       @reconnect_callback = block
     end
 
+    def on_no_data &block
+      @no_data_callback = block
+    end
+
     def on_max_reconnects &block
       @max_reconnects_callback = block
     end
@@ -111,12 +119,13 @@ def unbind
       end
       schedule_reconnect if @options[:auto_reconnect] && !@gracefully_closed
       @close_callback.call if @close_callback
-
+      @state = :init
     end
 
     # Receives raw data from the HTTP connection and pushes it into the
     # HTTP parser which then drives subsequent callbacks.
     def receive_data(data)
+      @last_data_received_at = Time.now
       @parser << data
     end
 
@@ -128,9 +137,20 @@ def connection_completed
     def post_init
       reset_state
       @on_inited_callback.call if @on_inited_callback
+      @reconnect_timer = EventMachine.add_periodic_timer(5) do
+        if @gracefully_closed
+          @reconnect_timer.cancel
+        elsif @last_data_received_at && Time.now - @last_data_received_at > NO_DATA_TIMEOUT
+          no_data
+        end
+      end
     end
 
   protected
+    def no_data
+      @no_data_callback.call if @no_data_callback
+    end
+
     def schedule_reconnect
       timeout = reconnect_timeout
       @reconnect_retries += 1

spec/twitter/json_stream_spec.rb

@@ -230,6 +230,22 @@ def receive_data data
     it_should_behave_like "network failure"
   end
 
+  context "on no data received" do
+    attr_reader :stream
+    before :each do
+      $data_to_send = ''
+      $close_connection = false
+    end
+
+    it "should call no data callback after no data received for 90 seconds" do
+      connect_stream :stop_in => 6 do
+        stream.last_data_received_at = Time.now - 88
+        stream.should_receive(:no_data).once
+      end
+    end
+
+  end
+
   context "on server unavailable" do
 
     attr_reader :stream