YOUSEFUL / DELPHI

ATLAS MQTT 2 components · Loopback publish and subscribe · Reconnect, will and retained messages

Atlas MQTT: FAQ, examples and tutorial

Learning center · Component reference

Atlas MQTT has two parts. TAtlasMqttClient is an MQTT 3.1.1 client component for programs that publish and subscribe. TAtlasMqttBroker is a small broker that runs inside your own program, for tests, device simulators, demos, or a message bus between the parts of one product on one machine. Both are on the Atlas MQTT palette page. Add Atlas.MqttClient or Atlas.MqttBroker to uses. Also add Atlas.MqttCodec, which declares the quality-of-service values (qosAtMostOnce, qosAtLeastOnce, qosExactlyOnce) and EAtlasMqttError. You need Atlas.MqttTransport only when you supply your own transport. The product depends on nothing but the RTL. It runs on Windows (Win32 and Win64) over IPv4 TCP.

FAQ

Which version of MQTT does Atlas MQTT implement? MQTT 3.1.1 (protocol level 4). It covers all fourteen packet types and qualities 0, 1 and 2 in both directions. MQTT 5.0 is not implemented, so there are no properties, reason codes, shared subscriptions, topic aliases or message expiry. The client always asks for level 4. The embedded broker answers any other level, MQTT 3.1 included, with return code 1 ("The server does not support this protocol level") and closes the connection.

Can I connect over TLS or WebSockets? No. The only transport included is TAtlasTcpTransport: plain TCP over IPv4. Host names resolve to IPv4 addresses only. The user name, password and payloads all travel unencrypted, so use it on loopback or on a network you trust. OnCreateTransport lets you plug in your own transport derived from TAtlasMqttTransport, for example one that wraps a TLS library, but none ships with the product.

Does the client or the broker keep sessions? Not across restarts. The client keeps its subscriptions and unacknowledged messages in memory. They survive a reconnect, and even Disconnect followed by Connect on the same component. They are lost when the component is freed or the process ends. The embedded broker treats every session as clean:

• CONNACK always says "no session present".
• Nothing is queued for a subscriber that is not connected.
• An unacknowledged delivery is not sent again.
• Retained messages are kept only while the broker runs; Stop discards them.

CleanSession := False is passed on as asked. It makes a difference only with a broker that stores sessions.

Do I need a separate MQTT server? Not for tests, simulators, demos or a message bus on one machine. Set Port := 0 on a TAtlasMqttBroker, call Start and read ActualPort. By default it accepts connections from this machine only. It is deliberately small:

• no persistence, no TLS;
• no per-topic access control, bridging or $SYS statistics;
• messages are routed under one lock, so one stalled client delays the others.

Do not expose it to the internet. For a production network, use a dedicated broker.

On which thread do the events run? With SynchronizeEvents True (the default), the client queues its events to the main thread. In a VCL application they run from the message loop and may touch controls. A console program or a service must call CheckSynchronize for them to run. ConnectAndWait and WaitForIdle do this while they wait, when they are called on the main thread. With SynchronizeEvents False, the events run on the client's own thread. That needs no message loop, but anything the handlers share must be locked. OnCreateTransport always runs on the client's thread. The broker's OnAuthenticate runs on the thread of the connection being checked. Publish, Subscribe, Unsubscribe, State and InFlightCount may be called from any thread. Call Connect and Disconnect from the thread that owns the component.

What happens when the connection drops? The client fires OnDisconnected with the reason. If AutoReconnect is True (the default), it tries again after ReconnectDelayMs (1000 ms by default), doubling the delay after each failed attempt up to MaxReconnectDelayMs (30 000 ms). Once connected again, it:

1. subscribes again to every remembered filter;
2. sends every unacknowledged message again, flagged as a duplicate, in the original order;
3. fires OnConnected.

OnSubscribed then follows for each filter as the broker acknowledges it. After a successful connection the delay starts again from ReconnectDelayMs.

Can I publish while the connection is down? At quality 1 and 2, yes. The message is held in flight (up to MaxInFlight, 1000 by default) and sent in publishing order once the client is connected. It is sent with the duplicate flag set, like any message the client sends again. At quality 0, Publish raises EAtlasMqttError ("The client is not connected"): a message the protocol never retries would otherwise vanish without notice. Held messages live in memory only.

Why does the client stop after the broker refuses it? Any CONNACK return code other than 0 counts as a refusal: wrong protocol level, rejected client identifier, server unavailable, bad user name or password, or not authorised. The client reports it once through OnDisconnected and then OnError, with the reason in words, and its thread ends instead of retrying. Correct the settings, call Disconnect to release the stopped client, then call Connect again. Other faults are retried when AutoReconnect is True: a lost connection, a broker that stops answering pings, or a broker that breaks the protocol.

How do I know the broker received a message? At quality 1 and 2, Publish returns the packet identifier. OnPublished reports that identifier when the broker acknowledges the message (PUBACK at quality 1, PUBCOMP at quality 2). WaitForIdle waits until nothing is unacknowledged, and InFlightCount tells you how many messages are waiting. At quality 0, Publish returns 0 and there is no acknowledgement. An acknowledgement means the broker has the message, not that any subscriber received it.

At what quality does a subscriber receive a message? At the lower of two qualities: the one the message was published with and the one the subscription was granted. The embedded broker grants every subscription at the quality requested. Other brokers may grant less, and OnSubscribed reports what was granted. When several of a client's filters match one topic, the embedded broker delivers the message once, at the highest quality among them.

Why is Password missing from the Object Inspector? Password is public, not published, so it is never written into a form file. Set it in code. The client sends it only together with a non-empty Username, because MQTT 3.1.1 does not allow a password without a user name. It travels in clear text.

Which platforms are supported? Windows only, Win32 and Win64, in VCL and console programs; the transport uses Winsock. Transports for macOS, Linux, Android and iOS are not implemented, and neither are FMX support, IPv6 or proxies.

Examples

A VCL form that subscribes and lists what arrives. The form owns the client, so the client is freed with the form. Its destructor calls Disconnect.

uses
  System.SysUtils, System.Classes, Vcl.Controls, Vcl.Forms, Vcl.StdCtrls,
  Atlas.MqttCodec, Atlas.MqttClient;

type
  TMainForm = class(TForm)
  private
    FClient: TAtlasMqttClient;
    FLog: TMemo;
    procedure ClientMessage(Sender: TObject; const Topic: string; const Payload: TBytes;
      QoS: TMqttQoS; Retained: Boolean);
  public
    constructor Create(AOwner: TComponent); override;
  end;

constructor TMainForm.Create(AOwner: TComponent);
begin
  inherited CreateNew(AOwner);
  FLog := TMemo.Create(Self);
  FLog.Parent := Self;
  FLog.Align := alClient;
  FClient := TAtlasMqttClient.Create(Self);
  FClient.Host := '192.168.1.20';
  FClient.ClientId := 'kitchen-panel';
  FClient.OnMessage := ClientMessage;
  FClient.Subscribe('home/kitchen/#', qosAtLeastOnce);  // remembered; sent once connected
  FClient.Connect;                                      // returns at once
end;

procedure TMainForm.ClientMessage(Sender: TObject; const Topic: string; const Payload: TBytes;
  QoS: TMqttQoS; Retained: Boolean);
begin
  // SynchronizeEvents is True by default, so this runs on the main thread.
  FLog.Lines.Add(Topic + ' = ' + TEncoding.UTF8.GetString(Payload));
end;

When the broker accepts the connection, the client sends the remembered subscription. After that, each message published on home/kitchen or below appears as one line, for example home/kitchen/temperature = 21.5. If the connection drops, the client reconnects and subscribes again by itself.

Publish at quality 1 and wait for the broker's acknowledgement. WaitForIdle waits until every message above quality 0 has been acknowledged, not only this one.

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

procedure PublishConfirmed(Client: TAtlasMqttClient; const Topic, Text: string);
var
  PacketId: Word;
begin
  PacketId := Client.Publish(Topic, Text, qosAtLeastOnce);
  if not Client.WaitForIdle(2000) then
    raise EAtlasMqttError.CreateFmt('Packet %d is not acknowledged yet; %d in flight',
      [PacketId, Client.InFlightCount]);
end;

The procedure returns once the broker has the message. If it raises, the message stays in flight and the client sends it again after the next reconnect. Nothing is lost unless the component is freed first.

Payloads are bytes; the string overload of Publish encodes text as UTF-8. This sends four 32-bit floats:

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

procedure PublishSamples(Client: TAtlasMqttClient; const Values: array of Single);
var
  Payload: TBytes;
begin
  SetLength(Payload, Length(Values) * SizeOf(Single));
  if Length(Values) > 0 then
    Move(Values[0], Payload[0], Length(Payload));
  Client.Publish('lab/adc/raw', Payload, qosAtLeastOnce);
end;

PublishSamples(Client, [1.5, 2.5, 3.5, 4.5]) publishes a 16-byte payload. A subscriber receives the same 16 bytes in the Payload argument of OnMessage.

Presence: a retained status that the device sets itself, and a retained will that the broker publishes if the device vanishes. Set the will before connecting.

uses
  Atlas.MqttCodec, Atlas.MqttClient;

procedure StartPresence(Client: TAtlasMqttClient);
begin
  Client.ClientId := 'pump-3';
  Client.WillTopic := 'status/pump-3';
  Client.WillMessage := 'offline';
  Client.WillQoS := qosAtLeastOnce;
  Client.WillRetain := True;
  if Client.ConnectAndWait(5000) then
    Client.Publish('status/pump-3', 'online', qosAtLeastOnce, True);
end;

Subscribers to status/# see online. A subscriber that joins later still gets it, with Retained True. If the process dies or its connection is cut, the broker publishes offline and keeps it retained. Disconnect says goodbye properly, so no will is sent then. To announce an orderly exit, publish offline yourself and call WaitForIdle before you disconnect.

An embedded broker that admits one set of credentials:

uses
  System.SysUtils, Atlas.MqttBroker;

type
  TBrokerHost = class
  private
    FBroker: TAtlasMqttBroker;
    procedure Authenticate(Sender: TObject; const ClientId, Username, Password: string;
      var Accept: Boolean);
  public
    constructor Create;
    destructor Destroy; override;
  end;

constructor TBrokerHost.Create;
begin
  inherited Create;
  FBroker := TAtlasMqttBroker.Create(nil);
  FBroker.Port := 1883;          // the default; 0 picks a free port, then read ActualPort
  FBroker.LoopbackOnly := True;  // the default: connections from this machine only
  FBroker.OnAuthenticate := Authenticate;
  FBroker.Start;
end;

destructor TBrokerHost.Destroy;
begin
  FBroker.Free;  // Stop closes every connection
  inherited;
end;

procedure TBrokerHost.Authenticate(Sender: TObject; const ClientId, Username, Password: string;
  var Accept: Boolean);
begin
  // Runs on the connection's broker thread. Accept is True on entry.
  Accept := (Username = 'station') and (Password = 'north-7');
end;

A client with other credentials receives return code 4, and a TAtlasMqttClient reports it through OnError as "Bad user name or password". Start raises EAtlasMqttTransportError if port 1883 is already taken.

Topic rules without a connection. TAtlasMqttCodec applies section 4.7 of the standard:

uses
  Atlas.MqttCodec;

procedure ShowTopicRules;
begin
  Writeln(TAtlasMqttCodec.TopicMatches('sport/tennis/+', 'sport/tennis/player1'));  // TRUE
  Writeln(TAtlasMqttCodec.TopicMatches('sport/#', 'sport'));                        // TRUE
  Writeln(TAtlasMqttCodec.TopicMatches('sport/+', 'sport'));                        // FALSE
  Writeln(TAtlasMqttCodec.TopicMatches('sport/+', 'sport/'));                       // TRUE
  Writeln(TAtlasMqttCodec.TopicMatches('#', '$SYS/uptime'));                        // FALSE
  Writeln(TAtlasMqttCodec.ValidTopicFilter('sport/tennis#'));                       // FALSE
  Writeln(TAtlasMqttCodec.ValidTopicName('sport/+'));                               // FALSE
end;

The output is TRUE TRUE FALSE TRUE FALSE FALSE FALSE, one value per line:

• # also matches its parent level, so sport/# matches sport.
• + matches exactly one level, and an empty level counts.
• Topics starting with $ are not caught by a wildcard in the first level.
• A wildcard must fill its whole level.
• Topics you publish to may not contain wildcards.

Connection status in a form caption. OnDisconnected also fires for every failed attempt while the client keeps retrying. OnError means the client has stopped.

uses
  System.SysUtils, System.Classes, Vcl.Forms, Atlas.MqttClient;

type
  TStatusForm = class(TForm)
  private
    FClient: TAtlasMqttClient;
    procedure ClientConnected(Sender: TObject; SessionPresent: Boolean);
    procedure ClientDisconnected(Sender: TObject; const Reason: string);
    procedure ClientError(Sender: TObject; const Reason: string);
  public
    constructor Create(AOwner: TComponent); override;
  end;

constructor TStatusForm.Create(AOwner: TComponent);
begin
  inherited CreateNew(AOwner);
  FClient := TAtlasMqttClient.Create(Self);
  FClient.Host := 'broker.lan';
  FClient.OnConnected := ClientConnected;
  FClient.OnDisconnected := ClientDisconnected;
  FClient.OnError := ClientError;
  FClient.Connect;
end;

procedure TStatusForm.ClientConnected(Sender: TObject; SessionPresent: Boolean);
begin
  Caption := 'Connected to ' + FClient.Host;
end;

procedure TStatusForm.ClientDisconnected(Sender: TObject; const Reason: string);
begin
  Caption := 'Offline: ' + Reason;
end;

procedure TStatusForm.ClientError(Sender: TObject; const Reason: string);
begin
  Caption := 'Refused: ' + Reason + ' (fix the settings, then Disconnect and Connect)';
end;

With no broker at broker.lan, the caption reads Offline: The host "broker.lan" could not be found, and it repeats as the client retries. With a wrong password, the caption ends on Refused: Bad user name or password ... and stays there.

Your own transport: this one counts the bytes the client sends and leaves the rest to TCP. The client asks for a new transport before every connection attempt, on its own thread, and frees it afterwards.

uses
  System.SysUtils, Atlas.MqttTransport, Atlas.MqttClient;

type
  TMeteredTransport = class(TAtlasTcpTransport)
  public
    class var BytesSent: Int64;
    procedure Send(const Data: TBytes); override;
  end;

  TTransportFactory = class
    procedure CreateTransport(Sender: TObject; var Transport: TAtlasMqttTransport);
  end;

procedure TMeteredTransport.Send(const Data: TBytes);
begin
  AtomicIncrement(BytesSent, Length(Data));
  inherited;
end;

procedure TTransportFactory.CreateTransport(Sender: TObject; var Transport: TAtlasMqttTransport);
begin
  Transport := TMeteredTransport.Create;
end;

Assign it with Client.OnCreateTransport := Factory.CreateTransport before Connect. After the first connection, TMeteredTransport.BytesSent holds at least the size of the CONNECT packet. If the handler leaves Transport nil, the client uses a plain TAtlasTcpTransport.

Tutorial

Sample project: LoopbackPublishSubscribe.

Goal: run a broker and a client in one console program, subscribe with a wildcard, publish at every quality of service and prove that each message arrived.

1. Create a console application named LoopbackPublishSubscribe. Add the Atlas MQTT src folder to the project's search path, or install the Atlas MQTT runtime package. Start the file with the program header and the uses clause. Atlas.MqttCodec is needed for the quality-of-service values, and Winapi.Windows for GetTickCount64.

   pascalCopy code

   program LoopbackPublishSubscribe;
   
   // Atlas MQTT quick start: a broker inside the program, one client that
   // subscribes with a wildcard and publishes at each quality of service, and a
   // check that every message came back on the expected topic. Nothing leaves the
   // machine: the broker listens on 127.0.0.1 on a port the system picks.
   
   {$APPTYPE CONSOLE}
   
   uses
     System.SysUtils, System.Classes, Winapi.Windows,
     Atlas.MqttCodec, Atlas.MqttClient, Atlas.MqttBroker;

2. Add a small class to receive the client's events: every message as a line of text, the quality the broker granted, and a count of acknowledgements. SynchronizeEvents stays True, so the events run on the main thread and the fields need no lock.

   pascalCopy code

   type
     // Receives the client's events. With SynchronizeEvents at its default (True)
     // they run on the main thread, so these fields need no lock. A console
     // program has no message loop, so the waits below call CheckSynchronize.
     TQuickStart = class
     private
       FReceived: TStringList;
       FGrantedQoS: Integer;
       FAcknowledged: Integer;
     public
       constructor Create;
       destructor Destroy; override;
       procedure MessageArrived(Sender: TObject; const Topic: string; const Payload: TBytes; QoS: TMqttQoS; Retained: Boolean);
       procedure Subscribed(Sender: TObject; const Filter: string; Granted: Boolean; QoS: TMqttQoS);
       procedure Published(Sender: TObject; PacketId: Word);
       // 'topic=payload q<quality>' per delivered message, in arrival order.
       property Received: TStringList read FReceived;
       property GrantedQoS: Integer read FGrantedQoS;
       property Acknowledged: Integer read FAcknowledged;
     end;
   
   constructor TQuickStart.Create;
   begin
     inherited Create;
     FReceived := TStringList.Create;
     FGrantedQoS := -1;
   end;
   
   destructor TQuickStart.Destroy;
   begin
     FReceived.Free;
     inherited;
   end;
   
   procedure TQuickStart.MessageArrived(Sender: TObject; const Topic: string; const Payload: TBytes; QoS: TMqttQoS; Retained: Boolean);
   var Line: string;
   begin
     Line := Format('%s=%s q%d', [Topic, TEncoding.UTF8.GetString(Payload), Ord(QoS)]);
     if Retained then Line := Line + ' retained';
     FReceived.Add(Line);
     Writeln('  received  ', Line);
   end;
   
   procedure TQuickStart.Subscribed(Sender: TObject; const Filter: string; Granted: Boolean; QoS: TMqttQoS);
   begin
     if Granted then FGrantedQoS := Ord(QoS) else FGrantedQoS := -2;
     Writeln(Format('  subscribed to %s: granted=%s quality=%d', [Filter, BoolToStr(Granted, True), Ord(QoS)]));
   end;
   
   procedure TQuickStart.Published(Sender: TObject; PacketId: Word);
   begin
     Inc(FAcknowledged);
     Writeln('  broker acknowledged packet ', PacketId);
   end;

3. Add two helpers. WaitUntil polls a condition with a timeout, and calls CheckSynchronize so that the queued events run while the main thread waits; this is what keeps the program from hanging. Check prints ok: or raises.

   pascalCopy code

   // Polls Condition until it holds or TimeoutMs pass. CheckSynchronize lets the
   // client's queued main-thread events run while this thread waits.
   function WaitUntil(const Condition: TFunc<Boolean>; TimeoutMs: Integer): Boolean;
   var Deadline: UInt64;
   begin
     Deadline := GetTickCount64 + UInt64(TimeoutMs);
     repeat
       if Condition() then Exit(True);
       CheckSynchronize(10);
     until GetTickCount64 >= Deadline;
     Result := Condition();
   end;
   
   procedure Check(Condition: Boolean; const What: string);
   begin
     if Condition then Writeln('ok: ', What)
     else raise Exception.Create('FAILED: ' + What);
   end;

4. Start the main block with the broker. Port := 0 lets the system choose a free port, and ActualPort tells you which one. LoopbackOnly is True by default, so only this machine can connect.

   pascalCopy code

   var
     Broker: TAtlasMqttBroker;
     Client: TAtlasMqttClient;
     Log: TQuickStart;
   
   begin
     ExitCode := 1;
     Log := TQuickStart.Create;
     Broker := TAtlasMqttBroker.Create(nil);
     Client := TAtlasMqttClient.Create(nil);
     try
       try
         // 1. A broker on any free port. LoopbackOnly is True by default, so only
         //    this machine can connect.
         Broker.Port := 0;
         Broker.Start;
         Writeln(Format('Broker listening on 127.0.0.1:%d', [Broker.ActualPort]));

5. Configure the client, attach the event handlers and connect. ConnectAndWait returns True once the broker has accepted the connection, or False after five seconds.

   pascalCopy code

         // 2. A client. Connect returns at once; ConnectAndWait also waits for the
         //    broker's answer.
         Client.Host := '127.0.0.1';
         Client.Port := Broker.ActualPort;
         Client.ClientId := 'quick-start';
         Client.OnMessage := Log.MessageArrived;
         Client.OnSubscribed := Log.Subscribed;
         Client.OnPublished := Log.Published;
         Check(Client.ConnectAndWait(5000), 'the client connects');
         Check(Broker.ClientCount = 1, 'the broker counts one client');

6. Subscribe with a single-level wildcard at quality 2, and wait for the broker to acknowledge it.

   pascalCopy code

         // 3. A wildcard subscription: '+' stands for exactly one topic level.
         Client.Subscribe('sensors/+/temperature', qosExactlyOnce);
         Check(WaitUntil(function: Boolean begin Result := Log.GrantedQoS = 2; end, 5000),
           'the broker grants the subscription at quality 2');

7. Publish one message at each quality, plus one that the filter does not match. The client subscribed to these topics itself, so the broker sends the matching messages back to it.

   pascalCopy code

         // 4. Publish at each quality. The broker routes them straight back to
         //    this client because its own subscription matches.
         Client.Publish('sensors/kitchen/temperature', '21.5');                 // quality 0: fire and forget
         Client.Publish('sensors/hall/temperature', '19.0', qosAtLeastOnce);    // quality 1: acknowledged
         Client.Publish('sensors/garage/temperature', '12.5', qosExactlyOnce);  // quality 2: four-way handshake
         Client.Publish('sensors/kitchen/humidity', '48', qosAtLeastOnce);      // does not match the filter

8. Wait, always with a timeout, for the deliveries and acknowledgements, then check every topic, payload and quality. WaitForIdle returns as soon as nothing is unacknowledged. The OnPublished events may still be queued at that point, so the program waits for them separately.

   pascalCopy code

         // 5. Wait, with a timeout, for the three matching messages and for the
         //    acknowledgements of the three messages above quality 0.
         Check(WaitUntil(function: Boolean begin Result := Log.Received.Count >= 3; end, 5000), 'three messages arrive');
         Check(Client.WaitForIdle(3000), 'nothing is left in flight');
         Check(WaitUntil(function: Boolean begin Result := Log.Acknowledged = 3; end, 3000),
           'OnPublished fired for each of the three acknowledged messages');
         CheckSynchronize(200); // time for a wrong delivery to show up, if there were one
   
         // 6. Verify topics, payloads and qualities.
         Check(Log.Received.Count = 3, 'the humidity message was filtered out by the broker');
         Check(Log.Received.IndexOf('sensors/kitchen/temperature=21.5 q0') >= 0, 'kitchen reading at quality 0');
         Check(Log.Received.IndexOf('sensors/hall/temperature=19.0 q1') >= 0, 'hall reading at quality 1');
         Check(Log.Received.IndexOf('sensors/garage/temperature=12.5 q2') >= 0, 'garage reading at quality 2');
         Check(Client.InFlightCount = 0, 'in-flight count is zero');
         Check(Broker.RoutedCount = 4, 'the broker accepted four messages');

9. Disconnect, stop the broker, and free the client before the broker. The exit code is 0 only when every check passed.

   pascalCopy code

         // 7. Say goodbye (so no will would be sent) and stop the broker.
         Client.Disconnect;
         Check(Client.State = msDisconnected, 'the client is disconnected');
         Broker.Stop;
         Check(not Broker.Active, 'the broker is stopped');
         Writeln('Done.');
         ExitCode := 0;
       except
         on E: Exception do Writeln(E.ClassName, ': ', E.Message);
       end;
     finally
       // The client before the broker: its goodbye needs somewhere to go.
       Client.Free;
       Broker.Free;
       Log.Free;
     end;
   end.

10. Build and run it. In the IDE, press F9. The port changes on every run, and the acknowledgement lines may come in another order:

    ExampleCopy code

    Broker listening on 127.0.0.1:62668
    ok: the client connects
    ok: the broker counts one client
      subscribed to sensors/+/temperature: granted=True quality=2
    ok: the broker grants the subscription at quality 2
      received  sensors/kitchen/temperature=21.5 q0
      received  sensors/hall/temperature=19.0 q1
      broker acknowledged packet 2
      received  sensors/garage/temperature=12.5 q2
    ok: three messages arrive
    ok: nothing is left in flight
      broker acknowledged packet 4
      broker acknowledged packet 3
    ok: OnPublished fired for each of the three acknowledged messages
    ok: the humidity message was filtered out by the broker
    ok: kitchen reading at quality 0
    ok: hall reading at quality 1
    ok: garage reading at quality 2
    ok: in-flight count is zero
    ok: the broker accepted four messages
    ok: the client is disconnected
    ok: the broker is stopped
    Done.

    To adapt it, change the filter to sensors/#. The humidity reading now matches too, so change the two message-count checks from 3 to 4. The ReconnectAndWill sample goes further, with wills, retained messages, a refused password, a cut connection and publishing while offline.

Troubleshooting

Events never fire in a console program, service or background thread. With SynchronizeEvents True, events are queued to the main thread and run only when it calls CheckSynchronize or runs a message loop. Call CheckSynchronize while you wait, as the tutorial's WaitUntil does, or set SynchronizeEvents := False. With it False, handlers run on the client's thread and must lock whatever they share.

The connection drops every time a certain message arrives. With SynchronizeEvents False, an exception that escapes an event handler is raised on the client's thread and ends the connection. OnDisconnected reports the exception's message as the reason, and the client reconnects. A message at quality 1 or 2 was not acknowledged yet at that point. The embedded broker does not send it again, but a broker that keeps sessions (CleanSession False) does, and the same thing happens again. A common cause is TEncoding.UTF8.GetString on a binary payload, which raises EEncodingError. Catch exceptions inside the handler. With SynchronizeEvents True, the exception is raised on the main thread instead, and the connection is not affected.

Publish raises "The client is not connected". Quality 0 needs an open connection, and the client has none: it has not connected yet, is waiting to reconnect, or was disconnected. Publish from OnConnected, after ConnectAndWait returns True, or check State = msConnected first. Alternatively, use quality 1 or 2, which the client holds until it can send.

Connect does nothing after a refusal, or after the connection was lost with AutoReconnect False. The client's thread has ended, but the component still holds it until Disconnect. Call Disconnect first, then Connect or ConnectAndWait.

ConnectAndWait returns False, but the client keeps trying. Running out of time does not stop the client. With AutoReconnect True it keeps retrying in the background, and OnDisconnected fires on each attempt with the reason, for example "Could not connect to 10.0.0.5:1883" or The host "broker.lan" could not be found. Call Disconnect to stop it. Host names are resolved to IPv4 only, so give an IPv4 address for a host that has only an IPv6 address.

Start raises "Binding port 1883 failed (Windows Sockets error 10048)". Another program, or another broker in the same process, already listens on the port. Choose another Port, or use Port := 0 and pass ActualPort to the clients.

Clients on other machines cannot reach the embedded broker. LoopbackOnly is True by default, so the broker accepts only connections from this machine. Set it to False before Start to listen on every IPv4 interface, and allow the port through the firewall. The broker has no TLS, so keep it on a network you trust.

A large message closes the connection to the embedded broker. The broker refuses any packet larger than 1 MiB (1 048 576 bytes, header included) and closes the connection. At quality 1 or 2, the client reconnects, sends the message again, and is cut off again. Split large payloads. The client itself accepts incoming packets of up to 16 MiB.

Publish raises "1000 messages are waiting for the broker; none more are accepted". MaxInFlight unacknowledged messages above quality 0 are already held, usually because the client has been offline for a while. Wait with WaitForIdle, raise MaxInFlight (up to 65 535), or drop the message.

A subscriber misses messages published while it was offline. The embedded broker keeps no sessions and queues nothing for clients that are not connected. Publish the latest state as a retained message (Retain = True), so that a client that subscribes again receives it at once.