spec/mqtt_client_spec.rb in mqtt-0.5.0 vs spec/mqtt_client_spec.rb in mqtt-0.6.0

@@ -21,10 +21,21 @@
     else
       socket
     end
   end
 
+  if Process.const_defined? :CLOCK_MONOTONIC
+    def now
+      Process.clock_gettime(Process::CLOCK_MONOTONIC)
+    end
+  else
+    # Support older Ruby
+    def now
+      Time.now.to_f
+    end
+  end
+
   describe "initializing a client" do
     it "with no arguments, it should use the defaults" do
       client = MQTT::Client.new
       expect(client.host).to eq(nil)
       expect(client.port).to eq(1883)
@@ -569,14 +580,52 @@
       client.disconnect
     end
   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
 
+    it "should respect timeouts" do
+      require "socket"
+      rd, wr = UNIXSocket.pair
+      client = MQTT::Client.new(:host => 'localhost', :ack_timeout => 1.0)
+      client.instance_variable_set('@socket', rd)
+      t = Thread.new {
+        Thread.current[:parent] = Thread.main
+        loop do
+          client.send :receive_packet
+        end
+      }
+      start = now
+      expect(client.publish('topic','payload', false, 1)).to eq(-1)
+      elapsed = now - start
+      t.kill
+      expect(elapsed).to be_within(0.1).of(1.0)
+    end
+
     it "should write a valid PUBLISH packet to the socket without the retain flag" do
       client.publish('topic','payload', false, 0)
       expect(socket.string).to eq("\x30\x0e\x00\x05topicpayload")
     end
 
@@ -589,10 +638,23 @@
       inject_puback(1)
       client.publish('topic','payload', false, 1)
       expect(socket.string).to eq("\x32\x10\x00\x05topic\x00\x01payload")
     end
 
+    it "should wrap the packet id after 65535" do
+      0xffff.times do |n|
+        inject_puback(n + 1)
+        client.publish('topic','payload', false, 1)
+      end
+      expect(client.instance_variable_get(:@last_packet_id)).to eq(0xffff)
+
+      socket.string = ""
+      inject_puback(1)
+      client.publish('topic','payload', false, 1)
+      expect(socket.string).to eq("\x32\x10\x00\x05topic\x00\x01payload")
+    end
+
     it "should write a valid PUBLISH packet to the socket with the QoS set to 2" do
       inject_puback(1)
       client.publish('topic','payload', false, 2)
       expect(socket.string).to eq("\x34\x10\x00\x05topic\x00\x01payload")
     end
@@ -688,10 +750,19 @@
       inject_packet(:topic => 'topic0', :payload => 'payload0', :qos => 0)
       expect(client.queue_empty?).to be_falsey
     end
   end
 
+  describe "when calling the 'clear_queue' method" do
+    it "should clear the waiting incoming messages" do
+      inject_packet(:topic => 'topic0', :payload => 'payload0', :qos => 0)
+      expect(client.queue_length).to eq(1)
+      client.clear_queue
+      expect(client.queue_length).to eq(0)
+    end
+  end
+
   describe "when calling the 'get' method" do
     before(:each) do
       client.instance_variable_set('@socket', socket)
     end
 
@@ -866,34 +937,34 @@
 
     it "should update last_ping_response when receiving a Pingresp" do
       allow(MQTT::Packet).to receive(:read).and_return MQTT::Packet::Pingresp.new
       client.instance_variable_set '@last_ping_response', Time.at(0)
       client.send :receive_packet
-      expect(client.last_ping_response).to be_within(1).of Time.now
+      expect(client.last_ping_response).to be_within(1).of now
     end
   end
 
   describe "when calling the 'keep_alive!' method" do
     before(:each) do
       client.instance_variable_set('@socket', socket)
     end
 
     it "should send a ping packet if one is due" do
-      client.instance_variable_set('@last_ping_request', Time.at(0))
+      client.instance_variable_set('@last_ping_request', 0.0)
       client.send('keep_alive!')
       expect(socket.string).to eq("\xC0\x00")
     end
 
     it "should update the time a ping was last sent" do
-      client.instance_variable_set('@last_ping_request', Time.at(0))
+      client.instance_variable_set('@last_ping_request', 0.0)
       client.send('keep_alive!')
-      expect(client.instance_variable_get('@last_ping_request')).not_to eq(0)
+      expect(client.instance_variable_get('@last_ping_request')).to be_within(0.01).of(now)
     end
 
     it "should raise an exception if no ping response has been received" do
-      client.instance_variable_set('@last_ping_request', Time.now)
-      client.instance_variable_set('@last_ping_response', Time.at(0))
+      client.instance_variable_set('@last_ping_request', now)
+      client.instance_variable_set('@last_ping_response', 0.0)
       expect {
         client.send('keep_alive!')
       }.to raise_error(
         MQTT::ProtocolException,
         /No Ping Response received for \d+ seconds/
@@ -949,9 +1020,9 @@
     client.instance_variable_get('@read_queue').push(packet)
   end
 
   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