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 <p-goudet@ruby-dev.jp>
+#
+# 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"