Class: PahoMqtt::ConnectionHelper

Inherits:
Object
  • Object
show all
Defined in:
lib/paho_mqtt/connection_helper.rb

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(host, port, ssl, ssl_context, ack_timeout) ⇒ ConnectionHelper

Returns a new instance of ConnectionHelper.



22
23
24
25
26
27
28
29
30
31
 
# File 'lib/paho_mqtt/connection_helper.rb', line 22

def initialize(host, port, ssl, ssl_context, ack_timeout)
  @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)
end

Instance Attribute Details

#sender ⇒ Object

Returns the value of attribute sender.



20
21
22
 
# File 'lib/paho_mqtt/connection_helper.rb', line 20

def sender
  @sender
end

Instance Method Details

#check_keep_alive(persistent, last_ping_resp, keep_alive) ⇒ Object 



146
147
148
149
150
151
152
153
154
155
156
157
158
159
 
# File 'lib/paho_mqtt/connection_helper.rb', line 146

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?
    @sender.send_pingreq
  end
  timeout_resp = last_ping_resp + (keep_alive * 1.1).ceil
  if timeout_resp <= now
    PahoMqtt.logger.debug("No activity is over timeout, disconnecting from #{@host}.") if PahoMqtt.logger?
    @cs = MQTT_CS_DISCONNECT
  end
  @cs
end

#clean_start(host, port) ⇒ Object 



105
106
107
108
109
110
111
112
 
# File 'lib/paho_mqtt/connection_helper.rb', line 105

def clean_start(host, port)
  self.host = host
  self.port = port
  unless @socket.nil?
    @socket.close unless @socket.closed?
    @socket = nil
  end
end

#config_socket ⇒ Object 



80
81
82
83
84
85
86
87
88
89
90
91
92
 
# File 'lib/paho_mqtt/connection_helper.rb', line 80

def config_socket
  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
end

#do_connect(reconnection = false) ⇒ Object 



37
38
39
40
41
42
43
44
45
46
47
48
49
50
 
# File 'lib/paho_mqtt/connection_helper.rb', line 37

def do_connect(reconnection=false)
  @cs = MQTT_CS_NEW
  @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
    @cs = @handler.receive_packet
  end
  unless is_connected?
    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
end

#do_disconnect(publisher, explicit, mqtt_thread) ⇒ Object 



56
57
58
59
60
61
62
63
 
# File 'lib/paho_mqtt/connection_helper.rb', line 56

def do_disconnect(publisher, explicit, mqtt_thread)
  PahoMqtt.logger.debug("Disconnecting from #{@host}.") if PahoMqtt.logger?
  if explicit
    explicit_disconnect(publisher, mqtt_thread)
  end
  @socket.close unless @socket.nil? || @socket.closed?
  @socket = nil
end

#encrypted_socket(tcp_socket, ssl_context) ⇒ Object 



94
95
96
97
98
99
100
101
102
103
 
# File 'lib/paho_mqtt/connection_helper.rb', line 94

def encrypted_socket(tcp_socket, ssl_context)
  unless ssl_context.nil?
    @socket = OpenSSL::SSL::SSLSocket.new(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?
    raise Exception
  end
end

#explicit_disconnect(publisher, mqtt_thread) ⇒ Object 



65
66
67
68
69
70
 
# File 'lib/paho_mqtt/connection_helper.rb', line 65

def explicit_disconnect(publisher, mqtt_thread)
  @sender.flush_waiting_packet(false)
  send_disconnect
  mqtt_thread.kill if mqtt_thread && mqtt_thread.alive?
  publisher.flush_publisher unless publisher.nil?
end

#handler=(handler) ⇒ Object 



33
34
35
 
# File 'lib/paho_mqtt/connection_helper.rb', line 33

def handler=(handler)
  @handler = handler
end

#host=(host) ⇒ Object 



114
115
116
117
118
119
120
121
 
# File 'lib/paho_mqtt/connection_helper.rb', line 114

def host=(host)
  if host.nil? || host == ""
    PahoMqtt.logger.error("The host was found as nil while the connection setup.") if PahoMqtt.logger?
    raise ArgumentError
  else
    @host = host
  end
end

#is_connected? ⇒ Boolean

Returns:

  • (Boolean) 


52
53
54
 
# File 'lib/paho_mqtt/connection_helper.rb', line 52

def is_connected?
  @cs == MQTT_CS_CONNECTED
end

#port=(port) ⇒ Object 



123
124
125
126
127
128
129
130
 
# File 'lib/paho_mqtt/connection_helper.rb', line 123

def port=(port)
  if port.to_i <= 0
    PahoMqtt.logger.error("The port value is invalid (<= 0). Could not setup the connection.") if PahoMqtt.logger?
    raise ArgumentError
  else
    @port = port
  end
end

#send_connect(session_params) ⇒ Object 



132
133
134
135
136
137
138
 
# File 'lib/paho_mqtt/connection_helper.rb', line 132

def send_connect(session_params)
  setup_connection
  packet = PahoMqtt::Packet::Connect.new(session_params)
  @handler.clean_session = session_params[:clean_session]
  @sender.send_packet(packet)
  MQTT_ERR_SUCCESS
end

#send_disconnect ⇒ Object 



140
141
142
143
144
 
# File 'lib/paho_mqtt/connection_helper.rb', line 140

def send_disconnect
  packet = PahoMqtt::Packet::Disconnect.new
  @sender.send_packet(packet)
  MQTT_ERR_SUCCESS
end

#setup_connection ⇒ Object 



72
73
74
75
76
77
78
 
# File 'lib/paho_mqtt/connection_helper.rb', line 72

def setup_connection
  clean_start(@host, @port)
  config_socket
  unless @socket.nil?
    @sender.socket = @socket
  end
end