YOUSEFUL / DELPHI / COMPONENTS

`TAtlasMqttParser`

Turns a TCP byte stream into MQTT packets, however the bytes are split, with a size limit checked on the header.

Class Unit Atlas.MqttCodec · Inherits TObject · Product Atlas MQTT

Description

Feed it whatever Receive returned, then call Next until it returns False. Packets may arrive one byte at a time or several per read. A peer announcing a packet larger than MaxPacketSize is refused as soon as the length bytes arrive, before any of the body is buffered. The client uses a 16 MiB limit and the embedded broker 1 MiB. Not thread-safe: use one parser per connection, from one thread.

Members

Member	Kind	What it does
`Create`	constructor	Creates an empty parser that refuses packets larger than `AMaxPacketSize` bytes, header included.
`Feed`	method	Appends the first `Count` bytes of `Data`.
`Next`	method	Takes the next complete packet off the buffer.
`Buffered`	property	Bytes received but not yet returned as a packet.
`MaxPacketSize`	property	The limit given to `Create`.

Constructors

`Create` constructor

constructor Create(AMaxPacketSize: Integer = 16 * 1024 * 1024);

Creates an empty parser that refuses packets larger than AMaxPacketSize bytes, header included. Raises EAtlasMqttError below 16.

Parser := TAtlasMqttParser.Create(1024 * 1024);  // refuse packets above 1 MiB

Properties

`Buffered` property

property Buffered: Integer read FCount;

Bytes received but not yet returned as a packet.

Count := Parser.Buffered;

`MaxPacketSize` property

property MaxPacketSize: Integer read FMaxPacketSize;

The limit given to Create.

Count := Parser.MaxPacketSize;

Methods

`Feed` method

procedure Feed(const Data: TBytes; Count: Integer);

Appends the first Count bytes of Data. Raises EAtlasMqttError when Count is negative or larger than Data.

Parser.Feed(Buffer, Count);

`Next` method

function Next(out Packet: TMqttPacket): Boolean;

Takes the next complete packet off the buffer. Returns False when no complete packet is buffered yet. Raises EAtlasMqttProtocolError for a malformed packet or one over the size limit; close the connection then, because the stream cannot be resynchronised.

while Parser.Next(Packet) do
  Text := Text + ' ' + IntToStr(Ord(Packet.Kind));

Examples

Collect the packets in a chunk of bytes

Returns the packets that are complete so far; an incomplete tail stays buffered for the next chunk. Feeding C0 00 D0 returns one PINGREQ; feeding 00 next returns the PINGRESP.

uses
  System.SysUtils, Atlas.MqttCodec;

function DrainPackets(Parser: TAtlasMqttParser; const Chunk: TBytes): TArray<TMqttPacket>;
var
  Packet: TMqttPacket;
begin
  Result := nil;
  Parser.Feed(Chunk, Length(Chunk));
  while Parser.Next(Packet) do
    Result := Result + [Packet];
end;

TAtlasMqttCodec · TMqttPacket · TAtlasTcpTransport

Also in Atlas MQTT

TAtlasMqttClient · TAtlasMqttBroker

TAtlasMqttParser