Fix reader thread raising exceptions into wrong thread

lib/mqtt/client.rb

@@ -180,6 +180,8 @@ def initialize(*args)
       @read_thread = nil
       @write_semaphore = Mutex.new
       @pubacks_semaphore = Mutex.new
+      @last_exception = nil
+      @exception_mutex = Mutex.new
     end
 
     # Get the OpenSSL context, that is used if SSL/TLS is enabled
@@ -282,8 +284,8 @@ def connect(clientid = nil)
         receive_connack
 
         # Start packet reading thread
-        @read_thread = Thread.new(Thread.current) do |parent|
-          Thread.current[:parent] = parent
+        @exception_mutex.synchronize { @last_exception = nil }
+        @read_thread = Thread.new do
           receive_packet while connected?
         end
       end
@@ -305,6 +307,9 @@ def disconnect(send_msg = true)
       @read_thread.kill if @read_thread && @read_thread.alive?
       @read_thread = nil
 
+      # Clear any stored exception so a reconnect starts clean
+      @exception_mutex.synchronize { @last_exception = nil }
+
       return unless connected?
 
       # Close the socket if it is open
@@ -324,6 +329,7 @@ def connected?
 
     # Publish a message on a particular topic to the MQTT server.
     def publish(topic, payload = '', retain = false, qos = 0)
+      check_for_exception!
       raise ArgumentError, 'Topic name cannot be nil' if topic.nil?
       raise ArgumentError, 'Topic name cannot be empty' if topic.empty?
 
@@ -374,6 +380,7 @@ def publish(topic, payload = '', retain = false, qos = 0)
     #   client.subscribe( 'a/b' => 0, 'c/d' => 1 )
     #
     def subscribe(*topics)
+      check_for_exception!
       packet = MQTT::Packet::Subscribe.new(
         :id => next_packet_id,
         :topics => topics
@@ -420,19 +427,28 @@ def get(topic = nil, options = {})
     #   end
     #
     def get_packet(topic = nil)
+      check_for_exception!
       # Subscribe to a topic, if an argument is given
       subscribe(topic) unless topic.nil?
 
       if block_given?
         # Loop forever!
         loop do
           packet = @read_queue.pop
+          if packet == :close
+            check_for_exception!
+            raise MQTT::NotConnectedException
+          end
           yield(packet)
           puback_packet(packet) if packet.qos > 0
         end
       else
         # Wait for one packet to be available
         packet = @read_queue.pop
+        if packet == :close
+          check_for_exception!
+          raise MQTT::NotConnectedException
+        end
         puback_packet(packet) if packet.qos > 0
         return packet
       end
@@ -485,7 +501,7 @@ def receive_packet
         @socket = nil
         handle_close
       end
-      Thread.current[:parent].raise(exp)
+      @exception_mutex.synchronize { @last_exception = exp }
     end
 
     def wait_for_puback(id)
@@ -519,6 +535,7 @@ def handle_close
       @pubacks_semaphore.synchronize do
         @pubacks.each_value { |q| q << :close }
       end
+      @read_queue << :close
     end
 
     if Process.const_defined? :CLOCK_MONOTONIC
@@ -578,6 +595,18 @@ def send_packet(data)
       end
     end
 
+    # Raise any exception stored by the reader thread so the calling thread
+    # receives it, regardless of which thread originally called connect.
+    def check_for_exception!
+      @exception_mutex.synchronize do
+        e = @last_exception
+        if e
+          @last_exception = nil
+          raise e
+        end
+      end
+    end
+
     def parse_uri(uri)
       uri = URI.parse(uri) unless uri.is_a?(URI)
       if uri.scheme == 'mqtt'

spec/mqtt_client_spec.rb

@@ -953,6 +953,27 @@ def wait_for_puback(id)
         expect(packets.map{|p| p.payload}).to eq(['payload0', 'payload1'])
       end
     end
+
+    context "when the connection drops while waiting" do
+      it "should raise a stored reader thread exception" do
+        e = MQTT::Exception.new('connection lost')
+        client.instance_variable_set('@last_exception', e)
+        client.instance_variable_get('@read_queue') << :close
+        expect { client.get_packet }.to raise_error(MQTT::Exception, 'connection lost')
+      end
+
+      it "should raise NotConnectedException if there is no stored exception" do
+        client.instance_variable_get('@read_queue') << :close
+        expect { client.get_packet }.to raise_error(MQTT::NotConnectedException)
+      end
+
+      it "should raise a stored exception when using the block form" do
+        e = Errno::ECONNRESET.new
+        client.instance_variable_set('@last_exception', e)
+        client.instance_variable_get('@read_queue') << :close
+        expect { client.get_packet { |_p| } }.to raise_error(Errno::ECONNRESET)
+      end
+    end
   end
 
   describe "when calling the 'unsubscribe' method" do
@@ -981,8 +1002,6 @@ def wait_for_puback(id)
       client.instance_variable_set('@socket', socket)
       allow(IO).to receive(:select).and_return([[socket], [], []])
       @read_queue = client.instance_variable_get('@read_queue')
-      @parent_thread = Thread.current[:parent] = double('Parent Thread')
-      allow(@parent_thread).to receive(:raise)
     end
 
     it "should put PUBLISH messages on to the read queue" do
@@ -1011,18 +1030,18 @@ def wait_for_puback(id)
       client.send(:receive_packet)
     end
 
-    it "should pass exceptions up to parent thread" do
+    it "should store MQTT exceptions so the calling thread can raise them" do
       e = MQTT::Exception.new
-      expect(@parent_thread).to receive(:raise).with(e).once
       allow(MQTT::Packet).to receive(:read).and_raise(e)
       client.send(:receive_packet)
+      expect { client.send(:check_for_exception!) }.to raise_error(MQTT::Exception)
     end
 
-    it "should pass a system call error up to parent thread" do
+    it "should store system call errors so the calling thread can raise them" do
       e = Errno::ECONNRESET.new
-      expect(@parent_thread).to receive(:raise).with(e).once
       allow(MQTT::Packet).to receive(:read).and_raise(e)
       client.send(:receive_packet)
+      expect { client.send(:check_for_exception!) }.to raise_error(Errno::ECONNRESET)
     end
 
     it "should update last_ping_response when receiving a Pingresp" do
@@ -1033,6 +1052,20 @@ def wait_for_puback(id)
     end
   end
 
+  describe "when calling the 'handle_close' method" do
+    it "should push :close onto the read queue" do
+      client.send(:handle_close)
+      expect(client.instance_variable_get('@read_queue').pop).to eq(:close)
+    end
+
+    it "should notify puback waiters with :close" do
+      queue = Queue.new
+      client.instance_variable_get('@pubacks')[1] = queue
+      client.send(:handle_close)
+      expect(queue.pop).to eq(:close)
+    end
+  end
+
   describe "when calling the 'keep_alive!' method" do
     before(:each) do
       client.instance_variable_set('@socket', socket)

spec/zz_client_integration_spec.rb

@@ -164,8 +164,9 @@ def connect_and_timeout(keep_alive)
 
     context "when keep-alive=1" do
       it "the server should have received at least one ping" do
+        connect_and_timeout(1)
         expect {
-          connect_and_timeout(1)
+          @client.publish('test', '')
         }.to raise_error(
           MQTT::ProtocolException,
           'No Ping Response received for 2 seconds'