Loopback publish and subscribe

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.

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;

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;

// 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;

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]));

      // 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');

      // 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');

      // 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

      // 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');

      // 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.