abstract class MQTT::ClientBase

Overview

The version agnostic half of an MQTT client: transport lifecycle, framing, the request pipeline, packet identifier allocation, the registry of responses we're waiting on, keep alive and reconnection.

Everything that depends on the wire format of a particular protocol version — building packets, parsing them, and what an acknowledgement means — is left to the subclass.

Direct Known Subclasses

Defined in:

mqtt/client_base.cr

Constant Summary

DEFAULT_TIMEOUT = 30.seconds

How long to wait for a broker response before giving up

Constructors

Instance Method Summary

Constructor Detail

def self.new(timeout : Time::Span | Nil, max_packet_size : UInt32, factory : Proc(Transport), reconnect : MQTT::Reconnect) #

Drives the client over transports produced by factory, re-establishing the connection whenever it drops


[View source]
def self.new(transport : Transport, timeout : Time::Span | Nil = DEFAULT_TIMEOUT, max_packet_size : UInt32 = MQTT::DEFAULT_MAX_PACKET_SIZE) #

Drives the client over a single transport, with no reconnection


[View source]

Instance Method Detail

def closed? #

[View source]
def last_ping_response : Time | Nil #

[View source]
def max_packet_size : UInt32 #

Largest packet we're willing to buffer from the broker


[View source]
abstract def parse_message(io) #

Handles a decoded packet. Implemented per protocol version


[View source]
abstract def ping(timeout : Time::Span | Nil) : Nil #

Sends whatever this protocol version uses for a liveness check, and waits for the response


[View source]
def terminated? : Bool #

[View source]
def timeout : Time::Span | Nil #

Applied to any request that isn't given an explicit timeout. Set to nil to wait indefinitely


[View source]
def timeout=(timeout : Time::Span | Nil) #

Applied to any request that isn't given an explicit timeout. Set to nil to wait indefinitely


[View source]
def wait_close : Nil #

Returns once the MQTT connection has terminated


[View source]