From d07b24bce8ff94fcbaba468950ad8474af58cb72 Mon Sep 17 00:00:00 2001 From: Pierre Goudet Date: Mon, 25 Jun 2018 12:13:08 +0900 Subject: [PATCH] Update to latest source --- .gitignore | 1 - lib/paho-mqtt.rb | 104 ++++++++---------- lib/paho_mqtt/client.rb | 158 +++++++++++++++------------- lib/paho_mqtt/connection_helper.rb | 49 ++++----- lib/paho_mqtt/exception.rb | 52 +++++++++ lib/paho_mqtt/handler.rb | 53 ++++------ lib/paho_mqtt/packet/base.rb | 46 ++++---- lib/paho_mqtt/packet/connack.rb | 16 +-- lib/paho_mqtt/packet/connect.rb | 43 ++++---- lib/paho_mqtt/packet/disconnect.rb | 3 +- lib/paho_mqtt/packet/pingresp.rb | 3 +- lib/paho_mqtt/packet/puback.rb | 3 +- lib/paho_mqtt/packet/pubcomp.rb | 3 +- lib/paho_mqtt/packet/publish.rb | 26 +++-- lib/paho_mqtt/packet/pubrec.rb | 3 +- lib/paho_mqtt/packet/pubrel.rb | 3 +- lib/paho_mqtt/packet/suback.rb | 10 +- lib/paho_mqtt/packet/subscribe.rb | 17 +-- lib/paho_mqtt/packet/unsuback.rb | 3 +- lib/paho_mqtt/packet/unsubscribe.rb | 12 ++- lib/paho_mqtt/publisher.rb | 134 +++++++++++------------ lib/paho_mqtt/sender.rb | 105 ++++++++++++------ lib/paho_mqtt/ssl_helper.rb | 31 ++++-- lib/paho_mqtt/subscriber.rb | 98 +++++++++-------- lib/paho_mqtt/version.rb | 2 +- paho-mqtt.gemspec | 1 + 26 files changed, 549 insertions(+), 430 deletions(-) create mode 100644 lib/paho_mqtt/exception.rb diff --git a/.gitignore b/.gitignore index e44d2e5..185a39f 100644 --- a/.gitignore +++ b/.gitignore @@ -10,4 +10,3 @@ *.gem *.log -.DS_Store \ No newline at end of file diff --git a/lib/paho-mqtt.rb b/lib/paho-mqtt.rb index e4b5e2c..06761d7 100644 --- a/lib/paho-mqtt.rb +++ b/lib/paho-mqtt.rb @@ -11,41 +11,39 @@ # # Contributors: # Pierre Goudet - initial committer +# And Others. require "paho_mqtt/version" require "paho_mqtt/client" +require "paho_mqtt/exception" require "paho_mqtt/packet" require 'logger' module PahoMqtt extend self - attr_accessor :logger + MAX_PACKET_ID = 65535 + # Default connection setup - DEFAULT_SSL_PORT = 8883 - DEFAULT_PORT = 1883 - SELECT_TIMEOUT = 0 - LOOP_TEMPO = 0.005 - RECONNECT_RETRY_TIME = 3 - RECONNECT_RETRY_TEMPO = 5 + DEFAULT_SSL_PORT = 8883 + DEFAULT_PORT = 1883 + SELECT_TIMEOUT = 0.002 # MAX size of queue - MAX_READ = 10 - MAX_PUBACK = 20 - MAX_PUBREC = 20 - MAX_PUBREL = 20 - MAX_PUBCOMP = 20 - MAX_WRITING = MAX_PUBACK + MAX_PUBREC + MAX_PUBREL + MAX_PUBCOMP + MAX_SUBACK = 10 + MAX_UNSUBACK = 10 + MAX_PUBLISH = 1000 + MAX_QUEUE = 1000 # Connection states values - MQTT_CS_NEW = 0 - MQTT_CS_CONNECTED = 1 + MQTT_CS_NEW = 0 + MQTT_CS_CONNECTED = 1 MQTT_CS_DISCONNECT = 2 # Error values MQTT_ERR_SUCCESS = 0 - MQTT_ERR_FAIL = 1 + MQTT_ERR_FAIL = 1 PACKET_TYPES = [ nil, @@ -74,40 +72,40 @@ module PahoMqtt } CLIENT_ATTR_DEFAULTS = { - :host => "", - :port => nil, - :mqtt_version => '3.1.1', + :host => "", + :port => nil, + :mqtt_version => '3.1.1', :clean_session => true, - :persistent => false, - :blocking => false, - :client_id => nil, - :username => nil, - :password => nil, - :ssl => false, - :will_topic => nil, - :will_payload => nil, - :will_qos => 0, - :will_retain => false, - :keep_alive => 60, - :ack_timeout => 5, - :on_connack => nil, - :on_suback => nil, - :on_unsuback => nil, - :on_puback => nil, - :on_pubrel => nil, - :on_pubrec => nil, - :on_pubcomp => nil, - :on_message => nil, + :persistent => false, + :blocking => false, + :client_id => nil, + :username => nil, + :password => nil, + :ssl => false, + :will_topic => nil, + :will_payload => nil, + :will_qos => 0, + :will_retain => false, + :keep_alive => 60, + :ack_timeout => 5, + :on_connack => nil, + :on_suback => nil, + :on_unsuback => nil, + :on_puback => nil, + :on_pubrel => nil, + :on_pubrec => nil, + :on_pubcomp => nil, + :on_message => nil, } - + Thread.abort_on_exception = true def logger=(logger_path) - file = File.open(logger_path, "a+") - file.sync = true - log_file = Logger.new(file) + file = File.open(logger_path, "a+") + file.sync = true + log_file = Logger.new(file) log_file.level = Logger::DEBUG - @logger = log_file + @logger = log_file end def logger @@ -162,22 +160,4 @@ def check_topics(topics, filters) raise ArgumentError end end - - class Exception < ::Exception - end - - class ProtocolViolation < PahoMqtt::Exception - end - - class WritingException < PahoMqtt::Exception - end - - class ReadingException < PahoMqtt::Exception - end - - class PacketException < PahoMqtt::Exception - end - - class LowVersionException < PahoMqtt::Exception - end end diff --git a/lib/paho_mqtt/client.rb b/lib/paho_mqtt/client.rb index f16f0df..07129f2 100644 --- a/lib/paho_mqtt/client.rb +++ b/lib/paho_mqtt/client.rb @@ -28,6 +28,8 @@ class Client attr_accessor :mqtt_version attr_accessor :clean_session attr_accessor :persistent + attr_accessor :reconnect_limit + attr_accessor :reconnect_delay attr_accessor :blocking attr_accessor :client_id attr_accessor :username @@ -49,17 +51,19 @@ class Client attr_reader :ssl_context def initialize(*args) - @last_ping_resp = Time.now - @last_packet_id = 0 - @ssl_context = nil - @sender = nil - @handler = Handler.new - @connection_helper = nil - @connection_state = MQTT_CS_DISCONNECT + @last_ping_resp = Time.now + @last_packet_id = 0 + @ssl_context = nil + @sender = nil + @handler = Handler.new + @connection_helper = nil + @connection_state = MQTT_CS_DISCONNECT @connection_state_mutex = Mutex.new - @mqtt_thread = nil - @reconnect_thread = nil - @id_mutex = Mutex.new + @mqtt_thread = nil + @reconnect_thread = nil + @id_mutex = Mutex.new + @reconnect_limit = 3 + @reconnect_delay = 5 if args.last.is_a?(Hash) attr = args.pop @@ -100,17 +104,19 @@ def config_ssl_context(cert_path, key_path, ca_path=nil) def connect(host=@host, port=@port, keep_alive=@keep_alive, persistent=@persistent, blocking=@blocking) @persistent = persistent - @blocking = blocking - @host = host - @port = port.to_i + @blocking = blocking + @host = host + @port = port.to_i @keep_alive = keep_alive - @connection_state_mutex.synchronize { + @connection_state_mutex.synchronize do @connection_state = MQTT_CS_NEW - } + end @mqtt_thread.kill unless @mqtt_thread.nil? - init_connection + + init_connection unless reconnect? @connection_helper.send_connect(session_params) begin + init_pubsub @connection_state = @connection_helper.do_connect(reconnect?) if connected? build_pubsub @@ -130,7 +136,7 @@ def daemon_mode end rescue SystemCallError => e if @persistent - reconnect() + reconnect else raise e end @@ -146,9 +152,9 @@ def reconnect? Thread.current == @reconnect_thread end - def loop_write(max_packet=MAX_WRITING) + def loop_write begin - @sender.writing_loop(max_packet) + @sender.writing_loop rescue WritingException if check_persistence reconnect @@ -158,16 +164,19 @@ def loop_write(max_packet=MAX_WRITING) end end - def loop_read(max_packet=MAX_READ) - max_packet.times do - begin - @handler.receive_packet - rescue ReadingException - if check_persistence - reconnect - else - raise ReadingException - end + def loop_read + begin + MAX_QUEUE.times do + result = @handler.receive_packet + break if result.nil? + end + rescue FullQueueException + PahoMqtt.logger.warn("Early exit in reading loop. The maximum packets have been reach for #{packet.type_name}") if PahoMqtt.logger? + rescue ReadingException + if check_persistence + reconnect + else + raise ReadingException end end end @@ -176,7 +185,6 @@ def mqtt_loop loop_read loop_write loop_misc - sleep LOOP_TEMPO end def loop_misc @@ -185,32 +193,38 @@ def loop_misc end @publisher.check_waiting_publisher @subscriber.check_waiting_subscriber + sleep SELECT_TIMEOUT end def reconnect @reconnect_thread = Thread.new do - RECONNECT_RETRY_TIME.times do - PahoMqtt.logger.debug("New reconnect atempt...") if PahoMqtt.logger? + counter = 0 + while (@reconnect_limit >= counter || @reconnect_limit == -1) do + counter += 1 + PahoMqtt.logger.debug("New reconnect attempt...") if PahoMqtt.logger? connect if connected? break else - sleep RECONNECT_RETRY_TIME + sleep @reconnect_delay end end unless connected? - PahoMqtt.logger.error("Reconnection atempt counter is over.(#{RECONNECT_RETRY_TIME} times)") if PahoMqtt.logger? + PahoMqtt.logger.error("Reconnection attempt counter is over. (#{@reconnect_limit} times)") if PahoMqtt.logger? disconnect(false) end end end def disconnect(explicit=true) - @last_packet_id = 0 if explicit @connection_helper.do_disconnect(@publisher, explicit, @mqtt_thread) - @connection_state_mutex.synchronize { + @connection_state_mutex.synchronize do @connection_state = MQTT_CS_DISCONNECT - } + end + if explicit && @clean_session + @last_packet_id = 0 + @subscriber.clear_queue + end MQTT_ERR_SUCCESS end @@ -221,7 +235,6 @@ def publish(topic, payload="", retain=false, qos=0) end id = next_packet_id @publisher.send_publish(topic, payload, retain, qos, id) - MQTT_ERR_SUCCESS end def subscribe(*topics) @@ -233,7 +246,6 @@ def subscribe(*topics) MQTT_ERR_SUCCESS rescue ProtocolViolation PahoMqtt.logger.error("Subscribe topics need one topic or a list of topics.") if PahoMqtt.logger? - disconnect(false) raise ProtocolViolation end end @@ -246,8 +258,7 @@ def unsubscribe(*topics) end MQTT_ERR_SUCCESS rescue ProtocolViolation - PahoMqtt.logger.error("Unsubscribe need at least one topics.") if PahoMqtt.logger? - disconnect(false) + PahoMqtt.logger.error("Unsubscribe need at least one topic.") if PahoMqtt.logger? raise ProtocolViolation end end @@ -348,58 +359,53 @@ def subscribed_topics private def next_packet_id - @id_mutex.synchronize { - @last_packet_id = ( @last_packet_id || 0 ).next - } + @id_mutex.synchronize do + @last_packet_id = 0 if @last_packet_id >= MAX_PACKET_ID + @last_packet_id = @last_packet_id.next + end end def downgrade_version - PahoMqtt.logger.debug("Unable to connect to the server with the version #{@mqtt_version}, trying 3.1") if PahoMqtt.logger? + PahoMqtt.logger.debug("Connection refused: unacceptable protocol version #{@mqtt_version}, trying 3.1") if PahoMqtt.logger? if @mqtt_version != "3.1" @mqtt_version = "3.1" connect(@host, @port, @keep_alive) else - raise "Unsupported MQTT version" + raise ProtocolVersionException.new("Unsupported MQTT version") end end - def build_pubsub - if @subscriber.nil? - @subscriber = Subscriber.new(@sender) - else - @subscriber.sender = @sender - @subscriber.config_subscription(next_packet_id) - end - if @publisher.nil? - @publisher = Publisher.new(@sender) - else - @publisher.sender = @sender - @publisher.config_all_message_queue - end + def init_pubsub + @subscriber.nil? ? @subscriber = Subscriber.new(@sender) : @subscriber.sender = @sender + @publisher.nil? ? @publisher = Publisher.new(@sender) : @publisher.sender = @sender @handler.config_pubsub(@publisher, @subscriber) - @sender.flush_waiting_packet(true) + end + + def build_pubsub + @subscriber.config_subscription(next_packet_id) + @sender.flush_waiting_packet + @publisher.config_all_message_queue end def init_connection - unless reconnect? - @connection_helper = ConnectionHelper.new(@host, @port, @ssl, @ssl_context, @ack_timeout) - @connection_helper.handler = @handler - @sender = @connection_helper.sender - end - @connection_helper.setup_connection + @connection_helper = ConnectionHelper.new(@host, @port, @ssl, @ssl_context, @ack_timeout) + @connection_helper.handler = @handler + @sender = @connection_helper.sender end def session_params - {:version => @mqtt_version, - :clean_session => @clean_session, - :keep_alive => @keep_alive, - :client_id => @client_id, - :username => @username, - :password => @password, - :will_topic => @will_topic, - :will_payload => @will_payload, - :will_qos => @will_qos, - :will_retain => @will_retain} + { + :version => @mqtt_version, + :clean_session => @clean_session, + :keep_alive => @keep_alive, + :client_id => @client_id, + :username => @username, + :password => @password, + :will_topic => @will_topic, + :will_payload => @will_payload, + :will_qos => @will_qos, + :will_retain => @will_retain + } end def check_persistence diff --git a/lib/paho_mqtt/connection_helper.rb b/lib/paho_mqtt/connection_helper.rb index 6fd8a69..15500f4 100644 --- a/lib/paho_mqtt/connection_helper.rb +++ b/lib/paho_mqtt/connection_helper.rb @@ -21,14 +21,14 @@ class ConnectionHelper attr_accessor :sender def initialize(host, port, ssl, ssl_context, ack_timeout) - @cs = MQTT_CS_DISCONNECT - @socket = nil - @host = host - @port = port - @ssl = ssl + @cs = MQTT_CS_DISCONNECT + @socket = nil + @host = host + @port = port + @ssl = ssl @ssl_context = ssl_context @ack_timeout = ack_timeout - @sender = Sender.new(ack_timeout) + @sender = Sender.new(ack_timeout) end def handler=(handler) @@ -40,12 +40,11 @@ def do_connect(reconnection=false) @handler.socket = @socket # Waiting a Connack packet for "ack_timeout" second from the remote connect_timeout = Time.now + @ack_timeout - while (Time.now <= connect_timeout) && (!is_connected?) do + while (Time.now <= connect_timeout) && !is_connected? do @cs = @handler.receive_packet - sleep 0.0001 end unless is_connected? - PahoMqtt.logger.warn("Connection failed. Couldn't recieve a Connack packet from: #{@host}, socket is \"#{@socket}\".") if PahoMqtt.logger? + PahoMqtt.logger.warn("Connection failed. Couldn't recieve a Connack packet from: #{@host}.") if PahoMqtt.logger? raise Exception.new("Connection failed. Check log for more details.") unless reconnection end @cs @@ -56,7 +55,7 @@ def is_connected? end def do_disconnect(publisher, explicit, mqtt_thread) - PahoMqtt.logger.debug("Disconnecting from #{@host}") if PahoMqtt.logger? + PahoMqtt.logger.debug("Disconnecting from #{@host}.") if PahoMqtt.logger? if explicit explicit_disconnect(publisher, mqtt_thread) end @@ -65,7 +64,7 @@ def do_disconnect(publisher, explicit, mqtt_thread) end def explicit_disconnect(publisher, mqtt_thread) - @sender.flush_waiting_packet + @sender.flush_waiting_packet(false) send_disconnect mqtt_thread.kill if mqtt_thread && mqtt_thread.alive? publisher.flush_publisher unless publisher.nil? @@ -80,16 +79,16 @@ def setup_connection end def config_socket - PahoMqtt.logger.debug("Atempt to connect to host: #{@host}") if PahoMqtt.logger? + PahoMqtt.logger.debug("Attempt to connect to host: #{@host}...") if PahoMqtt.logger? begin tcp_socket = TCPSocket.new(@host, @port) + if @ssl + encrypted_socket(tcp_socket, @ssl_context) + else + @socket = tcp_socket + end rescue StandardError - PahoMqtt.logger.warn("Could not open a socket with #{@host} on port #{@port}") if PahoMqtt.logger? - end - if @ssl - encrypted_socket(tcp_socket, @ssl_context) - else - @socket = tcp_socket + PahoMqtt.logger.warn("Could not open a socket with #{@host} on port #{@port}.") if PahoMqtt.logger? end end @@ -99,7 +98,7 @@ def encrypted_socket(tcp_socket, ssl_context) @socket.sync_close = true @socket.connect else - PahoMqtt.logger.error("The ssl context was found as nil while the socket's opening.") if PahoMqtt.logger? + PahoMqtt.logger.error("The SSL context was found as nil while the socket's opening.") if PahoMqtt.logger? raise Exception end end @@ -145,22 +144,16 @@ def send_disconnect MQTT_ERR_SUCCESS end - def send_pingreq - packet = PahoMqtt::Packet::Pingreq.new - @sender.send_packet(packet) - MQTT_ERR_SUCCESS - end - def check_keep_alive(persistent, last_ping_resp, keep_alive) now = Time.now timeout_req = (@sender.last_ping_req + (keep_alive * 0.7).ceil) if timeout_req <= now && persistent - PahoMqtt.logger.debug("Checking if server is still alive.") if PahoMqtt.logger? - send_pingreq + PahoMqtt.logger.debug("Checking if server is still alive...") if PahoMqtt.logger? + @sender.send_pingreq end timeout_resp = last_ping_resp + (keep_alive * 1.1).ceil if timeout_resp <= now - PahoMqtt.logger.debug("No activity period over timeout, disconnecting from #{@host}") if PahoMqtt.logger? + PahoMqtt.logger.debug("No activity is over timeout, disconnecting from #{@host}.") if PahoMqtt.logger? @cs = MQTT_CS_DISCONNECT end @cs diff --git a/lib/paho_mqtt/exception.rb b/lib/paho_mqtt/exception.rb new file mode 100644 index 0000000..c4e637d --- /dev/null +++ b/lib/paho_mqtt/exception.rb @@ -0,0 +1,52 @@ +# Copyright (c) 2016-2018 Pierre Goudet +# +# All rights reserved. This program and the accompanying materials +# are made available under the terms of the Eclipse Public License v1.0 +# and Eclipse Distribution License v1.0 which accompany this distribution. +# +# The Eclipse Public License is available at +# https://eclipse.org/org/documents/epl-v10.php. +# and the Eclipse Distribution License is available at +# https://eclipse.org/org/documents/edl-v10.php. +# +# Contributors: +# Pierre Goudet - initial committer + + +module PahoMqtt + class Exception < ::StandardError + def initialize(msg="") + super + end + end + + class ProtocolViolation < Exception + end + + class WritingException < Exception + end + + class ReadingException < Exception + end + + class PacketException < Exception + end + + class PacketFormatException < Exception + end + + class ProtocolVersionException < Exception + end + + class LowVersionException < Exception + end + + class FullWritingException < Exception + end + + class FullQueueException < Exception + end + + class NotSupportedEncryptionException < Exception + end +end diff --git a/lib/paho_mqtt/handler.rb b/lib/paho_mqtt/handler.rb index d882d86..8ad524a 100644 --- a/lib/paho_mqtt/handler.rb +++ b/lib/paho_mqtt/handler.rb @@ -13,8 +13,6 @@ # Pierre Goudet - initial committer # And Others. - - module PahoMqtt class Handler @@ -24,9 +22,9 @@ class Handler def initialize @registered_callback = [] - @last_ping_resp = -1 - @publisher = nil - @subscriber = nil + @last_ping_resp = -1 + @publisher = nil + @subscriber = nil end def config_pubsub(publisher, subscriber) @@ -39,27 +37,28 @@ def socket=(socket) end def receive_packet - result = IO.select([@socket], [], [], SELECT_TIMEOUT) unless @socket.nil? || @socket.closed? + result = IO.select([@socket], nil, nil, SELECT_TIMEOUT) unless @socket.nil? || @socket.closed? unless result.nil? packet = PahoMqtt::Packet::Base.read(@socket) unless packet.nil? if packet.is_a?(PahoMqtt::Packet::Connack) @last_ping_resp = Time.now - handle_connack(packet) + return handle_connack(packet) else handle_packet(packet) @last_ping_resp = Time.now end end end + return result end - + def handle_packet(packet) - PahoMqtt.logger.info("New packet #{packet.class} recieved.") if PahoMqtt.logger? + PahoMqtt.logger.info("New packet #{packet.class} received.") if PahoMqtt.logger? type = packet_type(packet) self.send("handle_#{type}", packet) end - + def register_topic_callback(topic, callback, &block) if topic.nil? PahoMqtt.logger.error("The topics where the callback is trying to be registered have been found nil.") if PahoMqtt.logger? @@ -79,16 +78,17 @@ def clear_topic_callback(topic) PahoMqtt.logger.error("The topics where the callback is trying to be unregistered have been found nil.") if PahoMqtt.logger? raise ArgumentError end - @registered_callback.delete_if {|pair| pair.first == topic} + @registered_callback.delete_if { |pair| pair.first == topic } MQTT_ERR_SUCCESS end def handle_connack(packet) if packet.return_code == 0x00 - PahoMqtt.logger.debug("Connack receive and connection accepted.") if PahoMqtt.logger? + PahoMqtt.logger.debug(packet.return_msg) if PahoMqtt.logger? handle_connack_accepted(packet.session_present) else - handle_connack_error(packet.return_code) + PahoMqtt.logger.warn(packet.return_msg) if PahoMqtt.logger? + MQTT_CS_DISCONNECT end @on_connack.call(packet) unless @on_connack.nil? MQTT_CS_CONNECTED @@ -102,7 +102,7 @@ def handle_connack_accepted(session_flag) def new_session?(session_flag) if !@clean_session && !session_flag - PahoMqtt.logger.debug("New session created for the client") if PahoMqtt.logger? + PahoMqtt.logger.debug("New session created for the client.") if PahoMqtt.logger? end end @@ -124,9 +124,9 @@ def handle_pingresp(_packet) def handle_suback(packet) max_qos = packet.return_codes - id = packet.id - topics = [] - topics = @subscriber.add_subscription(max_qos, id, topics) + id = packet.id + topics = [] + topics = @subscriber.add_subscription(max_qos, id, topics) unless topics.empty? @on_suback.call(topics) unless @on_suback.nil? end @@ -178,18 +178,6 @@ def handle_pubcomp(packet) end end - def handle_connack_error(return_code) - if return_code == 0x01 - raise LowVersionException - elsif CONNACK_ERROR_MESSAGE.has_key(return_code.to_sym) - PahoMqtt.logger.warm(CONNACK_ERRO_MESSAGE[return_code]) - MQTT_CS_DISCONNECTED - else - PahoMqtt.logger("Unknown return code for CONNACK packet: #{return_code}") - raise PacketException - end - end - def on_connack(&block) @on_connack = block if block_given? @on_connack @@ -267,9 +255,8 @@ def packet_type(packet) if PahoMqtt::PACKET_TYPES[3..13].include?(type) type.to_s.split('::').last.downcase else - puts "Packet: #{packet.inspect}" - PahoMqtt.logger.error("Received an unexpeceted packet: #{packet}") if PahoMqtt.logger? - raise PacketException + PahoMqtt.logger.error("Received an unexpeceted packet: #{packet}.") if PahoMqtt.logger? + raise PacketException.new('Invalid packet type id') end end @@ -280,7 +267,7 @@ def check_callback(packet) end unless callbacks.empty? callbacks.each do |callback| - callback.call(packet) + callback.call(packet) end end end diff --git a/lib/paho_mqtt/packet/base.rb b/lib/paho_mqtt/packet/base.rb index 09d6f6c..d1b954e 100644 --- a/lib/paho_mqtt/packet/base.rb +++ b/lib/paho_mqtt/packet/base.rb @@ -38,8 +38,8 @@ class Base # Default attribute values ATTR_DEFAULTS = { - :version => '3.1.0', - :id => 0, + :version => '3.1.0', + :id => 0, :body_length => nil } @@ -49,6 +49,7 @@ def self.read(socket) packet = create_from_header( read_byte(socket) ) + unless packet.nil? packet.validate_flags @@ -61,22 +62,22 @@ def self.read(socket) body_length += ((digit & 0x7F) * multiplier) multiplier *= 0x80 pos += 1 - end while ((digit & 0x80) != 0x00) and pos <= 4 + end while ((digit & 0x80) != 0x00) && pos <= 4 # Store the expected body length in the packet packet.instance_variable_set('@body_length', body_length) # Read in the packet body - packet.parse_body( socket.read(body_length) ) + packet.parse_body(socket.read(body_length)) end - return packet + packet end # Parse buffer into new packet object def self.parse(buffer) packet = parse_header(buffer) packet.parse_body(buffer) - return packet + packet end # Parse the header and create a new packet object of the correct type @@ -84,7 +85,8 @@ def self.parse(buffer) def self.parse_header(buffer) # Check that the packet is a long as the minimum packet size if buffer.bytesize < 2 - raise "Invalid packet: less than 2 bytes long" + raise PahoMqtt::PacketFormatException.new( + "Invalid packet: less than 2 bytes long") end # Create a new packet object @@ -98,13 +100,14 @@ def self.parse_header(buffer) pos = 1 begin if buffer.bytesize <= pos - raise "The packet length header is incomplete" + raise PahoMqtt::PacketFormatException.new( + "The packet length header is incomplete") end digit = bytes[pos] body_length += ((digit & 0x7F) * multiplier) multiplier *= 0x80 pos += 1 - end while ((digit & 0x80) != 0x00) and pos <= 4 + end while ((digit & 0x80) != 0x00) && pos <= 4 # Store the expected body length in the packet packet.instance_variable_set('@body_length', body_length) @@ -112,7 +115,7 @@ def self.parse_header(buffer) # Delete the fixed header from the raw packet passed in buffer.slice!(0...pos) - return packet + packet end # Create a new packet object from the first byte of a MQTT packet @@ -122,7 +125,8 @@ def self.create_from_header(byte) type_id = ((byte & 0xF0) >> 4) packet_class = PahoMqtt::PACKET_TYPES[type_id] if packet_class.nil? - raise "Invalid packet type identifier: #{type_id}" + raise PahoMqtt::PacketFormatException.new( + "Invalid packet type identifier: #{type_id}") end # Convert the last 4 bits of byte into array of true/false @@ -143,7 +147,7 @@ def initialize(args={}) # Set packet attributes from a hash of attribute names and values def update_attributes(attr={}) attr.each_pair do |k,v| - if v.is_a?(Array) or v.is_a?(Hash) + if v.is_a?(Array) || v.is_a?(Hash) send("#{k}=", v.dup) else send("#{k}=", v) @@ -155,9 +159,10 @@ def update_attributes(attr={}) def type_id index = PahoMqtt::PACKET_TYPES.index(self.class) if index.nil? - raise "Invalid packet type: #{self.class}" + raise PahoMqtt::PacketFormatException.new( + "Invalid packet type: #{self.class}") end - return index + index end # Get the name of the packet type as a string in capitals @@ -181,7 +186,8 @@ def body_length=(arg) # Parse the body (variable header and payload) of a packet def parse_body(buffer) if buffer.bytesize != body_length - raise "Failed to parse packet - input buffer (#{buffer.bytesize}) is not the same as the body length header (#{body_length})" + raise PahoMqtt::PacketFormatException.new( + "Failed to parse packet - input buffer (#{buffer.bytesize}) is not the same as the body length header (#{body_length})") end end @@ -207,7 +213,8 @@ def to_s # Check that that packet isn't too big body_length = body.bytesize if body_length > 268435455 - raise "Error serialising packet: body is more than 256MB" + raise PahoMqtt::PacketFormatException.new( + "Error serialising packet: body is more than 256MB") end # Build up the body length field bytes @@ -227,7 +234,8 @@ def to_s # @private def validate_flags if flags != [false, false, false, false] - raise "Invalid flags in #{type_name} packet header" + raise PahoMqtt::PacketFormatException.new( + "Invalid flags in #{type_name} packet header") end end @@ -245,7 +253,7 @@ def encode_bytes(*bytes) # Encode an array of bits and return them def encode_bits(bits) - [bits.map{|b| b ? '1' : '0'}.join].pack('b*') + [bits.map { |b| b ? '1' : '0' }.join].pack('b*') end # Encode a 16-bit unsigned integer and return it @@ -276,7 +284,7 @@ def shift_byte(buffer) # Remove 8 bits from the front of buffer def shift_bits(buffer) - buffer.slice!(0...1).unpack('b8').first.split('').map {|b| b == '1'} + buffer.slice!(0...1).unpack('b8').first.split('').map { |b| b == '1' } end # Remove n bytes from the front of buffer diff --git a/lib/paho_mqtt/packet/connack.rb b/lib/paho_mqtt/packet/connack.rb index 074edac..3a7f439 100644 --- a/lib/paho_mqtt/packet/connack.rb +++ b/lib/paho_mqtt/packet/connack.rb @@ -29,7 +29,7 @@ class Connack < PahoMqtt::Packet::Base attr_accessor :return_code # Default attribute values - ATTR_DEFAULTS = {:return_code => 0x00} + ATTR_DEFAULTS = { :return_code => 0x00 } # Create a new Client Connect packet def initialize(args={}) @@ -56,9 +56,9 @@ def session_present=(arg) def return_msg case return_code when 0x00 - "Connection Accepted" + "Connection accepted" when 0x01 - "Connection refused: unacceptable protocol version" + raise LowVersionException when 0x02 "Connection refused: client identifier rejected" when 0x03 @@ -77,19 +77,21 @@ def encode_body body = '' body += encode_bits(@connack_flags) body += encode_bytes(@return_code.to_i) - return body + body end # Parse the body (variable header and payload) of a Connect Acknowledgment packet def parse_body(buffer) super(buffer) @connack_flags = shift_bits(buffer) - unless @connack_flags[1,7] == [false, false, false, false, false, false, false] - raise "Invalid flags in Connack variable header" + unless @connack_flags[1, 7] == [false, false, false, false, false, false, false] + raise PacketFormatException.new( + "Invalid flags in Connack variable header") end @return_code = shift_byte(buffer) unless buffer.empty? - raise "Extra bytes at end of Connect Acknowledgment packet" + raise PacketFormatException.new( + "Extra bytes at end of Connect Acknowledgment packet") end end diff --git a/lib/paho_mqtt/packet/connect.rb b/lib/paho_mqtt/packet/connect.rb index 948aec6..65f53fe 100644 --- a/lib/paho_mqtt/packet/connect.rb +++ b/lib/paho_mqtt/packet/connect.rb @@ -57,29 +57,30 @@ class Connect < PahoMqtt::Packet::Base # Default attribute values ATTR_DEFAULTS = { - :client_id => nil, + :client_id => nil, :clean_session => true, - :keep_alive => 15, - :will_topic => nil, - :will_qos => 0, - :will_retain => false, - :will_payload => '', - :username => nil, - :password => nil, + :keep_alive => 15, + :will_topic => nil, + :will_qos => 0, + :will_retain => false, + :will_payload => '', + :username => nil, + :password => nil, } # Create a new Client Connect packet def initialize(args={}) super(ATTR_DEFAULTS.merge(args)) - if version == '3.1.0' or version == '3.1' + if version == '3.1.0' || version == '3.1' self.protocol_name ||= 'MQIsdp' self.protocol_level ||= 0x03 elsif version == '3.1.1' self.protocol_name ||= 'MQTT' self.protocol_level ||= 0x04 else - raise ArgumentError.new("Unsupported protocol version: #{version}") + raise PahoMqtt::PacketFormatException.new( + "Unsupported protocol version: #{version}") end end @@ -90,7 +91,8 @@ def encode_body body += encode_string(@protocol_name) body += encode_bytes(@protocol_level.to_i) if @keep_alive < 0 - raise "Invalid keep-alive value: cannot be less than 0" + raise PahoMqtt::PacketFormatException.new( + "Invalid keep-alive value: cannot be less than 0") end body += encode_flags(@connect_flags) @@ -108,10 +110,12 @@ def encode_body def check_version if @version == '3.1.0' - if @client_id.nil? or @client_id.bytesize < 1 - raise "Client identifier too short while serialising packet" + if @client_id.nil? || @client_id.bytesize < 1 + raise PahoMqtt::PacketFormatException.new( + "Client identifier too short while serialising packet") elsif @client_id.bytesize > 23 - raise "Client identifier too long when serialising packet" + raise PahoMqtt::PacketFormatException.new( + "Client identifier too long when serialising packet") end end end @@ -133,12 +137,13 @@ def parse_body(buffer) super(buffer) @protocol_name = shift_string(buffer) @protocol_level = shift_byte(buffer).to_i - if @protocol_name == 'MQIsdp' and @protocol_level == 3 + if @protocol_name == 'MQIsdp' && @protocol_level == 3 @version = '3.1.0' - elsif @protocol_name == 'MQTT' and @protocol_level == 4 + elsif @protocol_name == 'MQTT' && @protocol_level == 4 @version = '3.1.1' else - raise "Unsupported protocol: #{@protocol_name}/#{@protocol_level}" + raise PahoMqtt::PacketFormatException.new( + "Unsupported protocol: #{@protocol_name}/#{@protocol_level}") end @connect_flags = shift_byte(buffer) @@ -157,10 +162,10 @@ def parse_connect_flags(flags, buffer) # The MQTT v3.1 specification says that the payload is a UTF-8 string @will_payload = shift_string(buffer) end - if ((@connect_flags & 0x80) >> 7) == 0x01 and buffer.bytesize > 0 + if ((@connect_flags & 0x80) >> 7) == 0x01 && buffer.bytesize > 0 @username = shift_string(buffer) end - if ((@connect_flags & 0x40) >> 6) == 0x01 and buffer.bytesize > 0 + if ((@connect_flags & 0x40) >> 6) == 0x01 && buffer.bytesize > 0 @password = shift_string(buffer) end end diff --git a/lib/paho_mqtt/packet/disconnect.rb b/lib/paho_mqtt/packet/disconnect.rb index 994c92d..5915fbe 100644 --- a/lib/paho_mqtt/packet/disconnect.rb +++ b/lib/paho_mqtt/packet/disconnect.rb @@ -31,7 +31,8 @@ def initialize(args={}) def parse_body(buffer) super(buffer) unless buffer.empty? - raise "Extra bytes at end of Disconnect packet" + raise PacketFormatException.new( + "Extra bytes at the end of Disconnect packet") end end end diff --git a/lib/paho_mqtt/packet/pingresp.rb b/lib/paho_mqtt/packet/pingresp.rb index 276783d..6d8f6d1 100644 --- a/lib/paho_mqtt/packet/pingresp.rb +++ b/lib/paho_mqtt/packet/pingresp.rb @@ -31,7 +31,8 @@ def initialize(args={}) def parse_body(buffer) super(buffer) unless buffer.empty? - raise "Extra bytes at end of Ping Response packet" + raise PahoMqtt::PacketFormatException.new( + "Extra bytes at end of Ping Response packet") end end end diff --git a/lib/paho_mqtt/packet/puback.rb b/lib/paho_mqtt/packet/puback.rb index b47bbd8..ac63779 100644 --- a/lib/paho_mqtt/packet/puback.rb +++ b/lib/paho_mqtt/packet/puback.rb @@ -32,7 +32,8 @@ def parse_body(buffer) super(buffer) @id = shift_short(buffer) unless buffer.empty? - raise "Extra bytes at end of Publish Acknowledgment packet" + raise PahoMqtt::PacketFormatException.new( + "Extra bytes at end of Publish Acknowledgment packet") end end diff --git a/lib/paho_mqtt/packet/pubcomp.rb b/lib/paho_mqtt/packet/pubcomp.rb index 97e47a2..d494727 100644 --- a/lib/paho_mqtt/packet/pubcomp.rb +++ b/lib/paho_mqtt/packet/pubcomp.rb @@ -32,7 +32,8 @@ def parse_body(buffer) super(buffer) @id = shift_short(buffer) unless buffer.empty? - raise "Extra bytes at end of Publish Complete packet" + raise PahoMqtt::PacketFormatException.new( + "Extra bytes at end of Publish Complete packet") end end diff --git a/lib/paho_mqtt/packet/publish.rb b/lib/paho_mqtt/packet/publish.rb index 826c172..aaba29f 100644 --- a/lib/paho_mqtt/packet/publish.rb +++ b/lib/paho_mqtt/packet/publish.rb @@ -39,7 +39,7 @@ class Publish < PahoMqtt::Packet::Base # Default attribute values ATTR_DEFAULTS = { - :topic => nil, + :topic => nil, :payload => '' } @@ -81,8 +81,9 @@ def qos # Set the Quality of Service level (0/1/2) def qos=(arg) @qos = arg.to_i - if @qos < 0 or @qos > 2 - raise "Invalid QoS value: #{@qos}" + if @qos < 0 || @qos > 2 + raise PahoMqtt::PacketFormatException.new( + "Invalid QoS value: #{@qos}") else @flags[1] = (arg & 0x01 == 0x01) @flags[2] = (arg & 0x02 == 0x02) @@ -92,8 +93,9 @@ def qos=(arg) # Get serialisation of packet's body def encode_body body = '' - if @topic.nil? or @topic.to_s.empty? - raise "Invalid topic name when serialising packet" + if @topic.nil? || @topic.to_s.empty? + raise PahoMqtt::PacketFormatException.new( + "Invalid topic name when serialising packet") end body += encode_string(@topic) body += encode_short(@id) unless qos == 0 @@ -104,8 +106,8 @@ def encode_body # Parse the body (variable header and payload) of a Publish packet def parse_body(buffer) super(buffer) - @topic = shift_string(buffer) - @id = shift_short(buffer) unless qos == 0 + @topic = shift_string(buffer) + @id = shift_short(buffer) unless qos == 0 @payload = buffer end @@ -113,10 +115,12 @@ def parse_body(buffer) # @private def validate_flags if qos == 3 - raise "Invalid packet: QoS value of 3 is not allowed" + raise PahoMqtt::PacketFormatException.new( + "Invalid packet: QoS value of 3 is not allowed") end - if qos == 0 and duplicate - raise "Invalid packet: DUP cannot be set for QoS 0" + if qos == 0 && duplicate + raise PahoMqtt::PacketFormatException.new( + "Invalid packet: DUP cannot be set for QoS 0") end end @@ -135,7 +139,7 @@ def inspect def inspect_payload str = payload.to_s - if str.bytesize < 16 and str =~ /^[ -~]*$/ + if str.bytesize < 16 && str =~ /^[ -~]*$/ "'#{str}'" else "... (#{str.bytesize} bytes)" diff --git a/lib/paho_mqtt/packet/pubrec.rb b/lib/paho_mqtt/packet/pubrec.rb index 2587d12..494f08d 100644 --- a/lib/paho_mqtt/packet/pubrec.rb +++ b/lib/paho_mqtt/packet/pubrec.rb @@ -32,7 +32,8 @@ def parse_body(buffer) super(buffer) @id = shift_short(buffer) unless buffer.empty? - raise "Extra bytes at end of Publish Received packet" + raise PahoMqtt::PacketFormatException.new( + "Extra bytes at end of Publish Received packet") end end diff --git a/lib/paho_mqtt/packet/pubrel.rb b/lib/paho_mqtt/packet/pubrel.rb index 461a308..37db3e9 100644 --- a/lib/paho_mqtt/packet/pubrel.rb +++ b/lib/paho_mqtt/packet/pubrel.rb @@ -50,7 +50,8 @@ def parse_body(buffer) # @private def validate_flags if @flags != [false, true, false, false] - raise "Invalid flags in PUBREL packet header" + raise PahoMqtt::PacketFormatException.new( + "Invalid flags in #{type_name} packet header") end end diff --git a/lib/paho_mqtt/packet/suback.rb b/lib/paho_mqtt/packet/suback.rb index e8c21fb..d176381 100644 --- a/lib/paho_mqtt/packet/suback.rb +++ b/lib/paho_mqtt/packet/suback.rb @@ -43,14 +43,16 @@ def return_codes=(value) elsif value.is_a?(Integer) @return_codes = [value] else - raise "return_codes should be an integer or an array of return codes" + raise PahoMqtt::PacketFormatException.new( + "return_codes should be an integer or an array of return codes") end end # Get serialisation of packet's body def encode_body if @return_codes.empty? - raise "no granted QoS given when serialising packet" + raise PahoMqtt::PacketFormatException.new( + "No granted QoS given when serialising packet") end body = encode_short(@id) return_codes.each { |qos| body += encode_bytes(qos) } @@ -61,14 +63,14 @@ def encode_body def parse_body(buffer) super(buffer) @id = shift_short(buffer) - while(buffer.bytesize>0) + while buffer.bytesize > 0 @return_codes << shift_byte(buffer) end end # Returns a human readable string, summarising the properties of the packet def inspect - "\#<#{self.class}: 0x%2.2X, rc=%s>" % [id, return_codes.map{|rc| "0x%2.2X" % rc}.join(',')] + "\#<#{self.class}: 0x%2.2X, rc=%s>" % [id, return_codes.map { |rc| "0x%2.2X" % rc }.join(',')] end end end diff --git a/lib/paho_mqtt/packet/subscribe.rb b/lib/paho_mqtt/packet/subscribe.rb index 0d309f9..3387d72 100644 --- a/lib/paho_mqtt/packet/subscribe.rb +++ b/lib/paho_mqtt/packet/subscribe.rb @@ -66,13 +66,14 @@ def topics=(value) # Peek at the next item in the array, and remove it if it is an integer if input.first.is_a?(Integer) qos = input.shift - @topics << [item,qos] + @topics << [item, qos] else - @topics << [item,0] + @topics << [item, 0] end else # Meh? - raise "Invalid topics input: #{value.inspect}" + raise PahoMqtt::PacketFormatException.new( + "Invalid topics input: #{value.inspect}") end end @topics @@ -81,7 +82,8 @@ def topics=(value) # Get serialisation of packet's body def encode_body if @topics.empty? - raise "no topics given when serialising packet" + raise PahoMqtt::PacketFormatException.new( + "No topics given when serialising packet") end body = encode_short(@id) topics.each do |item| @@ -99,7 +101,7 @@ def parse_body(buffer) while(buffer.bytesize>0) topic_name = shift_string(buffer) topic_qos = shift_byte(buffer) - @topics << [topic_name,topic_qos] + @topics << [topic_name, topic_qos] end end @@ -107,7 +109,8 @@ def parse_body(buffer) # @private def validate_flags if @flags != [false, true, false, false] - raise "Invalid flags in SUBSCRIBE packet header" + raise PahoMqtt::PacketFormatException.new( + "Invalid flags in SUBSCRIBE packet header") end end @@ -115,7 +118,7 @@ def validate_flags def inspect _str = "\#<#{self.class}: 0x%2.2X, %s>" % [ id, - topics.map {|t| "'#{t[0]}':#{t[1]}"}.join(', ') + topics.map { |t| "'#{t[0]}':#{t[1]}" }.join(', ') ] end end diff --git a/lib/paho_mqtt/packet/unsuback.rb b/lib/paho_mqtt/packet/unsuback.rb index 09b7947..fe760a9 100644 --- a/lib/paho_mqtt/packet/unsuback.rb +++ b/lib/paho_mqtt/packet/unsuback.rb @@ -37,7 +37,8 @@ def parse_body(buffer) super(buffer) @id = shift_short(buffer) unless buffer.empty? - raise "Extra bytes at end of Unsubscribe Acknowledgment packet" + raise PahoMqtt::PacketFormatException.new( + "Extra bytes at end of Unsubscribe Acknowledgment packet") end end diff --git a/lib/paho_mqtt/packet/unsubscribe.rb b/lib/paho_mqtt/packet/unsubscribe.rb index 4707748..a553dcb 100644 --- a/lib/paho_mqtt/packet/unsubscribe.rb +++ b/lib/paho_mqtt/packet/unsubscribe.rb @@ -28,7 +28,7 @@ class Unsubscribe < PahoMqtt::Packet::Base # Default attribute values ATTR_DEFAULTS = { :topics => [], - :flags => [false, true, false, false], + :flags => [false, true, false, false], } # Create a new Unsubscribe packet @@ -48,7 +48,8 @@ def topics=(value) # Get serialisation of packet's body def encode_body if @topics.empty? - raise "no topics given when serialising packet" + raise PahoMqtt::PacketFormatException.new( + "No topics given when serialising packet") end body = encode_short(@id) topics.each { |topic| body += encode_string(topic) } @@ -59,7 +60,7 @@ def encode_body def parse_body(buffer) super(buffer) @id = shift_short(buffer) - while(buffer.bytesize>0) + while buffer.bytesize > 0 @topics << shift_string(buffer) end end @@ -68,7 +69,8 @@ def parse_body(buffer) # @private def validate_flags if @flags != [false, true, false, false] - raise "Invalid flags in UNSUBSCRIBE packet header" + raise PahoMqtt::PacketFormatException.new( + "Invalid flags in UNSUBSCRIBE packet header") end end @@ -76,7 +78,7 @@ def validate_flags def inspect "\#<#{self.class}: 0x%2.2X, %s>" % [ id, - topics.map {|t| "'#{t}'"}.join(', ') + topics.map { |t| "'#{t}'" }.join(', ') ] end end diff --git a/lib/paho_mqtt/publisher.rb b/lib/paho_mqtt/publisher.rb index af3a65d..aee97a6 100644 --- a/lib/paho_mqtt/publisher.rb +++ b/lib/paho_mqtt/publisher.rb @@ -13,20 +13,19 @@ # Pierre Goudet - initial committer # And Others. - module PahoMqtt class Publisher def initialize(sender) - @waiting_puback = [] - @waiting_pubrec = [] - @waiting_pubrel = [] + @waiting_puback = [] + @waiting_pubrec = [] + @waiting_pubrel = [] @waiting_pubcomp = [] - @puback_mutex = Mutex.new - @pubrec_mutex = Mutex.new - @pubrel_mutex = Mutex.new - @pubcomp_mutex = Mutex.new - @sender = sender + @puback_mutex = Mutex.new + @pubrec_mutex = Mutex.new + @pubrel_mutex = Mutex.new + @pubcomp_mutex = Mutex.new + @sender = sender end def sender=(sender) @@ -35,22 +34,34 @@ def sender=(sender) def send_publish(topic, payload, retain, qos, new_id) packet = PahoMqtt::Packet::Publish.new( - :id => new_id, - :topic => topic, + :id => new_id, + :topic => topic, :payload => payload, - :retain => retain, - :qos => qos + :retain => retain, + :qos => qos ) + begin + case qos + when 1 + push_queue(@waiting_puback, @puback_mutex, MAX_QUEUE, packet, new_id) + when 2 + push_queue(@waiting_pubrec, @pubrec_mutex, MAX_QUEUE, packet, new_id) + end + rescue FullQueueException + PahoMqtt.logger.warn("PUBLISH queue is full, waiting for publishing #{packet.inspect}") if PahoMqtt.logger? + sleep SELECT_TIMEOUT + retry + end @sender.append_to_writing(packet) - case qos - when 1 - @puback_mutex.synchronize{ - @waiting_puback.push({:id => new_id, :packet => packet, :timestamp => Time.now}) - } - when 2 - @pubrec_mutex.synchronize{ - @waiting_pubrec.push({:id => new_id, :packet => packet, :timestamp => Time.now}) - } + MQTT_ERR_SUCCESS + end + + def push_queue(waiting_queue, queue_mutex, max_packet, packet, new_id) + if waiting_queue.length >= max_packet + raise FullQueueException + end + queue_mutex.synchronize do + waiting_queue.push(:id => new_id, :packet => packet, :timestamp => Time.now) end MQTT_ERR_SUCCESS end @@ -63,8 +74,8 @@ def do_publish(qos, packet_id) when 2 send_pubrec(packet_id) else - @logger.error("The packet qos value is invalid in publish.") if logger? - raise PacketException + PahoMqtt.logger.error("The packet QoS value is invalid in publish.") if PahoMqtt.logger? + raise PacketException.new('Invalid publish QoS value') end MQTT_ERR_SUCCESS end @@ -78,48 +89,42 @@ def send_puback(packet_id) end def do_puback(packet_id) - @puback_mutex.synchronize{ + @puback_mutex.synchronize do @waiting_puback.delete_if { |pck| pck[:id] == packet_id } - } - MQTT_ERR_SUCCESS + end + MQTT_ERR_SUCCESS end - + def send_pubrec(packet_id) packet = PahoMqtt::Packet::Pubrec.new( :id => packet_id ) + push_queue(@waiting_pubrel, @pubrel_mutex, MAX_QUEUE, packet, packet_id) @sender.append_to_writing(packet) - @pubrel_mutex.synchronize{ - @waiting_pubrel.push({:id => packet_id , :packet => packet, :timestamp => Time.now}) - } MQTT_ERR_SUCCESS end def do_pubrec(packet_id) - @pubrec_mutex.synchronize { + @pubrec_mutex.synchronize do @waiting_pubrec.delete_if { |pck| pck[:id] == packet_id } - } + end send_pubrel(packet_id) - MQTT_ERR_SUCCESS end def send_pubrel(packet_id) packet = PahoMqtt::Packet::Pubrel.new( :id => packet_id ) + push_queue(@waiting_pubcomp, @pubcomp_mutex, MAX_QUEUE, packet, packet_id) @sender.append_to_writing(packet) - @pubcomp_mutex.synchronize{ - @waiting_pubcomp.push({:id => packet_id, :packet => packet, :timestamp => Time.now}) - } MQTT_ERR_SUCCESS end def do_pubrel(packet_id) - @pubrel_mutex.synchronize { + @pubrel_mutex.synchronize do @waiting_pubrel.delete_if { |pck| pck[:id] == packet_id } - } + end send_pubcomp(packet_id) - MQTT_ERR_SUCCESS end def send_pubcomp(packet_id) @@ -131,52 +136,47 @@ def send_pubcomp(packet_id) end def do_pubcomp(packet_id) - @pubcomp_mutex.synchronize { + @pubcomp_mutex.synchronize do @waiting_pubcomp.delete_if { |pck| pck[:id] == packet_id } - } + end MQTT_ERR_SUCCESS end def config_all_message_queue - config_message_queue(@waiting_puback, @puback_mutex, MAX_PUBACK) - config_message_queue(@waiting_pubrec, @pubrec_mutex, MAX_PUBREC) - config_message_queue(@waiting_pubrel, @pubrel_mutex, MAX_PUBREL) - config_message_queue(@waiting_pubcomp, @pubcomp_mutex, MAX_PUBCOMP) + config_message_queue(@waiting_puback, @puback_mutex) + config_message_queue(@waiting_pubrec, @pubrec_mutex) + config_message_queue(@waiting_pubrel, @pubrel_mutex) + config_message_queue(@waiting_pubcomp, @pubcomp_mutex) end - def config_message_queue(queue, mutex, max_packet) - mutex.synchronize { - cnt = 0 + def config_message_queue(queue, mutex) + mutex.synchronize do queue.each do |pck| - pck[:packet].dup ||= true - if cnt <= max_packet - @sender.append_to_writing(pck[:packet]) - cnt += 1 - end + pck[:timestamp] = Time.now end - } + end end def check_waiting_publisher - @sender.check_ack_alive(@waiting_puback, @puback_mutex, MAX_PUBACK) - @sender.check_ack_alive(@waiting_pubrec, @pubrec_mutex, MAX_PUBREC) - @sender.check_ack_alive(@waiting_pubrel, @pubrel_mutex, MAX_PUBREL) - @sender.check_ack_alive(@waiting_pubcomp, @pubcomp_mutex, MAX_PUBCOMP) + @sender.check_ack_alive(@waiting_puback, @puback_mutex) + @sender.check_ack_alive(@waiting_pubrec, @pubrec_mutex) + @sender.check_ack_alive(@waiting_pubrel, @pubrel_mutex) + @sender.check_ack_alive(@waiting_pubcomp, @pubcomp_mutex) end def flush_publisher - @puback_mutex.synchronize { + @puback_mutex.synchronize do @waiting_puback = [] - } - @pubrec_mutex.synchronize { + end + @pubrec_mutex.synchronize do @waiting_pubrec = [] - } - @pubrel_mutex.synchronize { + end + @pubrel_mutex.synchronize do @waiting_pubrel = [] - } - @pubcomp_mutex.synchronize { + end + @pubcomp_mutex.synchronize do @waiting_pubcomp = [] - } + end end end end diff --git a/lib/paho_mqtt/sender.rb b/lib/paho_mqtt/sender.rb index 194417a..fb3487d 100644 --- a/lib/paho_mqtt/sender.rb +++ b/lib/paho_mqtt/sender.rb @@ -17,13 +17,15 @@ module PahoMqtt class Sender attr_accessor :last_ping_req - + def initialize(ack_timeout) - @socket = nil - @writing_queue = [] - @writing_mutex = Mutex.new - @last_ping_req = -1 - @ack_timeout = ack_timeout + @socket = nil + @writing_queue = [] + @publish_queue = [] + @publish_mutex = Mutex.new + @writing_mutex = Mutex.new + @last_ping_req = -1 + @ack_timeout = ack_timeout end def socket=(socket) @@ -35,57 +37,90 @@ def send_packet(packet) @socket.write(packet.to_s) unless @socket.nil? || @socket.closed? @last_ping_req = Time.now MQTT_ERR_SUCCESS - rescue StandardError - raise WritingException + end + rescue StandardError + raise WritingException + rescue IO::WaitWritable + IO.select(nil, [@socket], nil, SELECT_TIMEOUT) + retry + end + + def send_pingreq + send_packet(PahoMqtt::Packet::Pingreq.new) + end + + def prepare_sending(queue, mutex, max_packet, packet) + if queue.length < max_packet + mutex.synchronize do + queue.push(packet) + end + else + PahoMqtt.logger.error('Writing queue is full, slowing down') if PahoMqtt.logger? + raise FullWritingException end end - def append_to_writing(packet) - @writing_mutex.synchronize { - @writing_queue.push(packet) - } + def append_to_writing(packet) + begin + if packet.is_a?(PahoMqtt::Packet::Publish) + prepare_sending(@publish_queue, @publish_mutex, MAX_PUBLISH, packet) + else + prepare_sending(@writing_queue, @writing_mutex, MAX_QUEUE, packet) + end + rescue FullWritingException + sleep SELECT_TIMEOUT + retry + end MQTT_ERR_SUCCESS end - def writing_loop(max_packet) - @writing_mutex.synchronize { - cnt = 0 - while !@writing_queue.empty? && cnt < max_packet do + def writing_loop + @writing_mutex.synchronize do + MAX_QUEUE.times do + break if @writing_queue.empty? packet = @writing_queue.shift send_packet(packet) - cnt += 1 end - } + end + @publish_mutex.synchronize do + MAX_PUBLISH.times do + break if @publish_queue.empty? + packet = @publish_queue.shift + send_packet(packet) + end + end MQTT_ERR_SUCCESS end - + def flush_waiting_packet(sending=true) if sending - @writing_mutex.synchronize { - @writing_queue.each do |m| - send_packet(m) + @writing_mutex.synchronize do + @writing_queue.each do |packet| + send_packet(packet) end - } - else - @writing_queue = [] + end + @publish_mutex.synchronize do + @publish_queue.each do |packet| + send_packet(packet) + end + end end + @writing_queue = [] + @publish_queue = [] end - - def check_ack_alive(queue, mutex, max_packet) - mutex.synchronize { + + def check_ack_alive(queue, mutex) + mutex.synchronize do now = Time.now - cnt = 0 queue.each do |pck| if now >= pck[:timestamp] + @ack_timeout pck[:packet].dup ||= true unless pck[:packet].class == PahoMqtt::Packet::Subscribe || pck[:packet].class == PahoMqtt::Packet::Unsubscribe - unless cnt > max_packet - append_to_writing(pck[:packet]) - pck[:timestamp] = now - cnt += 1 - end + PahoMqtt.logger.info("Acknowledgement timeout is over, resending #{pck[:packet].inspect}") if PahoMqtt.logger? + send_packet(pck[:packet]) + pck[:timestamp] = now end end - } + end end end end diff --git a/lib/paho_mqtt/ssl_helper.rb b/lib/paho_mqtt/ssl_helper.rb index 770e55c..fc1800b 100644 --- a/lib/paho_mqtt/ssl_helper.rb +++ b/lib/paho_mqtt/ssl_helper.rb @@ -18,21 +18,40 @@ module PahoMqtt module SSLHelper extend self - def config_ssl_context(cert_path, key_path, ca_path=nil) + def config_ssl_context(cert_path=nil, key_path=nil, ca_path=nil) ssl_context = OpenSSL::SSL::SSLContext.new set_cert(cert_path, ssl_context) set_key(key_path, ssl_context) set_root_ca(ca_path, ssl_context) - #ssl_context.verify_mode = OpenSSL::SSL::VERIFY_PEER unless ca_path.nil? + # ssl_context.verify_mode = OpenSSL::SSL::VERIFY_PEER unless ca_path.nil? ssl_context end - def set_cert(cert_path, ssl_context) - ssl_context.cert = OpenSSL::X509::Certificate.new(File.read(cert_path)) + def set_cert(cert_path=nil, ssl_context) + unless cert_path.nil? + ssl_context.cert = OpenSSL::X509::Certificate.new(File.read(cert_path)) + end end - def set_key(key_path, ssl_context) - ssl_context.key = OpenSSL::PKey::RSA.new(File.read(key_path)) + def set_key(key_path=nil, ssl_context) + unless key_path.nil? + return MQTT_ERR_SUCCESS if try_rsa_key(key_path, ssl_context) == MQTT_ERR_SUCCESS + begin + ssl_context.key = OpenSSL::PKey::EC.new(File.read(key_path)) + return MQTT_ERR_SUCCESS + rescue OpenSSL::PKey::ECError + raise NotSupportedEncryptionException.new("Could not support the type of the provided key (supported: RSA and EC)") + end + end + end + + def try_rsa_key(key_path, ssl_context) + begin + ssl_context.key = OpenSSL::PKey::RSA.new(File.read(key_path)) + return MQTT_ERR_SUCCESS + rescue OpenSSL::PKey::RSAError + return MQTT_ERR_FAIL + end end def set_root_ca(ca_path, ssl_context) diff --git a/lib/paho_mqtt/subscriber.rb b/lib/paho_mqtt/subscriber.rb index 9040ba6..14bfb51 100644 --- a/lib/paho_mqtt/subscriber.rb +++ b/lib/paho_mqtt/subscriber.rb @@ -19,13 +19,13 @@ class Subscriber attr_reader :subscribed_topics def initialize(sender) - @waiting_suback = [] - @waiting_unsuback = [] - @subscribed_mutex = Mutex.new + @waiting_suback = [] + @waiting_unsuback = [] + @subscribed_mutex = Mutex.new @subscribed_topics = [] - @suback_mutex = Mutex.new - @unsuback_mutex = Mutex.new - @sender = sender + @suback_mutex = Mutex.new + @unsuback_mutex = Mutex.new + @sender = sender end def sender=(sender) @@ -35,24 +35,28 @@ def sender=(sender) def config_subscription(new_id) unless @subscribed_topics == [] || @subscribed_topics.nil? packet = PahoMqtt::Packet::Subscribe.new( - :id => new_id, + :id => new_id, :topics => @subscribed_topics ) - @subscribed_mutex.synchronize { + @subscribed_mutex.synchronize do @subscribed_topics = [] - } - @suback_mutex.synchronize { - @waiting_suback.push({ :id => new_id, :packet => packet, :timestamp => Time.now }) - } - @sender.send_packet(packet) + end + @suback_mutex.synchronize do + if @waiting_suback.length >= MAX_SUBACK + PahoMqtt.logger.error('SUBACK queue is full, could not send subscribe') if PahoMqtt.logger? + return MQTT_ERR_FAILURE + end + @waiting_suback.push(:id => new_id, :packet => packet, :timestamp => Time.now) + end + @sender.append_to_writing(packet) end MQTT_ERR_SUCCESS end def add_subscription(max_qos, packet_id, adjust_qos) - @suback_mutex.synchronize { + @suback_mutex.synchronize do adjust_qos, @waiting_suback = @waiting_suback.partition { |pck| pck[:id] == packet_id } - } + end if adjust_qos.length == 1 adjust_qos = adjust_qos.first[:packet].topics adjust_qos.each do |t| @@ -61,51 +65,54 @@ def add_subscription(max_qos, packet_id, adjust_qos) elsif max_qos[0] == 128 adjust_qos.delete(t) else - - @logger.error("The qos value is invalid in subscribe.") if PahoMqtt.logger? - raise PacketException + PahoMqtt.logger.error("The QoS value is invalid in subscribe.") if PahoMqtt.logger? + raise PacketException.new('Invalid suback QoS value') end end else - @logger.error("The packet id is invalid, already used.") if PahoMqtt.logger? - raise PacketException + PahoMqtt.logger.error("The packet id is invalid, already used.") if PahoMqtt.logger? + raise PacketException.new("Invalid suback packet id: #{packet_id}") end - @subscribed_mutex.synchronize { + @subscribed_mutex.synchronize do @subscribed_topics.concat(adjust_qos) - } + end return adjust_qos end def remove_subscription(packet_id, to_unsub) - @unsuback_mutex.synchronize { + @unsuback_mutex.synchronize do to_unsub, @waiting_unsuback = @waiting_unsuback.partition { |pck| pck[:id] == packet_id } - } - + end + if to_unsub.length == 1 to_unsub = to_unsub.first[:packet].topics else - @logger.error("The packet id is invalid, already used.") if PahoMqtt.logger? - raise PacketException + PahoMqtt.logger.error("The packet id is invalid, already used.") if PahoMqtt.logger? + raise PacketException.new("Invalid unsuback packet id: #{packet_id}") end - @subscribed_mutex.synchronize { + @subscribed_mutex.synchronize do to_unsub.each do |filter| @subscribed_topics.delete_if { |topic| PahoMqtt.match_filter(topic.first, filter) } end - } + end return to_unsub end - + def send_subscribe(topics, new_id) unless valid_topics?(topics) == MQTT_ERR_FAIL packet = PahoMqtt::Packet::Subscribe.new( - :id => new_id, + :id => new_id, :topics => topics - ) + ) @sender.append_to_writing(packet) - @suback_mutex.synchronize { - @waiting_suback.push({ :id => new_id, :packet => packet, :timestamp => Time.now }) - } + @suback_mutex.synchronize do + if @waiting_suback.length >= MAX_SUBACK + PahoMqtt.logger.error('SUBACK queue is full, could not send subscribe') if PahoMqtt.logger? + return MQTT_ERR_FAILURE + end + @waiting_suback.push(:id => new_id, :packet => packet, :timestamp => Time.now) + end MQTT_ERR_SUCCESS else raise ProtocolViolation @@ -115,14 +122,17 @@ def send_subscribe(topics, new_id) def send_unsubscribe(topics, new_id) unless valid_topics?(topics) == MQTT_ERR_FAIL packet = PahoMqtt::Packet::Unsubscribe.new( - :id => new_id, + :id => new_id, :topics => topics ) - @sender.append_to_writing(packet) - @unsuback_mutex.synchronize { - @waiting_unsuback.push({:id => new_id, :packet => packet, :timestamp => Time.now}) - } + @unsuback_mutex.synchronize do + if @waiting_suback.length >= MAX_UNSUBACK + PahoMqtt.logger.error('UNSUBACK queue is full, could not send unbsubscribe') if PahoMqtt.logger? + return MQTT_ERR_FAIL + end + @waiting_unsuback.push(:id => new_id, :packet => packet, :timestamp => Time.now) + end MQTT_ERR_SUCCESS else raise ProtocolViolation @@ -130,8 +140,12 @@ def send_unsubscribe(topics, new_id) end def check_waiting_subscriber - @sender.check_ack_alive(@waiting_suback, @suback_mutex, @waiting_suback.length) - @sender.check_ack_alive(@waiting_unsuback, @unsuback_mutex, @waiting_unsuback.length) + @sender.check_ack_alive(@waiting_suback, @suback_mutex) + @sender.check_ack_alive(@waiting_unsuback, @unsuback_mutex) + end + + def clear_queue + @waiting_suback = [] end def valid_topics?(topics) diff --git a/lib/paho_mqtt/version.rb b/lib/paho_mqtt/version.rb index 26d518e..9c6cdf7 100644 --- a/lib/paho_mqtt/version.rb +++ b/lib/paho_mqtt/version.rb @@ -1,3 +1,3 @@ module PahoMqtt - VERSION = "1.0.7" + VERSION = "1.0.12" end diff --git a/paho-mqtt.gemspec b/paho-mqtt.gemspec index 230d3fa..55fdb61 100644 --- a/paho-mqtt.gemspec +++ b/paho-mqtt.gemspec @@ -11,6 +11,7 @@ Gem::Specification.new do |spec| spec.summary = %q{A simple mqtt client gem} spec.description = %q{A simple mqtt client gem} + spec.homepage = "https://github.com/eclipse/paho.mqtt.ruby" spec.license = "MIT"