Support DPUB to send deferred messages

.ruby-version

@@ -1 +1 @@
-2.1.2
+2.3.1

Gemfile.lock

@@ -28,7 +28,7 @@ GEM
       nokogiri (>= 1.5.10)
       rake
       rdoc
-    json (1.8.1)
+    json (1.8.3)
     jwt (1.0.0)
     mini_portile (0.6.0)
     multi_json (1.10.1)
@@ -71,3 +71,6 @@ DEPENDENCIES
   jeweler (~> 2.0)
   nsq-cluster (~> 1.1)
   rspec (~> 3.0)
+
+BUNDLED WITH
+   1.12.5

README.md

@@ -25,6 +25,15 @@ producer.write('some-message')
 # Write a bunch of messages to NSQ (uses mpub)
 producer.write('one', 'two', 'three', 'four', 'five')
 
+# Write a deferred message to NSQ (uses dpub)
+
+# Message deferred of 10s
+producer.deferred_write(10, 'one')
+
+# Message deferred of 1250ms
+producer.deferred_write(1.25, 'one')
+
+
 # Close the connection
 producer.terminate
 ```

lib/nsq/connection.rb

@@ -97,6 +97,9 @@ def pub(topic, message)
       write ["PUB #{topic}\n", message.bytesize, message].pack('a*l>a*')
     end
 
+    def dpub(topic, delay_in_ms, message)
+      write ["DPUB #{topic} #{delay_in_ms}\n", message.bytesize, message].pack('a*l>a*')
+    end
 
     def mpub(topic, messages)
       body = messages.map do |message|

lib/nsq/producer.rb

@@ -29,7 +29,6 @@ def initialize(opts = {})
       at_exit{terminate}
     end
 
-
     def write(*raw_messages)
       if !@topic
         raise 'No topic specified. Either specify a topic when instantiating the Producer or use write_to_topic.'
@@ -38,6 +37,17 @@ def write(*raw_messages)
       write_to_topic(@topic, *raw_messages)
     end
 
+    # Arg 'delay' in seconds
+    def deferred_write(delay, *raw_messages)
+      if !@topic
+        raise 'No topic specified. Either specify a topic when instantiating the Producer or use write_to_topic.'
+      end
+      if delay < 0.0
+        raise "Delay can't be negative, use a positive float."
+      end
+
+      deferred_write_to_topic(@topic, delay, *raw_messages)
+    end
 
     def write_to_topic(topic, *raw_messages)
       # return error if message(s) not provided
@@ -56,6 +66,14 @@ def write_to_topic(topic, *raw_messages)
       end
     end
 
+    def deferred_write_to_topic(topic, delay, *raw_messages)
+      raise ArgumentError, 'message not provided' if raw_messages.empty?
+      messages = raw_messages.map(&:to_s)
+      connection = connection_for_write
+      messages.each do |msg|
+        connection.dpub(topic, (delay * 1000).to_i, msg)
+      end
+    end
 
     private
     def connection_for_write

spec/lib/nsq/producer_spec.rb

@@ -2,7 +2,6 @@
 require 'json'
 
 describe Nsq::Producer do
-
   def message_count(topic = @producer.topic)
     topics_info = JSON.parse(@nsqd.stats.body)['data']['topics']
     topic_info = topics_info.select{|t| t['topic_name'] == topic }.first
@@ -71,6 +70,22 @@ def new_consumer(topic = TOPIC)
         expect(message_count).to eq(10)
       end
 
+      it 'can queue a deferred message' do
+        @producer.deferred_write 1.0, 1
+        wait_for{message_count==1}
+        expect(message_count).to eq(1)
+      end
+
+      it 'can queue multiple deferred messages' do
+        @producer.deferred_write 1.0, 1, 2, 3
+        wait_for{message_count==3}
+        expect(message_count).to eq(3)
+      end
+
+      it 'raises an exception if delay is negative' do
+        expect {@producer.deferred_write -10, 1}.to raise_error
+      end
+
       it 'shouldn\'t raise an error when nsqd is down' do
         @nsqd.stop