TAtlasMqttBroker

Create constructor

constructor Create(AOwner: TComponent); override;

Creates a stopped broker with Port 1883 and LoopbackOnly True.

Broker := TAtlasMqttBroker.Create(nil);

Destroy destructor

destructor Destroy; override;

Calls Stop, then frees the broker.

Broker.Free;

ActualPort property

property ActualPort: Word read FActualPort;

The port really bound by Start: the same as Port, or the one the system chose when Port is 0. 0 while stopped.

Client.Port := Broker.ActualPort;

ClientCount property

property ClientCount: Integer read GetClientCount;

Clients whose CONNECT was accepted and whose connection is still open. A socket that has not sent CONNECT yet does not count.

Count := Broker.ClientCount;

RetainedCount property

property RetainedCount: Integer read GetRetainedCount;

Topics with a retained message. A retained publication with an empty payload removes its topic.

Count := Broker.RetainedCount;

RoutedCount property

property RoutedCount: Int64 read FRouted;

Messages accepted from publishers since Start, whether or not a subscription matched. Wills count too. A repeated quality-2 message counts once.

Text := IntToStr(Broker.RoutedCount);

Port property

property Port: Word read FPort write FPort default 1883;

The TCP port to listen on; 0 lets the system choose a free one (read ActualPort after Start). A change takes effect at the next Start.

Broker.Port := 0;

LoopbackOnly property

property LoopbackOnly: Boolean read FLoopbackOnly write FLoopbackOnly default True;

True listens on 127.0.0.1 only, so only programs on this machine can connect. False listens on every IPv4 interface. The broker has no TLS and no per-topic access control, so do not expose it to the internet. A change takes effect at the next Start.

Broker.LoopbackOnly := False;

Active property

property Active: Boolean read GetActive write SetActive default False;

True while started. Setting it calls Start or Stop. In the form designer, setting it has no effect, so start the broker at run time. When a form file does contain Active = True, the broker starts when loading finishes, after Port and LoopbackOnly have been read.

Broker.Active := True;

OnAuthenticate event

property OnAuthenticate: TAtlasMqttAuthenticateEvent read FOnAuthenticate write FOnAuthenticate;

Called for every CONNECT that passes the protocol checks, on the thread of that connection. It is not synchronised with the main thread, so the handler must be thread-safe. Accept is True on entry; set it to False to refuse the client with return code 4 ("Bad user name or password"). Password is the UTF-8 text of the password field, empty when none was sent. Without a handler, every client is accepted.

Broker.OnAuthenticate := Handler.Authenticate;

Start method

procedure Start;

Opens the listening socket on Port and starts accepting connections. ActualPort tells which port was bound, and RoutedCount starts from 0 again. Raises EAtlasMqttTransportError when the port cannot be bound, for example "Binding port 1883 failed (Windows Sockets error 10048)" when it is taken. Does nothing when already started.

Broker.Port := 0;
Broker.Start;

Stop method

procedure Stop;

Stops accepting, closes every client connection and waits for the connection threads to end. No wills are published while stopping. All retained messages are discarded, and ActualPort becomes 0. Clients see the connection close; with AutoReconnect they keep trying and find the broker again after the next Start on the same port. Does nothing when already stopped.

Broker.Stop;

DropClient method

function DropClient(const ClientId: string): Boolean;

Cuts the connection of the client with this identifier without ceremony, as a network failure would. Its will, if it has one, is published. Returns False when no connected client has that identifier. Meant for tests and demonstrations of reconnect and will handling.

Ok := Broker.DropClient('pump-3');

A broker and a client for a test

A broker on a free loopback port with a client connected to it, and a network failure on demand. Free the client before the broker.

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

function StartTestBroker(out Client: TAtlasMqttClient): TAtlasMqttBroker;
begin
  Result := TAtlasMqttBroker.Create(nil);
  Result.Port := 0;  // any free port
  Result.Start;
  Client := TAtlasMqttClient.Create(nil);
  Client.Host := '127.0.0.1';
  Client.Port := Result.ActualPort;
  Client.ClientId := 'under-test';
  Client.SynchronizeEvents := False;
  Client.ReconnectDelayMs := 100;
  if not Client.ConnectAndWait(3000) then
  begin
    FreeAndNil(Client);
    Result.Free;
    raise Exception.Create('The test client did not connect');
  end;
end;

// Simulates a network failure: the client's will is published and it reconnects.
procedure CutConnection(Broker: TAtlasMqttBroker);
begin
  if not Broker.DropClient('under-test') then
    raise Exception.Create('under-test is not connected');
end;

Report what a running broker is doing

The counters are safe to read while the broker runs.

uses
  System.SysUtils, Atlas.MqttBroker;

function BrokerSummary(Broker: TAtlasMqttBroker): string;
begin
  if not Broker.Active then
    Exit('stopped');
  Result := Format('port %d, %d clients, %d retained topics, %d messages routed',
    [Broker.ActualPort, Broker.ClientCount, Broker.RetainedCount, Broker.RoutedCount]);
end;