{$mode objfpc}{$H+}

/// <summary>
/// test_MQTTIntegration — FPCUnit integration tests for uCICMQTTClient.
///
/// These tests require network access to test.mosquitto.org.  Each test
/// creates its own client instance(s) in SetUp and frees them in TearDown.
/// If the broker is unreachable the test is skipped rather than failed.
///
/// Public test endpoints on test.mosquitto.org (as documented at
/// https://test.mosquitto.org):
///   1883 — plain TCP, no auth
///   8886 — TLS (Let's Encrypt cert), no auth
///   1884 — plain TCP, requires username 'rw' / password 'readwrite'
/// </summary>
unit test_MQTTIntegration;

interface

uses
  Classes, SysUtils, DateUtils,
  fpcunit, testregistry,
  uCICMQTTClient;

const
  // Local Docker broker (start with: tests/docker/start_broker.sh)
  // Falls back gracefully (skip, not fail) when broker is unreachable.
  TEST_BROKER         = 'localhost';
  TEST_PORT_PLAIN     = 1883;
  TEST_PORT_TLS       = 8883;
  TEST_PORT_AUTH      = 1884;
  TEST_USERNAME       = 'rw';
  TEST_PASSWORD       = 'readwrite';
  TEST_TOPIC_PREFIX   = 'cictest/';

  /// Maximum milliseconds to wait for an asynchronous event (message, etc.)
  ASYNC_WAIT_MS = 5000;
  /// Polling interval when waiting for an async event
  POLL_INTERVAL_MS = 50;

type
  TMQTTIntegrationTests = class(TTestCase)
  private
    FClient:          TCICMQTTClient;
    FUniqueTopicBase: String; // unique per test run to avoid collisions

    // Fields used by the subscribe-and-receive test
    FReceivedTopic:   String;
    FReceivedPayload: String;
    FMessageReceived: Boolean;
    FReceivedCount:   LongInt;  // tally for the concurrent read/write stress test

    // Event handler wired to FClient.OnMessage in relevant tests
    procedure OnMessage(const ATopic, APayload: String;
      AQoS: TCICMQTTQoS);

    // Attempt to connect FClient; returns False and logs reason if
    // the broker cannot be reached (network error / timeout).
    function TryConnect: Boolean;

    // Poll until FMessageReceived becomes True or ASYNC_WAIT_MS elapses.
    procedure WaitForMessage;

    // Generate a unique topic suffix using a simple timestamp + counter.
    function UniqueTopicSuffix: String;

  protected
    procedure SetUp; override;
    procedure TearDown; override;

  published
    procedure TestConnect_PlainText;
    procedure TestConnect_TLS;
    procedure TestConnect_WithAuth;
    procedure TestConnect_BadCredentials;
    procedure TestDisconnect;
    procedure TestPublish_QoS0;
    procedure TestPublish_QoS1;
    procedure TestSubscribeAndReceive;
    procedure TestKeepAlive;
    procedure TestConcurrentPublishWhileReceiving;
    procedure TestPublishBurst_AllDelivered;
    procedure TestPublishBurst_QoS1;
    procedure TestConcurrentPublish_MultipleThreads;
    procedure TestTwoClients_Exchange;
    procedure TestReconnectAfterDisconnect_Publishes;
    procedure TestRapidConnectDisconnectCycles;
    procedure TestLargePayload_RoundTrips;
    procedure TestPublishWhenNotConnected_Raises;
  end;

implementation

// ---------------------------------------------------------------------------
// TMQTTBurstPublisherThread — publishes ACount QoS0 messages to ATopic on
// AClient from its own thread, used to stress-test Publish() called
// concurrently from multiple threads.
// ---------------------------------------------------------------------------

type
  TMQTTBurstPublisherThread = class(TThread)
  private
    FClient: TCICMQTTClient;
    FTopic:  String;
    FCount:  Integer;
    FTag:    String;
  public
    constructor Create(AClient: TCICMQTTClient; const ATopic, ATag: String;
      ACount: Integer);
    procedure Execute; override;
  end;

constructor TMQTTBurstPublisherThread.Create(AClient: TCICMQTTClient;
  const ATopic, ATag: String; ACount: Integer);
begin
  inherited Create(True); // create suspended
  FreeOnTerminate := False;
  FClient := AClient;
  FTopic  := ATopic;
  FTag    := ATag;
  FCount  := ACount;
end;

procedure TMQTTBurstPublisherThread.Execute;
var
  i: Integer;
begin
  for i := 1 to FCount do
  begin
    try
      FClient.Publish(FTopic, 'thread-' + FTag + '-msg-' + IntToStr(i), mqQoS0);
    except
      // Swallow — the main thread will detect a dropped connection via
      // IsConnected after Join; we don't want an unhandled exception to
      // crash a worker thread mid-test.
    end;
    Sleep(1);
  end;
end;

// ---------------------------------------------------------------------------
// Helper — generate a unique topic suffix for this test run
// ---------------------------------------------------------------------------

var
  GRunCounter: Integer = 0;

function TMQTTIntegrationTests.UniqueTopicSuffix: String;
begin
  Inc(GRunCounter);
  Result := FormatDateTime('yyyymmddhhnnsszzz', Now)
    + '_' + IntToStr(GRunCounter);
end;

// ---------------------------------------------------------------------------
// SetUp / TearDown
// ---------------------------------------------------------------------------

procedure TMQTTIntegrationTests.SetUp;
begin
  FClient          := TCICMQTTClient.Create;
  FMessageReceived := False;
  FReceivedTopic   := '';
  FReceivedPayload := '';
  FReceivedCount   := 0;
  FUniqueTopicBase := TEST_TOPIC_PREFIX + UniqueTopicSuffix + '/';
end;

procedure TMQTTIntegrationTests.TearDown;
begin
  if Assigned(FClient) then
  begin
    try
      if FClient.IsConnected then
        FClient.Disconnect(mqrcSuccess);
    except
      // Swallow disconnect errors during teardown
    end;
    FreeAndNil(FClient);
  end;
end;

// ---------------------------------------------------------------------------
// Event handler
// ---------------------------------------------------------------------------

procedure TMQTTIntegrationTests.OnMessage(const ATopic, APayload: String;
  AQoS: TCICMQTTQoS);
begin
  FReceivedTopic   := ATopic;
  FReceivedPayload := APayload;
  FMessageReceived := True;
  // Called on the reader thread (possibly many in quick succession during the
  // concurrent-stress test) — use an atomic increment for the tally.
  InterLockedIncrement(FReceivedCount);
end;

// ---------------------------------------------------------------------------
// TryConnect — connect FClient; return False if broker unreachable
// ---------------------------------------------------------------------------

function TMQTTIntegrationTests.TryConnect: Boolean;
begin
  Result := False;
  try
    FClient.Connect;
    Result := True;
  except
    on E: Exception do
    begin
      // Log but do not fail — caller decides whether to skip
      {$ifdef FPC}
      // FPC writeln goes to stdout; acceptable for a console test runner
      WriteLn('  [skip] Broker unreachable: ', E.Message);
      {$endif}
    end;
  end;
end;

// ---------------------------------------------------------------------------
// WaitForMessage
// ---------------------------------------------------------------------------

procedure TMQTTIntegrationTests.WaitForMessage;
var
  WaitStart: TDateTime;
begin
  WaitStart := Now;
  while (not FMessageReceived) and
    (MilliSecondsBetween(Now, WaitStart) < ASYNC_WAIT_MS) do
  begin
    Sleep(POLL_INTERVAL_MS);
  end;
end;

// ---------------------------------------------------------------------------
// Tests
// ---------------------------------------------------------------------------

// Connect to port 1883 (plain TCP, no authentication).
procedure TMQTTIntegrationTests.TestConnect_PlainText;
begin
  FClient.BrokerHostName := TEST_BROKER;
  FClient.BrokerPort     := TEST_PORT_PLAIN;
  FClient.UseSSL         := False;
  FClient.ClientID       := 'cictest_plain_' + UniqueTopicSuffix;
  FClient.CleanStart     := True;
  FClient.KeepAliveSecs  := 30;

  if not TryConnect then
  begin
    WriteLn('  [skip] TestConnect_PlainText: broker unreachable');
    Exit;
  end;

  AssertTrue('Should be connected after successful Connect', FClient.IsConnected);
  FClient.Disconnect(mqrcSuccess);
  AssertFalse('Should not be connected after Disconnect', FClient.IsConnected);
end;

// Connect to port 8886 (TLS, Let's Encrypt certificate, no authentication).
procedure TMQTTIntegrationTests.TestConnect_TLS;
begin
  FClient.BrokerHostName := TEST_BROKER;
  FClient.BrokerPort     := TEST_PORT_TLS;
  FClient.UseSSL         := True;
  FClient.ClientID       := 'cictest_tls_' + UniqueTopicSuffix;
  FClient.CleanStart     := True;
  FClient.KeepAliveSecs  := 30;

  if not TryConnect then
  begin
    WriteLn('  [skip] TestConnect_TLS: broker unreachable or TLS unavailable');
    Exit;
  end;

  AssertTrue('Should be connected via TLS', FClient.IsConnected);
end;

// Connect to port 1884 with the documented read-write credentials.
procedure TMQTTIntegrationTests.TestConnect_WithAuth;
begin
  FClient.BrokerHostName := TEST_BROKER;
  FClient.BrokerPort     := TEST_PORT_AUTH;
  FClient.UseSSL         := False;
  FClient.ClientID       := 'cictest_auth_' + UniqueTopicSuffix;
  FClient.Username       := TEST_USERNAME;
  FClient.Password       := TEST_PASSWORD;
  FClient.CleanStart     := True;
  FClient.KeepAliveSecs  := 30;

  if not TryConnect then
  begin
    WriteLn('  [skip] TestConnect_WithAuth: broker unreachable');
    Exit;
  end;

  AssertTrue('Should be connected with valid credentials', FClient.IsConnected);
end;

// Connecting with wrong credentials to the auth port must NOT result in
// an IsConnected state.
procedure TMQTTIntegrationTests.TestConnect_BadCredentials;
begin
  FClient.BrokerHostName := TEST_BROKER;
  FClient.BrokerPort     := TEST_PORT_AUTH;
  FClient.UseSSL         := False;
  FClient.ClientID       := 'cictest_badauth_' + UniqueTopicSuffix;
  FClient.Username       := TEST_USERNAME;
  FClient.Password       := 'WRONG_PASSWORD';
  FClient.CleanStart     := True;
  FClient.KeepAliveSecs  := 30;

  // Connect() may raise an exception or return with IsConnected=False;
  // either outcome is acceptable.
  try
    FClient.Connect;
  except
    // Exception on bad auth is acceptable
  end;

  AssertFalse('Must NOT be connected with wrong password', FClient.IsConnected);
end;

// After calling Disconnect the client must no longer report IsConnected.
procedure TMQTTIntegrationTests.TestDisconnect;
begin
  FClient.BrokerHostName := TEST_BROKER;
  FClient.BrokerPort     := TEST_PORT_PLAIN;
  FClient.UseSSL         := False;
  FClient.ClientID       := 'cictest_disc_' + UniqueTopicSuffix;
  FClient.CleanStart     := True;
  FClient.KeepAliveSecs  := 30;

  if not TryConnect then
  begin
    WriteLn('  [skip] TestDisconnect: broker unreachable');
    Exit;
  end;

  AssertTrue('Should be connected before Disconnect', FClient.IsConnected);
  FClient.Disconnect(mqrcSuccess);
  AssertFalse('Should not be connected after Disconnect', FClient.IsConnected);
end;

// Publish a QoS 0 message; expect no exception.
procedure TMQTTIntegrationTests.TestPublish_QoS0;
var
  Topic: String;
begin
  FClient.BrokerHostName := TEST_BROKER;
  FClient.BrokerPort     := TEST_PORT_PLAIN;
  FClient.UseSSL         := False;
  FClient.ClientID       := 'cictest_pub0_' + UniqueTopicSuffix;
  FClient.CleanStart     := True;
  FClient.KeepAliveSecs  := 30;

  if not TryConnect then
  begin
    WriteLn('  [skip] TestPublish_QoS0: broker unreachable');
    Exit;
  end;

  Topic := FUniqueTopicBase + 'qos0';
  FClient.Publish(Topic, 'hello QoS0', mqQoS0);

  // No exception means success for QoS 0 (fire-and-forget)
  AssertTrue('Still connected after QoS 0 publish', FClient.IsConnected);
end;

// Publish a QoS 1 message; the PUBACK must arrive without error.
procedure TMQTTIntegrationTests.TestPublish_QoS1;
var
  Topic: String;
begin
  FClient.BrokerHostName := TEST_BROKER;
  FClient.BrokerPort     := TEST_PORT_PLAIN;
  FClient.UseSSL         := False;
  FClient.ClientID       := 'cictest_pub1_' + UniqueTopicSuffix;
  FClient.CleanStart     := True;
  FClient.KeepAliveSecs  := 30;

  if not TryConnect then
  begin
    WriteLn('  [skip] TestPublish_QoS1: broker unreachable');
    Exit;
  end;

  Topic := FUniqueTopicBase + 'qos1';

  try
    FClient.Publish(Topic, 'hello QoS1', mqQoS1);
  except
    on E: Exception do
      Fail('Unexpected exception during QoS 1 publish: ' + E.Message);
  end;

  // Give the reader thread a moment to receive the PUBACK
  Sleep(500);

  AssertTrue('Still connected after QoS 1 publish', FClient.IsConnected);
end;

// Subscribe on one client, publish from a second client, verify the message
// arrives on the subscriber within ASYNC_WAIT_MS.
procedure TMQTTIntegrationTests.TestSubscribeAndReceive;
var
  Publisher: TCICMQTTClient;
  Topic:     String;
  Suffix:    String;
begin
  Publisher := nil;
  Suffix    := UniqueTopicSuffix;
  Topic     := TEST_TOPIC_PREFIX + Suffix + '/msg';

  // --- Set up subscriber (FClient) ---
  FClient.BrokerHostName := TEST_BROKER;
  FClient.BrokerPort     := TEST_PORT_PLAIN;
  FClient.UseSSL         := False;
  FClient.ClientID       := 'cictest_sub_' + Suffix;
  FClient.CleanStart     := True;
  FClient.KeepAliveSecs  := 30;
  FClient.OnMessage      := @OnMessage;

  if not TryConnect then
  begin
    WriteLn('  [skip] TestSubscribeAndReceive: broker unreachable (subscriber)');
    Exit;
  end;

  FClient.Subscribe(Topic, mqQoS0);

  // Brief pause to allow the SUBACK to arrive before publishing
  Sleep(300);

  // --- Set up publisher (second client) ---
  Publisher := TCICMQTTClient.Create;
  try
    Publisher.BrokerHostName := TEST_BROKER;
    Publisher.BrokerPort     := TEST_PORT_PLAIN;
    Publisher.UseSSL         := False;
    Publisher.ClientID       := 'cictest_pub_' + Suffix;
    Publisher.CleanStart     := True;
    Publisher.KeepAliveSecs  := 30;

    try
      Publisher.Connect;
    except
      on E: Exception do
      begin
        WriteLn('  [skip] TestSubscribeAndReceive: broker unreachable (publisher): ',
          E.Message);
        Exit;
      end;
    end;

    Publisher.Publish(Topic, 'integration-test-payload', mqQoS0);

    // --- Wait for message to arrive on subscriber ---
    WaitForMessage;

    AssertTrue('Message should have been received within ' +
      IntToStr(ASYNC_WAIT_MS) + ' ms', FMessageReceived);
    AssertEquals('Received topic should match published topic',
      Topic, FReceivedTopic);
    AssertEquals('Received payload should match published payload',
      'integration-test-payload', FReceivedPayload);

    Publisher.Disconnect(mqrcSuccess);
  finally
    FreeAndNil(Publisher);
  end;
end;

// Connect with a short keepalive, wait long enough for two keepalive intervals
// to elapse, and assert the client is still connected (PINGREQ/PINGRESP cycle
// is working).
procedure TMQTTIntegrationTests.TestKeepAlive;
const
  KEEPALIVE_SECS  = 5;
  WAIT_SECS       = 12;
begin
  FClient.BrokerHostName := TEST_BROKER;
  FClient.BrokerPort     := TEST_PORT_PLAIN;
  FClient.UseSSL         := False;
  FClient.ClientID       := 'cictest_ka_' + UniqueTopicSuffix;
  FClient.CleanStart     := True;
  FClient.KeepAliveSecs  := KEEPALIVE_SECS;

  if not TryConnect then
  begin
    WriteLn('  [skip] TestKeepAlive: broker unreachable');
    Exit;
  end;

  AssertTrue('Connected before keepalive wait', FClient.IsConnected);

  // Sleep long enough for at least two full keepalive intervals
  Sleep(WAIT_SECS * 1000);

  AssertTrue(
    Format('Should still be connected after %ds (keepalive=%ds)',
      [WAIT_SECS, KEEPALIVE_SECS]),
    FClient.IsConnected);
end;

// Thread-safety regression test for the concurrent SSL_read/SSL_write race.
//
// The client subscribes to a topic and publishes to that same topic, so the
// broker echoes every message straight back: the reader thread is doing
// SSL_read (incoming PUBLISH) while the main thread is doing SSL_write
// (Publish). Run over TLS, this is the exact condition that used to corrupt the
// OpenSSL/heap state and wedge the client before reads and writes were
// serialized under FLock.
//
// Deliberately QoS 0 with a 1 ms yield per publish: QoS 0 means the reader only
// ever reads (no PUBACK to send back), and the yield lets the reader keep
// draining the echo stream. Without that pacing a delay-free flood lets the
// main thread monopolize FLock, starve the reader, and saturate the socket
// buffers into a TCP flow-control stall — a pathological load, not the race
// under test. The pacing still leaves hundreds of read/write overlaps (echoes
// arrive asynchronously across the whole burst); with the fix the client must
// stay connected throughout and deliver the echoes back.
procedure TMQTTIntegrationTests.TestConcurrentPublishWhileReceiving;
const
  PUBLISH_COUNT = 300;
var
  Topic: String;
  i:     Integer;
begin
  FClient.BrokerHostName := TEST_BROKER;
  FClient.BrokerPort     := TEST_PORT_TLS;   // TLS — the race is OpenSSL-specific
  FClient.UseSSL         := True;
  FClient.ClientID       := 'cictest_concurrent_' + UniqueTopicSuffix;
  FClient.CleanStart     := True;
  FClient.KeepAliveSecs  := 30;
  FClient.OnMessage      := @OnMessage;

  if not TryConnect then
  begin
    WriteLn('  [skip] TestConcurrentPublishWhileReceiving: broker/TLS unreachable');
    Exit;
  end;

  Topic := FUniqueTopicBase + 'concurrent';
  FClient.Subscribe(Topic, mqQoS0);
  Sleep(300); // let the SUBACK land before publishing

  for i := 1 to PUBLISH_COUNT do
  begin
    try
      FClient.Publish(Topic, 'msg-' + IntToStr(i), mqQoS0);
    except
      on E: Exception do
        Fail(Format('Publish %d/%d raised (client wedged?): %s',
          [i, PUBLISH_COUNT, E.Message]));
    end;
    if not FClient.IsConnected then
      Fail(Format('Connection dropped mid-burst after %d/%d publishes',
        [i, PUBLISH_COUNT]));
    Sleep(1); // yield so the reader thread drains echoes (see header note)
  end;

  // Drain the trailing echoes.
  Sleep(2000);

  AssertTrue('Client must remain connected through the concurrent read/write burst',
    FClient.IsConnected);
  AssertTrue(Format('Should have received a substantial share of the %d echoed ' +
    'messages back (got %d) — proves reads flowed alongside writes',
    [PUBLISH_COUNT, FReceivedCount]),
    FReceivedCount >= PUBLISH_COUNT div 2);
end;

// Subscribe to a unique topic, publish N QoS0 messages to it with a 1 ms
// yield between each, then assert the client stayed connected and received
// a substantial share of the echoes back. Hardens the single-threaded
// publish-burst path (no other thread publishing concurrently).
procedure TMQTTIntegrationTests.TestPublishBurst_AllDelivered;
const
  PUBLISH_COUNT = 60;
var
  Topic: String;
  i:     Integer;
begin
  FClient.BrokerHostName := TEST_BROKER;
  FClient.BrokerPort     := TEST_PORT_PLAIN;
  FClient.UseSSL         := False;
  FClient.ClientID       := 'cictest_burst0_' + UniqueTopicSuffix;
  FClient.CleanStart     := True;
  FClient.KeepAliveSecs  := 30;
  FClient.OnMessage      := @OnMessage;

  if not TryConnect then
  begin
    WriteLn('  [skip] TestPublishBurst_AllDelivered: broker unreachable');
    Exit;
  end;

  Topic := FUniqueTopicBase + 'burst0';
  FClient.Subscribe(Topic, mqQoS0);
  Sleep(300); // let the SUBACK land before publishing

  for i := 1 to PUBLISH_COUNT do
  begin
    FClient.Publish(Topic, 'burst-' + IntToStr(i), mqQoS0);
    Sleep(1);
  end;

  Sleep(1500); // settle: drain trailing echoes

  AssertTrue('Client must remain connected through the QoS0 burst',
    FClient.IsConnected);
  AssertTrue(Format('Should have received a substantial share of the %d ' +
    'echoed messages (got %d)', [PUBLISH_COUNT, FReceivedCount]),
    FReceivedCount >= PUBLISH_COUNT div 2);
end;

// Same as above but QoS1, exercising the PUBACK round-trip under a burst.
procedure TMQTTIntegrationTests.TestPublishBurst_QoS1;
const
  PUBLISH_COUNT = 60;
var
  Topic: String;
  i:     Integer;
begin
  FClient.BrokerHostName := TEST_BROKER;
  FClient.BrokerPort     := TEST_PORT_PLAIN;
  FClient.UseSSL         := False;
  FClient.ClientID       := 'cictest_burst1_' + UniqueTopicSuffix;
  FClient.CleanStart     := True;
  FClient.KeepAliveSecs  := 30;
  FClient.OnMessage      := @OnMessage;

  if not TryConnect then
  begin
    WriteLn('  [skip] TestPublishBurst_QoS1: broker unreachable');
    Exit;
  end;

  Topic := FUniqueTopicBase + 'burst1';
  FClient.Subscribe(Topic, mqQoS0);
  Sleep(300);

  for i := 1 to PUBLISH_COUNT do
  begin
    FClient.Publish(Topic, 'burst1-' + IntToStr(i), mqQoS1);
    Sleep(1);
  end;

  Sleep(1500);

  AssertTrue('Client must remain connected through the QoS1 burst',
    FClient.IsConnected);
  AssertTrue(Format('Should have received a substantial share of the %d ' +
    'echoed messages (got %d)', [PUBLISH_COUNT, FReceivedCount]),
    FReceivedCount >= PUBLISH_COUNT div 2);
end;

// Spawn 3 threads that each publish K QoS0 messages to the same topic on the
// SAME client concurrently, exercising Publish() called from multiple
// threads at once (not just main-thread-writes/reader-thread-reads as in
// TestConcurrentPublishWhileReceiving).
procedure TMQTTIntegrationTests.TestConcurrentPublish_MultipleThreads;
const
  THREAD_COUNT     = 3;
  PER_THREAD_COUNT = 20;
  TOTAL_COUNT      = THREAD_COUNT * PER_THREAD_COUNT;
var
  Topic:   String;
  Threads: array[0..THREAD_COUNT - 1] of TMQTTBurstPublisherThread;
  i:       Integer;
begin
  FClient.BrokerHostName := TEST_BROKER;
  FClient.BrokerPort     := TEST_PORT_PLAIN;
  FClient.UseSSL         := False;
  FClient.ClientID       := 'cictest_mthread_' + UniqueTopicSuffix;
  FClient.CleanStart     := True;
  FClient.KeepAliveSecs  := 30;
  FClient.OnMessage      := @OnMessage;

  if not TryConnect then
  begin
    WriteLn('  [skip] TestConcurrentPublish_MultipleThreads: broker unreachable');
    Exit;
  end;

  Topic := FUniqueTopicBase + 'mthread';
  FClient.Subscribe(Topic, mqQoS0);
  Sleep(300);

  for i := 0 to THREAD_COUNT - 1 do
    Threads[i] := TMQTTBurstPublisherThread.Create(FClient, Topic,
      IntToStr(i), PER_THREAD_COUNT);

  try
    for i := 0 to THREAD_COUNT - 1 do
      Threads[i].Start;

    for i := 0 to THREAD_COUNT - 1 do
      Threads[i].WaitFor;
  finally
    for i := 0 to THREAD_COUNT - 1 do
      FreeAndNil(Threads[i]);
  end;

  Sleep(2000); // settle: drain trailing echoes

  AssertTrue('Client must remain connected after concurrent multi-thread publish',
    FClient.IsConnected);
  AssertTrue(Format('Should have received a substantial share of the %d ' +
    'messages published concurrently from %d threads (got %d)',
    [TOTAL_COUNT, THREAD_COUNT, FReceivedCount]),
    FReceivedCount >= 30);
end;

// Like TestSubscribeAndReceive, but the publisher sends M messages; the
// subscriber must receive all (or nearly all) of them, in order-agnostic
// fashion (the last-received payload/topic fields are checked, and the
// tally must reach M).
procedure TMQTTIntegrationTests.TestTwoClients_Exchange;
const
  MESSAGE_COUNT = 25;
var
  Publisher: TCICMQTTClient;
  Topic:     String;
  Suffix:    String;
  i:         Integer;
begin
  Publisher := nil;
  Suffix    := UniqueTopicSuffix;
  Topic     := TEST_TOPIC_PREFIX + Suffix + '/exchange';

  FClient.BrokerHostName := TEST_BROKER;
  FClient.BrokerPort     := TEST_PORT_PLAIN;
  FClient.UseSSL         := False;
  FClient.ClientID       := 'cictest_exchsub_' + Suffix;
  FClient.CleanStart     := True;
  FClient.KeepAliveSecs  := 30;
  FClient.OnMessage      := @OnMessage;

  if not TryConnect then
  begin
    WriteLn('  [skip] TestTwoClients_Exchange: broker unreachable (subscriber)');
    Exit;
  end;

  FClient.Subscribe(Topic, mqQoS0);
  Sleep(300);

  Publisher := TCICMQTTClient.Create;
  try
    Publisher.BrokerHostName := TEST_BROKER;
    Publisher.BrokerPort     := TEST_PORT_PLAIN;
    Publisher.UseSSL         := False;
    Publisher.ClientID       := 'cictest_exchpub_' + Suffix;
    Publisher.CleanStart     := True;
    Publisher.KeepAliveSecs  := 30;

    try
      Publisher.Connect;
    except
      on E: Exception do
      begin
        WriteLn('  [skip] TestTwoClients_Exchange: broker unreachable (publisher): ',
          E.Message);
        Exit;
      end;
    end;

    for i := 1 to MESSAGE_COUNT do
      Publisher.Publish(Topic, 'exchange-msg-' + IntToStr(i), mqQoS0);

    // Wait for the tally to reach MESSAGE_COUNT (or time out).
    WaitForMessage;
    Sleep(1500); // settle: allow trailing messages to arrive

    AssertTrue('Subscriber should have received all/most of the published messages',
      FReceivedCount >= (MESSAGE_COUNT * 8) div 10); // >= 80%
    AssertEquals('Last received topic should match the exchange topic',
      Topic, FReceivedTopic);

    Publisher.Disconnect(mqrcSuccess);
  finally
    FreeAndNil(Publisher);
  end;
end;

// Connect, cleanly Disconnect, assert not connected; reconnect, assert
// connected, publish one message, assert still connected. Exercises a
// clean disconnect -> reconnect -> publish cycle on the same client instance.
procedure TMQTTIntegrationTests.TestReconnectAfterDisconnect_Publishes;
var
  Topic: String;
begin
  FClient.BrokerHostName := TEST_BROKER;
  FClient.BrokerPort     := TEST_PORT_PLAIN;
  FClient.UseSSL         := False;
  FClient.ClientID       := 'cictest_reconnect_' + UniqueTopicSuffix;
  FClient.CleanStart     := True;
  FClient.KeepAliveSecs  := 30;

  if not TryConnect then
  begin
    WriteLn('  [skip] TestReconnectAfterDisconnect_Publishes: broker unreachable');
    Exit;
  end;

  AssertTrue('Should be connected after first Connect', FClient.IsConnected);

  FClient.Disconnect(mqrcSuccess);
  AssertFalse('Should not be connected after Disconnect', FClient.IsConnected);

  FClient.Connect;
  AssertTrue('Should be connected again after reconnect', FClient.IsConnected);

  Topic := FUniqueTopicBase + 'reconnect';
  FClient.Publish(Topic, 'post-reconnect-payload', mqQoS0);

  AssertTrue('Should still be connected after publishing post-reconnect',
    FClient.IsConnected);
end;

// Rapidly cycle Connect/Disconnect 5 times on the same client instance.
// Must not crash, and must end in a disconnected state.
procedure TMQTTIntegrationTests.TestRapidConnectDisconnectCycles;
const
  CYCLE_COUNT = 5;
var
  i: Integer;
begin
  FClient.BrokerHostName := TEST_BROKER;
  FClient.BrokerPort     := TEST_PORT_PLAIN;
  FClient.UseSSL         := False;
  FClient.ClientID       := 'cictest_cycle_' + UniqueTopicSuffix;
  FClient.CleanStart     := True;
  FClient.KeepAliveSecs  := 30;

  if not TryConnect then
  begin
    WriteLn('  [skip] TestRapidConnectDisconnectCycles: broker unreachable');
    Exit;
  end;

  // TryConnect already performed cycle 1's Connect; disconnect it now so the
  // loop below performs CYCLE_COUNT full connect/disconnect round-trips.
  AssertTrue('Connected after initial Connect', FClient.IsConnected);
  FClient.Disconnect(mqrcSuccess);
  AssertFalse('Disconnected after initial Disconnect', FClient.IsConnected);

  for i := 1 to CYCLE_COUNT do
  begin
    FClient.Connect;
    AssertTrue(Format('Connected on cycle %d/%d', [i, CYCLE_COUNT]),
      FClient.IsConnected);

    FClient.Disconnect(mqrcSuccess);
    AssertFalse(Format('Disconnected on cycle %d/%d', [i, CYCLE_COUNT]),
      FClient.IsConnected);
  end;

  AssertFalse('Must end disconnected after rapid connect/disconnect cycles',
    FClient.IsConnected);
end;

// Publish a ~4 KB payload and assert it round-trips intact.
procedure TMQTTIntegrationTests.TestLargePayload_RoundTrips;
const
  CHUNK = 'The quick brown fox jumps over the lazy dog. '; // 46 bytes
var
  Topic:   String;
  Payload: String;
  i:       Integer;
begin
  FClient.BrokerHostName := TEST_BROKER;
  FClient.BrokerPort     := TEST_PORT_PLAIN;
  FClient.UseSSL         := False;
  FClient.ClientID       := 'cictest_large_' + UniqueTopicSuffix;
  FClient.CleanStart     := True;
  FClient.KeepAliveSecs  := 30;
  FClient.OnMessage      := @OnMessage;

  if not TryConnect then
  begin
    WriteLn('  [skip] TestLargePayload_RoundTrips: broker unreachable');
    Exit;
  end;

  Payload := '';
  for i := 1 to 90 do // ~4140 bytes
    Payload := Payload + CHUNK;

  Topic := FUniqueTopicBase + 'large';
  FClient.Subscribe(Topic, mqQoS0);
  Sleep(300);

  FClient.Publish(Topic, Payload, mqQoS0);

  WaitForMessage;

  AssertTrue('Large payload message should have been received within ' +
    IntToStr(ASYNC_WAIT_MS) + ' ms', FMessageReceived);
  AssertEquals('Received topic should match published topic',
    Topic, FReceivedTopic);
  AssertEquals('Received payload length should match sent payload length',
    Length(Payload), Length(FReceivedPayload));
  AssertEquals('Received payload should match sent payload exactly',
    Payload, FReceivedPayload);
end;

// Calling Publish on a client that has never connected must raise, per the
// documented 'MQTT client is not connected' guard.
procedure TMQTTIntegrationTests.TestPublishWhenNotConnected_Raises;
var
  RaisedException: Boolean;
begin
  RaisedException := False;

  AssertFalse('Freshly-created client must not report connected',
    FClient.IsConnected);

  try
    FClient.Publish(FUniqueTopicBase + 'notconnected', 'should not send', mqQoS0);
  except
    on E: Exception do
      RaisedException := True;
  end;

  AssertTrue('Publish on a not-connected client must raise an exception',
    RaisedException);
end;

// ---------------------------------------------------------------------------
// Registration
// ---------------------------------------------------------------------------

initialization
  RegisterTest(TMQTTIntegrationTests);

end.