Class: Tina4::Mqtt

Inherits:
Object
  • Object
show all
Defined in:
lib/tina4/mqtt.rb

Overview

Zero-dependency MQTT 3.1.1 client — the protocol every broker and every IoT device already speaks. Built on socket and Array#pack/String#unpack only: no gem, so an app that talks to Mosquitto / EMQX / HiveMQ / AWS IoT adds nothing to its dependency tree.

Shaped like Tina4::Queue on purpose — publish / subscribe / consume:

mqtt = Tina4::Mqtt.new                          # TINA4_MQTT_URL
mqtt.publish("fleet/meter-42/telemetry", '{"kwh":12.5}', qos: 1)

mqtt.consume("fleet/+/telemetry", qos: 1) do |message|
next if message.duplicate?                    # QoS 1 is at-least-once
store(message.topic, message.payload)
end

Environment: TINA4_MQTT_URL (default mqtt://127.0.0.1:1883), TINA4_MQTT_CLIENT_ID, TINA4_MQTT_KEEPALIVE (seconds, default 60).

No background thread is started by default. When a connection is otherwise idle the broker will drop it past 1.5x keepalive, so opt in to the cooperative keepalive with start_keepalive — it registers a Tina4::Background task exactly like the queue consumers do, rather than spawning a thread of its own.

Single reader: like every MQTT client (paho included) the socket has ONE network reader. receive/consume from one thread; publish and the keepalive may be called from others (writes are serialised by a mutex).

Constant Summary collapse

CONNECT =

Control packet types. The low nibble of SUBSCRIBE/UNSUBSCRIBE is 0x2 because MQTT 3.1.1 mandates QoS 1 on those two packets.

0x10
CONNACK = 
0x20
PUBLISH = 
0x30
PUBACK = 
0x40
SUBSCRIBE = 
0x82
SUBACK = 
0x90
PINGREQ = 
0xC0
PINGRESP = 
0xD0
DISCONNECT = 
0xE0
PROTOCOL_LEVEL =

4 == MQTT 3.1.1

0x04
DEFAULT_PORT = 
1883
DEFAULT_TLS_PORT = 
8883
DEFAULT_URL = 
"mqtt://127.0.0.1:1883"
DEFAULT_KEEPALIVE = 
60
SUBSCRIPTION_REFUSED = 
0x80
MAX_REMAINING_LENGTH =

4 varint bytes

268_435_455
READ_CHUNK =

Read granularity. Always reading a full chunk keeps BOTH hidden buffers (Ruby's OpenSSL::Buffering @rbuf and OpenSSL's own SSL_pending) drained, so IO.select never blocks while decrypted plaintext is already waiting.

16_384
QOS2_REFUSED_MESSAGE =

QoS 2 is refused, never silently downgraded. A caller who asked for exactly-once and quietly got at-least-once would double-process every duplicate forever without ever seeing an error.

"MQTT QoS 2 (exactly-once delivery) is not supported by Tina4 — this " \
"client speaks QoS 0 and QoS 1 only, and refuses QoS 2 rather than " \
"silently downgrading it to QoS 1. Use QoS 1 with an idempotent " \
"consumer keyed on (device_id, device_timestamp): duplicates are then " \
"harmless, which is what exactly-once was for."
CONNACK_RETURN_CODES =

Human-readable CONNACK return codes (MQTT 3.1.1 section 3.2.2.3).

{
  1 => "unacceptable protocol version",
  2 => "client identifier rejected",
  3 => "server unavailable",
  4 => "bad user name or password",
  5 => "not authorised"
}.freeze

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(url: nil, client_id: nil, username: nil, password: nil, ca_file: nil, tls_verify: nil, keepalive: nil, clean_session: true, will_topic: nil, will_payload: nil, will_qos: 0, will_retain: false, timeout: 5, read_timeout: nil, connect: true) ⇒ Mqtt

url: mqtt://host:port, or mqtts://host:port for TLS. Credentials may be carried in the userinfo (mqtt://user:pass@host), and are percent-decoded, so a password may contain @ : or /. Falls back to TINA4_MQTT_URL, then localhost. client_id: broker-visible session id (TINA4_MQTT_CLIENT_ID, else generated). A durable session (clean_session: false) needs a STABLE id — the generated one changes every process. username: broker username; overrides the url's userinfo when given password: broker password; MQTT 3.1.1 forbids one without a username ca_file: PEM CA bundle used to verify the broker's certificate (TINA4_MQTT_CA_FILE). Needed for a private or self-signed CA. tls_verify: false disables certificate verification (TINA4_MQTT_TLS_VERIFY). NEVER the default, and it logs a warning naming the risk — an unverified TLS session is encrypted but not authenticated, so a man in the middle can read and rewrite it. keepalive: seconds (TINA4_MQTT_KEEPALIVE, default 60) clean_session: false keeps the subscription + queues QoS 1 messages for us while we are offline, and replays them on reconnect will_*: Last Will, published by the broker when we die WITHOUT a DISCONNECT — dead-device detection that is observed, not inferred from silence timeout: seconds to wait for a control-packet answer (CONNACK / PUBACK / SUBACK) read_timeout: seconds to wait for an application message; nil blocks connect: false builds the client without opening a socket



111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
 
# File 'lib/tina4/mqtt.rb', line 111

def initialize(url: nil, client_id: nil, username: nil, password: nil,
               ca_file: nil, tls_verify: nil, keepalive: nil, clean_session: true,
               will_topic: nil, will_payload: nil, will_qos: 0, will_retain: false,
               timeout: 5, read_timeout: nil, connect: true)
  parsed = self.class.parse_url(url || ENV["TINA4_MQTT_URL"] || DEFAULT_URL)
  @host = parsed[:host]
  @port = parsed[:port]
  @tls = parsed[:tls]
  # Explicit arguments win over the url's userinfo: the more specific source.
  @username = username || parsed[:username]
  @password = password || parsed[:password]
  if @password && (@username.nil? || @username.empty?)
    raise ArgumentError,
          "MQTT password without a username is not allowed by MQTT 3.1.1 — " \
          "supply both (mqtt://user:pass@host, or username:/password:) or neither"
  end
  @ca_file = ca_file || presence(ENV["TINA4_MQTT_CA_FILE"])
  @tls_verify = tls_verify.nil? ? Tina4::Env.is_truthy(ENV.fetch("TINA4_MQTT_TLS_VERIFY", "true")) : tls_verify
  @client_id = client_id || ENV["TINA4_MQTT_CLIENT_ID"]
  @client_id = "tina4-#{SecureRandom.hex(8)}" if @client_id.nil? || @client_id.empty?
  @keepalive = (keepalive || ENV["TINA4_MQTT_KEEPALIVE"] || DEFAULT_KEEPALIVE).to_i
  @clean_session = clean_session
  @will_topic = will_topic
  @will_payload = will_payload
  @will_qos = will_qos
  @will_retain = will_retain
  @timeout = timeout
  @read_timeout = read_timeout

  refuse_unsupported_qos(will_qos) if will_topic

  @packet_id = 0
  @inbox = []
  @write_mutex = Mutex.new
  @last_write_at = 0.0
  @socket = nil
  @keepalive_task = nil
  @read_buffer = String.new(capacity: READ_CHUNK, encoding: Encoding::BINARY)
  @read_cursor = 0

  self.connect if connect
end

Instance Attribute Details

#clean_session ⇒ Object (readonly)

Returns the value of attribute clean_session.



85
86
87
 
# File 'lib/tina4/mqtt.rb', line 85

def clean_session
  @clean_session
end

#client_id ⇒ Object (readonly)

Returns the value of attribute client_id.



85
86
87
 
# File 'lib/tina4/mqtt.rb', line 85

def client_id
  @client_id
end

#host ⇒ Object (readonly)

Returns the value of attribute host.



85
86
87
 
# File 'lib/tina4/mqtt.rb', line 85

def host
  @host
end

#keepalive ⇒ Object (readonly)

Returns the value of attribute keepalive.



85
86
87
 
# File 'lib/tina4/mqtt.rb', line 85

def keepalive
  @keepalive
end

#port ⇒ Object (readonly)

Returns the value of attribute port.



85
86
87
 
# File 'lib/tina4/mqtt.rb', line 85

def port
  @port
end

#username ⇒ Object (readonly)

Returns the value of attribute username.



85
86
87
 
# File 'lib/tina4/mqtt.rb', line 85

def username
  @username
end

Class Method Details

.encode_remaining_length(value) ⇒ Object

Remaining Length: 7 bits per byte, high bit means "another byte follows". A single-byte assumption works for every packet under 128 bytes and then fails, so this is exercised directly at 0 / 127 / 128 / 16383.



212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
 
# File 'lib/tina4/mqtt.rb', line 212

def self.encode_remaining_length(value)
  length = Integer(value)
  if length.negative? || length > MAX_REMAINING_LENGTH
    raise ArgumentError, "remaining length #{length} is outside 0..#{MAX_REMAINING_LENGTH}"
  end

  encoded = String.new(capacity: 4, encoding: Encoding::BINARY)
  loop do
    byte = length & 0x7F
    length >>= 7
    encoded << (length.positive? ? (byte | 0x80) : byte)
    break unless length.positive?
  end
  encoded
end

.parse_url(url) ⇒ Object

Split an MQTT url into its parts:

{ host:, port:, tls:, username:, password: }

"mqtt://host:port" and "tcp://host:port" are plain TCP (default port 1883), "mqtts://host:port" is TLS (default port 8883). A bare "host" or "host:port" works too, and an IPv6 literal is bracketed ("mqtt://[::1]:1883"). Credentials ride in the userinfo and are percent-decoded, so a password containing @ : or / survives.

Raises:

  • (ArgumentError) 


162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
 
# File 'lib/tina4/mqtt.rb', line 162

def self.parse_url(url)
  raw = url.to_s.strip
  raise ArgumentError, "MQTT url is empty — set TINA4_MQTT_URL (e.g. #{DEFAULT_URL})" if raw.empty?

  scheme = raw[%r{\A([A-Za-z][A-Za-z0-9+.-]*)://}, 1]&.downcase
  unless scheme.nil? || %w[mqtt tcp mqtts].include?(scheme)
    raise ArgumentError,
          "unsupported MQTT url scheme #{scheme.inspect} in #{raw.inspect} — " \
          "this client speaks mqtt://, tcp:// or mqtts:// (TLS). " \
          "WebSocket transports are not implemented."
  end

  tls = scheme == "mqtts"
  rest = scheme ? raw.sub(%r{\A[A-Za-z][A-Za-z0-9+.-]*://}, "") : raw

  # Split on the LAST "@" so a password containing an un-encoded "@" still
  # leaves the host intact.
  username = password = nil
  if rest.include?("@")
    userinfo, _, rest = rest.rpartition("@")
    raw_username, has_password, raw_password = userinfo.partition(":")
    username = percent_decode(raw_username)
    password = percent_decode(raw_password) unless has_password.empty?
  end

  match = rest.match(%r{\A(?<host>\[[^\]]+\]|[^:/]+)(?::(?<port>\d+))?(?:/.*)?\z})
  raise ArgumentError, "malformed MQTT url #{raw.inspect} — expected mqtt://host:port" unless match

  {
    host: match[:host].delete_prefix("[").delete_suffix("]"),
    port: (match[:port] || (tls ? DEFAULT_TLS_PORT : DEFAULT_PORT)).to_i,
    tls: tls,
    username: presence(username),
    password: password
  }
end

.percent_decode(value) ⇒ Object

%XX in url userinfo. Not CGI.unescape / URI.decode_www_form_component: those also turn "+" into a space, which silently corrupts a password.



201
202
203
 
# File 'lib/tina4/mqtt.rb', line 201

def self.percent_decode(value)
  value.to_s.gsub(/%([0-9A-Fa-f]{2})/) { ::Regexp.last_match(1).hex.chr }
end

.presence(value) ⇒ Object 



205
206
207
 
# File 'lib/tina4/mqtt.rb', line 205

def self.presence(value)
  value.nil? || value.empty? ? nil : value
end

Instance Method Details

#acknowledge(packet_id) ⇒ Object

PUBACK a QoS 1 delivery. Called by MqttMessage#acknowledge.



396
397
398
399
 
# File 'lib/tina4/mqtt.rb', line 396

def acknowledge(packet_id)
  write_packet(PUBACK, [packet_id].pack("n"))
  true
end

#cipher ⇒ Object

The negotiated cipher suite name, or nil on a plain connection. A real name here is proof the TLS handshake actually completed.



284
285
286
287
288
 
# File 'lib/tina4/mqtt.rb', line 284

def cipher
  return nil unless @tls && connected?

  @socket.cipher&.first
end

#connect ⇒ Object

Open the socket and complete the CONNECT / CONNACK handshake. Also the reconnect path: an existing socket is closed first, and a durable session (clean_session: false) resumes with the same client_id.



231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
 
# File 'lib/tina4/mqtt.rb', line 231

def connect
  close_socket
  raw_socket = Socket.tcp(@host, @port, connect_timeout: @timeout)
  # Telemetry frames are tiny; Nagle would add latency for no gain.
  raw_socket.setsockopt(Socket::IPPROTO_TCP, Socket::TCP_NODELAY, 1)
  @socket = @tls ? wrap_in_tls(raw_socket) : raw_socket
  @inbox.clear
  @read_buffer.clear
  @read_cursor = 0

  body = String.new(capacity: 64, encoding: Encoding::BINARY)
  body << mqtt_string("MQTT")
  body << [PROTOCOL_LEVEL, connect_flags].pack("C2")
  body << [@keepalive].pack("n")
  # Payload order is FIXED: client id, will topic, will message, username,
  # password. Emitting them in any other order shifts every field after it.
  body << mqtt_string(@client_id)
  if @will_topic
    body << mqtt_string(@will_topic)
    will_bytes = payload_bytes(@will_payload)
    body << [will_bytes.bytesize].pack("n") << will_bytes
  end
  body << mqtt_string(@username) if @username
  body << mqtt_string(@password) if @password
  write_packet(CONNECT, body)

  header, payload = read_packet(deadline_in(@timeout))
  unless header == CONNACK && payload.bytesize >= 2
    raise MqttError, format("expected CONNACK, got %#04x", header)
  end

  return_code = payload.getbyte(1)
  unless return_code.zero?
    reason = CONNACK_RETURN_CODES.fetch(return_code, "unknown return code")
    raise MqttError, "broker refused the connection: #{reason} (CONNACK return code #{return_code})"
  end
  true
rescue SystemCallError, IOError => e
  close_socket
  raise MqttError, "could not connect to MQTT broker at #{@host}:#{@port}: #{e.message}"
end

#connected? ⇒ Boolean

Returns:

  • (Boolean) 


273
274
275
 
# File 'lib/tina4/mqtt.rb', line 273

def connected?
  !@socket.nil? && !@socket.closed?
end

#consume(topic_filter = nil, qos: 1, iterations: 0, timeout: nil, &block) ⇒ Object

Long-running consumer, mirroring Queue#consume.

mqtt.consume("fleet/+/telemetry", qos: 1) { |message| store(message) }
mqtt.consume.each { |message| store(message) }   # already subscribed

The message is acknowledged AFTER the block returns, so a handler that raises leaves the message unacknowledged and the broker redelivers it (with DUP set) — at-least-once, which is the point of QoS 1. The raise propagates: a consumer that swallows storage failures loses data quietly.

iterations: > 0 stops after that many messages (bounded runs and tests).



384
385
386
387
388
389
390
391
392
393
 
# File 'lib/tina4/mqtt.rb', line 384

def consume(topic_filter = nil, qos: 1, iterations: 0, timeout: nil, &block)
  subscribe(topic_filter, qos: qos) if topic_filter
  unless block
    return Enumerator.new do |yielder|
      consume_loop(iterations, timeout) { |message| yielder << message }
    end
  end

  consume_loop(iterations, timeout, &block)
end

#disconnect ⇒ Object

Say goodbye properly: DISCONNECT then close. The broker discards the Last Will on a graceful disconnect, and a clean_session: false session survives for the next connect with the same client_id.



451
452
453
454
455
456
457
458
459
460
461
 
# File 'lib/tina4/mqtt.rb', line 451

def disconnect
  stop_keepalive
  begin
    write_packet(DISCONNECT, "") if connected?
  rescue MqttError, SystemCallError, IOError
    # Already gone — closing is still the right outcome.
  ensure
    close_socket
  end
  true
end

#inspect ⇒ Object

Never dump @password: a client can end up in a log line, an exception report or the debug overlay, and Ruby's default inspect prints every ivar.



299
300
301
302
303
 
# File 'lib/tina4/mqtt.rb', line 299

def inspect
  format("#<%s %s://%s:%d client_id=%s%s connected=%s>",
         self.class.name, @tls ? "mqtts" : "mqtt", @host, @port,
         @client_id.inspect, @username ? " username=#{@username.inspect}" : "", connected?)
end

#kill ⇒ Object

Drop the socket WITHOUT a DISCONNECT — what a crashed or unplugged device looks like to the broker, and therefore what fires the Last Will.



465
466
467
468
469
 
# File 'lib/tina4/mqtt.rb', line 465

def kill
  stop_keepalive
  close_socket
  true
end

#ping(timeout: nil) ⇒ Object

PINGREQ and wait for the PINGRESP. Use this when nothing else is reading the socket; under a consume loop use start_keepalive instead, because the reader there absorbs the PINGRESP.



404
405
406
407
408
409
410
411
412
413
414
 
# File 'lib/tina4/mqtt.rb', line 404

def ping(timeout: nil)
  send_keepalive
  deadline = deadline_in(timeout || @timeout)
  loop do
    header, payload = read_packet(deadline)
    return true if header == PINGRESP
    next if stash_publish(header, payload)

    raise MqttError, format("expected PINGRESP, got %#04x", header)
  end
end

#publish(topic, payload, qos: 0, retain: false) ⇒ Object

Publish an application message. Returns the packet identifier for QoS 1 (the broker's PUBACK must carry it back) and nil for QoS 0.

retain: true tells the broker to keep this as the topic's last known value and hand it to every FUTURE subscriber — how a dashboard shows current state the moment it connects. Publishing an EMPTY payload with retain: true is how a retained value is cleared.



312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
 
# File 'lib/tina4/mqtt.rb', line 312

def publish(topic, payload, qos: 0, retain: false)
  refuse_unsupported_qos(qos)
  bytes = payload_bytes(payload)
  topic_field = mqtt_string(topic)
  # The packet identifier exists ONLY when QoS > 0. Emitting it at QoS 0 (or
  # omitting it at QoS 1) desynchronises the stream and every later packet
  # mis-parses.
  packet_id = qos.positive? ? next_packet_id : nil

  body = String.new(capacity: topic_field.bytesize + bytes.bytesize + 2,
                    encoding: Encoding::BINARY)
  body << topic_field
  body << [packet_id].pack("n") if packet_id
  body << bytes
  write_packet(PUBLISH | (qos << 1) | (retain ? 0x01 : 0x00), body)

  wait_for_acknowledgement(PUBACK, packet_id, "PUBACK") if qos == 1
  packet_id
end

#receive(timeout: nil, ack: true) ⇒ Object

Read the next application message. Returns a Tina4::MqttMessage.

ack: true (the default) acknowledges a QoS 1 delivery immediately, which is right for a synchronous read. Pass ack: false when the message must be stored before the broker is allowed to forget it — an unacknowledged QoS 1 message is redelivered with DUP set. consume does exactly that.



367
368
369
370
371
 
# File 'lib/tina4/mqtt.rb', line 367

def receive(timeout: nil, ack: true)
  message = @inbox.shift || read_publish(deadline_in(timeout || @read_timeout))
  message.acknowledge if ack
  message
end

#send_keepalive ⇒ Object

Write a PINGREQ without waiting for the answer. The PINGRESP is absorbed by whatever is reading the socket (receive skips it), which is what makes the background keepalive safe alongside a consume loop.



419
420
421
422
 
# File 'lib/tina4/mqtt.rb', line 419

def send_keepalive
  write_packet(PINGREQ, "")
  true
end

#start_keepalive(interval: nil) ⇒ Object

Opt in to the cooperative keepalive. Registers a Tina4::Background task — the same mechanism the queue consumers use — that sends a PINGREQ only when the connection has actually gone quiet, so an actively publishing client costs no extra packets. Without a keepalive the broker drops a silent client past 1.5x keepalive (which is exactly what fires the Last Will).

Raises:



430
431
432
433
434
435
436
437
438
 
# File 'lib/tina4/mqtt.rb', line 430

def start_keepalive(interval: nil)
  return @keepalive_task if @keepalive_task
  raise MqttError, "keepalive is disabled (keepalive: 0) — nothing to schedule" unless @keepalive.positive?

  seconds = interval || [@keepalive / 2.0, 1.0].max
  @keepalive_task = Tina4::Background.register(interval: seconds) do
    send_keepalive if connected? && idle_for?(seconds)
  end
end

#stop_keepalive ⇒ Object 



440
441
442
443
444
445
446
 
# File 'lib/tina4/mqtt.rb', line 440

def stop_keepalive
  return false unless @keepalive_task

  Tina4::Background.stop_task(@keepalive_task)
  @keepalive_task = nil
  true
end

#subscribe(topic_filter, qos: 1) ⇒ Object

Subscribe to a topic filter ("fleet/+/telemetry", "fleet/#"). Returns the QoS the broker GRANTED, which can be lower than the one requested.

A SUBACK carrying 0x80 is a REFUSAL, not a success — treating any SUBACK as success means sitting on a dead subscription receiving nothing, so it raises here.

Raises:



338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
 
# File 'lib/tina4/mqtt.rb', line 338

def subscribe(topic_filter, qos: 1)
  refuse_unsupported_qos(qos)
  packet_id = next_packet_id
  filter_field = mqtt_string(topic_filter)

  body = String.new(capacity: filter_field.bytesize + 3, encoding: Encoding::BINARY)
  body << [packet_id].pack("n")
  body << filter_field
  body << qos
  write_packet(SUBSCRIBE, body)

  payload = wait_for_acknowledgement(SUBACK, packet_id, "SUBACK")
  raise MqttError, "malformed SUBACK: no return code for #{topic_filter.inspect}" if payload.bytesize < 3

  granted = payload.getbyte(2)
  if granted == SUBSCRIPTION_REFUSED
    raise MqttError,
          "broker refused the subscription to #{topic_filter.inspect} " \
          "(SUBACK return code 0x80) — check the topic filter and the broker ACLs"
  end
  granted
end

#tls? ⇒ Boolean

True when this connection runs over TLS (mqtts://).

Returns:

  • (Boolean) 


278
279
280
 
# File 'lib/tina4/mqtt.rb', line 278

def tls?
  @tls
end

#tls_version ⇒ Object

The negotiated TLS protocol version ("TLSv1.3"), or nil when plain.



291
292
293
294
295
 
# File 'lib/tina4/mqtt.rb', line 291

def tls_version
  return nil unless @tls && connected?

  @socket.ssl_version
end