Client

The MQTT client in Julia facilitates communication between a device and an MQTT broker over a network. It manages connections, message handling, and maintains the state of communication. The client operates through three main loops: the read loop listens for incoming messages from the broker and processes them using designated handlers; the write loop sends packets to the broker from a queue, ensuring thread safety with a socket lock; and the keep-alive loop periodically sends ping requests to the broker to maintain the connection and detect disconnections. This client uses atomic operations to ensure thread safety for shared variables and supports asynchronous task management for efficient, non-blocking operations.

Fields

• state::UInt8: client state.
• on_msg::TrieNode: A trie mapping topics to callback functions.
• keep_alive::UInt16: The keep-alive time in seconds.
• last_id::UInt16: The last packet identifier used.
• in_flight::Dict{UInt16, Future}: A dictionary mapping packet identifiers to futures.
• write_packets::AbstractChannel: A channel for writing packets.
• socket: The socket used for communication with the broker.
• socket_lock: A lock for synchronizing access to the socket.
• ping_timeout::UInt64: The ping timeout in seconds.
• ping_outstanding::Atomic{UInt8}: An atomic counter for the number of outstanding ping requests.
• last_sent::Atomic{Float64}: An atomic float representing the timestamp of the last sent packet.
• last_received::Atomic{Float64}: An atomic float representing the timestamp of the last received packet.

Constructor

Client(ping_timeout::UInt64=UInt64(60)) constructs a new Client object with the specified ping timeout (default: 60 seconds).

Connection{T <: AbstractIOConnection}

The Connection struct in Julia encapsulates the configuration and connection details required for an MQTT client to connect to an MQTT broker. This struct supports two types of connection protocols: TCP and Unix Domain Sockets (UDS), both of which are subtypes of AbstractIOConnection. The struct includes fields for protocol type, keep-alive interval, client ID, user credentials, a will message (a message that is sent by the broker if the client disconnects unexpectedly), and a flag indicating whether the session is clean (i.e., no persistent session state). The Connection constructor allows for flexible instantiation with default or specified values for each field, enabling easy setup of connection parameters tailored to the specific requirements of the MQTT client and broker interaction.

Fields

• protocol::T: The underlying IO connection.
• keep_alive::Int64: The keep-alive time in seconds.
• client_id::String: The client identifier.
• user::User: The user credentials.
• will::Message: The last will and testament message.
• clean_session::Bool: Whether to start a clean session.

Constructors

Connection(protocol::T; keep_alive::Int64=32, client_id::String=randstring(8), user::User=User("", ""), will::Message=Message(false, 0x00, false, "", UInt8[]), clean_session::Bool=true) where T <: AbstractIOConnection constructs a new Connection object with the specified protocol and optional keyword arguments.

Connection(protocol::T, keep_alive::Int64, client_id::String, user::User, will::Message, clean_session::Bool) where T <: AbstractIOConnection constructs a new Connection object with the specified arguments.

Example using TCP protocol with default and custom values

tcp_connection = Connection(
    TCP(Sockets.localhost, 1883);       # Using TCP with localhost and port 1883
    keep_alive=60,                      # Custom keep-alive interval of 60 seconds
    client_id="my_mqtt_client",         # Custom client ID
    user=User("username", "password"),  # Custom user credentials
    will=Message(false, 0x01, false, "last/will/topic", UInt8[]),  # Custom will message
    clean_session=true,                  # Default clean session flag
)

Example using UDS protocol with all custom values

uds_connection_full = Connection(
    UDS("/var/run/mqtt.sock"),          # Using UDS with specified socket path
    45,                                 # Custom keep-alive interval of 45 seconds
    "another_client",                   # Custom client ID
    User("user", "pass"),               # Custom user credentials
    Message(true, 0x00, true, "will/topic", UInt8[1, 2, 3]),  # Custom will message
    false,                               # Custom clean session flag
)

IOConnection(ip::IPAddr, port::Int64)

Constructs a new TCP object with the specified IP address and port number.

Arguments

• ip::IPAddr: The IP address of the remote host.
• port::Int64: The port number on the remote host.

Returns

• TCP: A new TCP object.

• IOConnection(ip::String, port::Int64)

Constructs a new TCP object with the specified IP address and port number.

Arguments

• ip::String: The IP address of the remote host as a string.
• port::Int64: The port number on the remote host.

Returns

• TCP: A new TCP object.

• IOConnection(path::AbstractString)

Constructs a new UDS object with the specified file system path.

Arguments

• path::AbstractString: The file system path of the socket.

Returns

• UDS: A new UDS object.

Message

A composite type representing a message.

Fields

• dup::Bool: a boolean indicating whether the message is a duplicate.
• qos::UInt8: an 8-bit unsigned integer representing the quality of service.
• retain::Bool: a boolean indicating whether the message should be retained.
• topic::String: a string representing the topic.
• payload::Array{UInt8}: an array of 8-bit unsigned integers representing the payload.

Constructors

• Message(qos::QOS, topic::String, payload...): constructs a new message with default values for dup and retain.
• Message(dup::Bool, qos::QOS, retain::Bool, topic::String, payload...): constructs a new message with all fields specified.
• Message(dup::Bool, qos::UInt8, retain::Bool, topic::String, payload...): constructs a new message with all fields specified.

User(name::String, password::String)

A struct that represents a user with a name and password.

Examples

user = User("John", "password")

User("John", "password")

Configuration

Container for the mqtt client and mqtt connection data. This is partially iterable, and can be spread to 2 variables with the ... splat operator or client, conn = conf variable assignment.

Example

# using the MakeConnection interface function
config = MakeConnection("/temp/mqtt.sock")

# using a defined IO
io = IOConnection("localhost", 1883)
config = Configuration(io)

# spreading the variables
client, connection = Configuration(...)