Use a queue to wait for Puback packets rather than polling

This commit changes the `publish` method to use a queue to wait for
Puback packets rather than polling a hash.  Every time the read loop
gets data or a timeout from `IO.select`, it will send a message to
everyone waiting for a `Puback` packet.  If the we're within the
deadline, then the loop executes again, if we got a packet, we'll return
the packet, and if we're outside the deadline, a `-1` is returned.

This upside is that this patch speeds up the publish method by over
100x.  Here is the benchmark:

```ruby
require "securerandom"
require "mqtt"
require "benchmark/ips"
require "stackprof"

client = MQTT::Client.new(username: 'testuser', password: 'testpasswd', client_id: "client_#{SecureRandom.hex(10)}", host: '127.0.0.1')
client.connect

Benchmark.ips do |x|
  x.report("send message") {
    i = rand(1..10)
    topic = "to/timebox#{i}/cameras"
    msg = "message #{Time.now} for timebox#{i}"
    client.publish(topic, msg, true, 1)
  }
end
```

Before this patch:

```
$ ruby -I lib thing.rb
Warming up --------------------------------------
        send message     8.000  i/100ms
Calculating -------------------------------------
        send message     85.042  (± 4.7%) i/s -    432.000  in   5.089261s
```

After this patch:

```
$ ruby -I lib thing.rb
Warming up --------------------------------------
        send message   915.000  i/100ms
Calculating -------------------------------------
        send message      9.453k (± 4.5%) i/s -     47.580k in   5.043716s
```

The downside is that the timeout isn't exact.  Since `IO.select` times
out every `0.5` seconds (according to the `SELECT_TIMEOUT` constant),
the deadline in the `publish` method could be missed by that amount of
time.

Refs njh#115

Author: tenderlove

lib/mqtt/client.rb

@@ -298,6 +298,7 @@ def disconnect(send_msg = true)
         send_packet(packet)
       end
       @socket.close unless @socket.nil?
+      handle_close
       @socket = nil
     end
 
@@ -324,18 +325,30 @@ def publish(topic, payload = '', retain = false, qos = 0)
 
       return if qos.zero?
 
-      Timeout.timeout(@ack_timeout) do
-        while connected?
+      queue = Queue.new
+
+      wait_for_puback packet.id, queue
+
+      deadline = Process.clock_gettime(Process::CLOCK_MONOTONIC) + @ack_timeout
+
+      loop do
+        response = queue.pop
+        case response
+        when :read_timeout
+          if Process.clock_gettime(Process::CLOCK_MONOTONIC) > deadline
+            return -1
+          end
+        when :close
+          return -1
+        else
           @pubacks_semaphore.synchronize do
-            return res if @pubacks.delete(packet.id)
+            @pubacks.delete packet.id
           end
-          # FIXME: make threads communicate with each other, instead of polling
-          # (using a pipe and select ?)
-          sleep 0.01
+          break
         end
       end
 
-      -1
+      res
     end
 
     # Send a subscribe message for one or more topics on the MQTT server.
@@ -459,10 +472,17 @@ def receive_packet
       unless @socket.nil?
         @socket.close
         @socket = nil
+        handle_close
       end
       Thread.current[:parent].raise(exp)
     end
 
+    def wait_for_puback(id, queue)
+      @pubacks_semaphore.synchronize do
+        @pubacks[id] = queue
+      end
+    end
+
     def handle_packet(packet)
       if packet.class == MQTT::Packet::Publish
         # Add to queue
@@ -471,13 +491,25 @@ def handle_packet(packet)
         @last_ping_response = Time.now
       elsif packet.class == MQTT::Packet::Puback
         @pubacks_semaphore.synchronize do
-          @pubacks[packet.id] = packet
+          @pubacks[packet.id] << packet
         end
       end
       # Ignore all other packets
       # FIXME: implement responses for QoS  2
     end
 
+    def handle_timeouts
+      @pubacks_semaphore.synchronize do
+        @pubacks.each_value { |q| q << :read_timeout }
+      end
+    end
+
+    def handle_close
+      @pubacks_semaphore.synchronize do
+        @pubacks.each_value { |q| q << :close }
+      end
+    end
+
     def keep_alive!
       return unless @keep_alive > 0 && connected?
 

spec/mqtt_client_spec.rb

@@ -571,6 +571,26 @@
   end
 
   describe "when calling the 'publish' method" do
+    class ClientWithPubackInjection < MQTT::Client
+      def initialize
+        super(:host => 'localhost')
+        @injected_pubacks = {}
+      end
+
+      def inject_puback(packet)
+        @injected_pubacks[packet.id] = packet
+      end
+
+      def wait_for_puback(id, queue)
+        packet = @injected_pubacks.fetch(id) {
+          return super
+        }
+        queue << packet
+      end
+    end
+
+    let(:client) { ClientWithPubackInjection.new }
+
     before(:each) do
       client.instance_variable_set('@socket', socket)
     end
@@ -973,7 +993,7 @@ def inject_packet(opts={})
 
   def inject_puback(packet_id)
     packet = MQTT::Packet::Puback.new(:id => packet_id)
-    client.instance_variable_get('@pubacks')[packet_id] = packet
+    client.inject_puback packet
   end
 
 end