Fixed timing bug in batch publishing, fixes #4

burtcorp · Jun 18, 2012 · cf54d36 · cf54d36
1 parent a5a734e
commit cf54d36
Show file tree

Hide file tree

Showing 2 changed files with 6 additions and 3 deletions.
diff --git a/lib/autobahn/publisher.rb b/lib/autobahn/publisher.rb
@@ -55,11 +55,12 @@ def start!
       if @batch_options[:timeout] && @batch_options[:timeout] > 0
         @scheduler = Concurrency::Executors.new_single_thread_scheduled_executor(Concurrency::NamingDaemonThreadFactory.new('batch_drainer'))
         @drainer_task = @scheduler.schedule_with_fixed_delay(
-          method(:force_drain).to_proc, 
+          method(:maybe_force_drain).to_proc, 
           @batch_options[:timeout], 
           @batch_options[:timeout], 
           Concurrency::TimeUnit::SECONDS
         )
+        @last_drain = Time.now
       end
       self
     end
@@ -101,9 +102,11 @@ def drain_batch
         end
       end
       @publisher.publish(batch)
+      @last_drain = Time.now
     end
 
-    def force_drain
+    def maybe_force_drain
+      return if (Time.now - @last_drain) < @batch_options[:timeout]
       drain
       drain_batch
     end

diff --git a/spec/integration/autobahn_intg_spec.rb b/spec/integration/autobahn_intg_spec.rb
@@ -150,7 +150,7 @@ def counting_down(n, options={})
       it 'sends a batch after a timeout, even if it is not full' do
         @publisher.publish('hello' => 'world')
         @publisher.publish('foo' => 'bar')
-        sleep(@batch_timeout + 0.5)
+        sleep(@batch_timeout * 2)
         message = @queues.map { |q| h, m = q.get; m }.compact.first
         @encoder.decode(message).should == [{'hello' => 'world'}, {'foo' => 'bar'}]
       end