TAtlasMqttClient

Create constructor

constructor Create(AOwner: TComponent); override;

Creates a disconnected client. Defaults: Port 1883, KeepAlive 60, CleanSession True, AutoReconnect True, ReconnectDelayMs 1000, MaxReconnectDelayMs 30000, ConnectTimeoutMs 10000, MaxInFlight 1000, SynchronizeEvents True, WillQoS qosAtMostOnce, WillRetain False. Host is empty and must be set before Connect.

Client := TAtlasMqttClient.Create(nil);  // or pass a form, which then frees it

Destroy destructor

destructor Destroy; override;

Calls Disconnect (so the broker gets a proper goodbye and publishes no will), then frees the client. No event runs after the client is gone.

Client.Free;

State property

property State: TAtlasMqttState read GetState;

Where the connection stands: msDisconnected, msConnecting, msConnected or msWaitingToReconnect. See TAtlasMqttState. Safe to read from any thread.

if Client.State = msConnected then
  Client.Publish('heartbeat', 'x');

Password property

property Password: string read FPassword write FPassword;

Password for the broker, sent as UTF-8 in CONNECT. It is sent only when Username is not empty, because MQTT 3.1.1 does not allow a password without a user name. Public, not published, so it is never written into a form file; set it in code. It travels in clear text: there is no TLS.

Client.Password := 'from-your-settings-store';

Host property

property Host: string read FHost write FHost;

The broker's IPv4 address or host name. Names resolve to IPv4 only. Required: Connect raises when it is empty. It is read at every connection attempt.

Client.Host := 'broker.lan';

Port property

property Port: Word read FPort write FPort default 1883;

The broker's TCP port. 1883 is the standard port for MQTT without TLS.

Client.Port := Broker.ActualPort;

ClientId property

property ClientId: string read FClientId write FClientId;

The client identifier sent in CONNECT. Keep it unique: a broker disconnects an older connection when a second one arrives with the same identifier. Empty asks the broker to assign one, which MQTT 3.1.1 allows only with CleanSession True; Connect raises otherwise. The embedded broker names such clients auto-<ticks>-<thread>.

Client.ClientId := 'pump-3';

Username property

property Username: string read FUsername write FUsername;

User name sent in CONNECT when not empty. The embedded broker passes it to OnAuthenticate.

Client.Username := 'pump';

KeepAlive property

property KeepAlive: Integer read FKeepAlive write SetKeepAlive default 60;

Keep-alive period in seconds, sent in CONNECT; 0 turns keep-alive off. The client sends PINGREQ after KeepAlive seconds in which it sent nothing. If no PINGRESP comes back within half a period (at least one second), it treats the connection as lost ("The broker stopped answering") and reconnects. A broker may cut a client that stays silent for one and a half periods, and the embedded broker does. Values outside 0 to 65535 raise EAtlasMqttError. Set it before Connect.

Client.KeepAlive := 30;

CleanSession property

property CleanSession: Boolean read FCleanSession write FCleanSession default True;

Sent in CONNECT. True asks the broker to start a fresh session. False asks it to keep the session, which the embedded broker does not do; OnConnected reports the broker's answer in SessionPresent. With False, the client also keeps its record of half-finished incoming quality-2 messages across reconnects. The client itself stores nothing across restarts. False needs a non-empty ClientId.

Client.CleanSession := False;  // needs a ClientId

AutoReconnect property

property AutoReconnect: Boolean read FAutoReconnect write FAutoReconnect default True;

True: after a lost connection or a failed attempt, the client waits and tries again, indefinitely. A refusal by the broker (reported through OnError) is never retried. False: the client's thread ends after the first loss; call Disconnect and then Connect to start again.

Client.AutoReconnect := False;

ReconnectDelayMs property

property ReconnectDelayMs: Integer read FReconnectDelayMs write FReconnectDelayMs default 1000;

Milliseconds before the first attempt to reconnect. The delay doubles after each failed attempt up to MaxReconnectDelayMs, and starts again from this value after a successful connection. Values below 50 count as 50.

Client.ReconnectDelayMs := 500;

MaxReconnectDelayMs property

property MaxReconnectDelayMs: Integer read FMaxReconnectDelayMs write FMaxReconnectDelayMs default 30000;

The longest delay between reconnection attempts, in milliseconds. Values below 50 count as 50.

Client.MaxReconnectDelayMs := 10000;

ConnectTimeoutMs property

property ConnectTimeoutMs: Integer read FConnectTimeoutMs write FConnectTimeoutMs default 10000;

Milliseconds allowed for opening the TCP connection, and again for the broker's CONNACK after CONNECT was sent. When either runs out, the attempt fails ("Could not connect to host:port" or "The broker did not answer the connection request") and counts as a lost connection.

Client.ConnectTimeoutMs := 3000;

MaxInFlight property

property MaxInFlight: Integer read FMaxInFlight write SetMaxInFlight default 1000;

The most messages above quality 0 the client holds unacknowledged, including those waiting while offline. Publish raises EAtlasMqttError ("N messages are waiting for the broker; none more are accepted") beyond it. Values outside 1 to 65535 raise EAtlasMqttError.

Client.MaxInFlight := 10000;

WillTopic property

property WillTopic: string read FWillTopic write FWillTopic;

Topic of the will: a message the broker publishes if this client vanishes without saying goodbye, for example when the process dies, the network fails, or keep-alive runs out. Empty means no will. It must be a valid topic name, or Connect raises. Disconnect says goodbye, so no will is published then.

Client.WillTopic := 'status/pump-3';

WillMessage property

property WillMessage: string read FWillMessage write FWillMessage;

Payload of the will, sent as UTF-8.

Client.WillMessage := 'offline';

WillQoS property

property WillQoS: TMqttQoS read FWillQoS write FWillQoS default qosAtMostOnce;

Quality at which the broker publishes the will.

Client.WillQoS := qosAtLeastOnce;

WillRetain property

property WillRetain: Boolean read FWillRetain write FWillRetain default False;

True makes the broker retain the will, so later subscribers see it too. Combine it with a retained "online" status that the client publishes itself after each connect.

Client.WillRetain := True;

SynchronizeEvents property

property SynchronizeEvents: Boolean read FSynchronizeEvents write FSynchronizeEvents default True;

True: events are queued to the main thread (TThread.Queue) and run from the message loop, or when the main thread calls CheckSynchronize. False: events run on the client's own thread, and the handlers must be thread-safe. An exception that escapes a handler then ends the connection, which is reported through OnDisconnected and retried. Either way, OnCreateTransport runs on the client's thread. Set it before Connect.

Client.SynchronizeEvents := False;

OnConnected event

property OnConnected: TAtlasMqttConnectedEvent read FOnConnected write FOnConnected;

Fires after the broker accepted the connection: after the first connect and after every reconnect. By then the remembered subscriptions and held messages have been sent. SessionPresent is the broker's answer to CleanSession := False; the embedded broker always answers False.

Client.OnConnected := Handler.ClientConnected;

OnDisconnected event

property OnDisconnected: TAtlasMqttTextEvent read FOnDisconnected write FOnDisconnected;

Fires when a connection ends or an attempt fails, with the reason in words, for example "The broker closed the connection", "Could not connect to 10.0.0.5:1883" or "Bad user name or password". While the client keeps retrying, it fires once per failed attempt. It does not fire for Disconnect.

Client.OnDisconnected := Handler.ClientDisconnected;

OnMessage event

property OnMessage: TAtlasMqttMessageEvent read FOnMessage write FOnMessage;

Fires for each message delivered by a subscription. QoS is the quality it was delivered at. Retained is True when the broker replays a retained message to a new subscription; live messages arrive with it False. A quality-2 message that the broker sends twice is delivered once. Decode text with TEncoding.UTF8.GetString, which raises EEncodingError for bytes that are not valid UTF-8. With SynchronizeEvents False, an exception that escapes the handler ends the connection before the message is acknowledged: OnDisconnected reports the exception's message and the client reconnects. Catch exceptions inside the handler.

Client.OnMessage := Handler.ClientMessage;

OnPublished event

property OnPublished: TAtlasMqttPacketEvent read FOnPublished write FOnPublished;

Fires with the packet identifier that Publish returned, once the broker has acknowledged the message: PUBACK at quality 1, PUBCOMP at quality 2. Never fires for quality 0.

Client.OnPublished := Handler.ClientPublished;

OnSubscribed event

property OnSubscribed: TAtlasMqttSubscribedEvent read FOnSubscribed write FOnSubscribed;

Fires when the broker answers a subscription, including the automatic ones after a reconnect. Granted is False when the broker refused the filter. QoS is the granted quality, which may be lower than the requested one.

Client.OnSubscribed := Handler.ClientSubscribed;

OnError event

property OnError: TAtlasMqttTextEvent read FOnError write FOnError;

Fires once when the broker refuses the connection with a non-zero CONNACK return code, right after OnDisconnected and with the same reason (for example "Bad user name or password" or "Not authorised"). The client has stopped: fix the settings, call Disconnect, then Connect. Every refusal counts, including return code 3 (server unavailable). Other faults, such as a lost connection, missing pings or a broker that breaks the protocol, do not fire OnError: they end the connection and are retried when AutoReconnect is True.

Client.OnError := Handler.ClientError;

OnCreateTransport event

property OnCreateTransport: TAtlasMqttTransportEvent read FOnCreateTransport write FOnCreateTransport;

Called before every connection attempt, always on the client's own thread. Set Transport to a new, unconnected TAtlasMqttTransport descendant. The client calls its Connect(Host, Port, ConnectTimeoutMs), uses it, and frees it after the connection ends. Leave Transport nil to get a TAtlasTcpTransport.

Client.OnCreateTransport := Handler.CreateTransport;

Connect method

procedure Connect;

Starts the client's background thread, which connects and returns at once. OnConnected fires when the broker has accepted the connection, and State is msConnecting until then. Remembered subscriptions and unacknowledged messages go out right after the broker's acceptance. Raises EAtlasMqttError when Host is empty, when ClientId is empty and CleanSession is False, or when WillTopic is set but is not a valid topic name. Network problems are not raised: they arrive through OnDisconnected.

Does nothing while the component still holds a thread, including a thread that has already ended after a refusal or (with AutoReconnect False) after a lost connection. Call Disconnect first in that case.

Client.Connect;  // returns at once; OnConnected follows

ConnectAndWait method

function ConnectAndWait(TimeoutMs: Integer = 10000): Boolean;

Calls Connect, then waits up to TimeoutMs milliseconds for the broker to accept the connection. Returns True when connected. Returns False when the time runs out, or as soon as the client's thread ends because the broker refused the connection (or the attempt failed with AutoReconnect False). Running out of time does not stop the client: with AutoReconnect it keeps trying in the background. Call Disconnect to stop it.

When called on the main thread, it runs queued events (CheckSynchronize) while it waits, so it also works in a console program with SynchronizeEvents True.

if not Client.ConnectAndWait(5000) then
  Client.Disconnect;  // stop trying

Disconnect method

procedure Disconnect;

If connected, sends DISCONNECT, so the broker discards the will. Then closes the connection and waits for the client's thread to end, and State becomes msDisconnected. Events that were queued for the main thread but have not run yet are withdrawn. OnDisconnected does not fire for this orderly end. Remembered subscriptions and unacknowledged messages are kept, and go out again on the next Connect. Does nothing when the client was never started. Do not call it from an event handler while SynchronizeEvents is False: it waits for the thread that is running the handler.

Client.Disconnect;

Publish method

function Publish(const Topic: string; const Payload: TBytes; QoS: TMqttQoS = qosAtMostOnce; Retain: Boolean = False): Word; overload;

Publishes Payload (copied, may be empty) to Topic. Retain asks the broker to keep the message as the topic's last known value. An empty retained payload clears what the broker kept.

• Quality 0 is sent at once and returns 0. When there is no open connection, it raises EAtlasMqttError ("The client is not connected"). A failing send raises EAtlasMqttTransportError.
• Quality 1 and 2 return the packet identifier, 1 to 65535. The message is held until the broker acknowledges it (PUBACK at quality 1, PUBCOMP at quality 2), which OnPublished reports. While offline it waits in flight and goes out once connected, in publishing order. Messages sent after a reconnect, including ones published while offline, carry the duplicate flag. It raises EAtlasMqttError when MaxInFlight messages are already held.

A topic that is empty or contains +, # or U+0000 raises EAtlasMqttError. Safe to call from any thread.

Id := Client.Publish('lab/raw', TBytes.Create($01, $02, $03), qosAtLeastOnce);

Publish method

function Publish(const Topic, Text: string; QoS: TMqttQoS = qosAtMostOnce; Retain: Boolean = False): Word; overload;

As the TBytes overload, with Text encoded as UTF-8.

Client.Publish('home/lamp/state', 'on', qosAtLeastOnce, True);  // retained

Subscribe method

procedure Subscribe(const Filter: string; QoS: TMqttQoS = qosAtMostOnce);

Subscribes to Filter at up to quality QoS. + matches one topic level and #, which must come last, matches any number of levels. The filter is remembered until Unsubscribe, and sent again after every reconnect. Subscribing to the same filter again replaces its quality. Offline, including before Connect, the filter is only remembered and goes out once connected. OnSubscribed reports each acknowledgement, including those after a reconnect. An invalid filter raises EAtlasMqttError. Safe to call from any thread.

Client.Subscribe('home/+/temperature', qosAtLeastOnce);

Unsubscribe method

procedure Unsubscribe(const Filter: string);

Forgets Filter and, when connected, tells the broker. There is no event for the broker's acknowledgement. Messages already on their way may still arrive. The filter must be spelled exactly as it was subscribed; an invalid filter raises EAtlasMqttError.

Client.Unsubscribe('home/+/temperature');

WaitForIdle method

function WaitForIdle(TimeoutMs: Integer): Boolean;

Waits up to TimeoutMs milliseconds until no message above quality 0 is waiting for an acknowledgement, and returns True if that happened. On the main thread it runs queued events while it waits. It returns as soon as the count reaches zero, so OnPublished events for the last messages may still be queued. While offline it can only time out, unless nothing is in flight.

Ok := Client.WaitForIdle(3000);

InFlightCount method

function InFlightCount: Integer;

The number of messages above quality 0 not yet fully acknowledged, including those held while offline. Quality 2 messages count until PUBCOMP arrives.

if Client.InFlightCount > 100 then
  Text := 'the broker is falling behind';

Receive commands on a background thread

With SynchronizeEvents False, no message loop is needed, and the handler locks what it shares. Freeing the client disconnects it first, so the handler never runs after Destroy.

uses
  System.SysUtils, System.SyncObjs, Atlas.MqttCodec, Atlas.MqttClient;

type
  TCommandInbox = class
  private
    FLock: TCriticalSection;
    FClient: TAtlasMqttClient;
    FLast: string;
    procedure ClientMessage(Sender: TObject; const Topic: string; const Payload: TBytes;
      QoS: TMqttQoS; Retained: Boolean);
  public
    constructor Create(const Host: string);
    destructor Destroy; override;
    function LastCommand: string;
  end;

constructor TCommandInbox.Create(const Host: string);
begin
  inherited Create;
  FLock := TCriticalSection.Create;
  FClient := TAtlasMqttClient.Create(nil);
  FClient.Host := Host;
  FClient.ClientId := 'device-42';
  FClient.KeepAlive := 30;
  FClient.SynchronizeEvents := False;  // events on the client's thread
  FClient.OnMessage := ClientMessage;
  FClient.Subscribe('commands/device-42', qosAtLeastOnce);
  FClient.Connect;
end;

destructor TCommandInbox.Destroy;
begin
  FClient.Free;  // disconnects first; no event runs afterwards
  FLock.Free;
  inherited;
end;

procedure TCommandInbox.ClientMessage(Sender: TObject; const Topic: string; const Payload: TBytes;
  QoS: TMqttQoS; Retained: Boolean);
begin
  FLock.Enter;
  try
    FLast := TEncoding.UTF8.GetString(Payload);
  finally
    FLock.Leave;
  end;
end;

function TCommandInbox.LastCommand: string;
begin
  FLock.Enter;
  try
    Result := FLast;
  finally
    FLock.Leave;
  end;
end;

Credentials, will and reconnection settings

Everything a gateway needs before Connect. The password is set in code, because Password is not published and never stored in a form file.

uses
  Atlas.MqttCodec, Atlas.MqttClient;

procedure ConfigureGateway(Client: TAtlasMqttClient);
begin
  Client.Host := '10.0.0.5';
  Client.Port := 1883;
  Client.ClientId := 'gateway-north';
  Client.Username := 'gateway';
  Client.Password := 'from-your-settings-store';
  Client.KeepAlive := 20;               // ping after 20 s without traffic
  Client.ReconnectDelayMs := 2000;      // then 4 s, 8 s ... up to
  Client.MaxReconnectDelayMs := 60000;  // one minute between attempts
  Client.MaxInFlight := 5000;           // messages held while offline
  Client.WillTopic := 'status/gateway-north';
  Client.WillMessage := 'offline';
  Client.WillQoS := qosAtLeastOnce;
  Client.WillRetain := True;
end;