Delphi WebSocket Client: robust verbinden, sauber stoppen, zuverlässig debuggen

unit Net.WebSocketClientEx;

interface

uses
  System.SysUtils,
  System.Classes,
  System.SyncObjs,
  System.Net.WebSockets;

type
  TWsLogLevel = (llDebug, llInfo, llWarn, llError);

  TWsLogEvent = reference to procedure(Level: TWsLogLevel; const Msg: string);
  TWsTextEvent = reference to procedure(const Text: string);
  TWsStateEvent = reference to procedure(const State: string);

  TDelphiWebSocketClient = class
  private
    FUrl: string;
    FOnLog: TWsLogEvent;
    FOnText: TWsTextEvent;
    FOnState: TWsStateEvent;

    FStopEvent: TEvent;
    FWorker: TThread;

    FMinBackoffMs: Integer;
    FMaxBackoffMs: Integer;
    FHeartbeatSec: Integer;
    FMaxMessageBytes: Integer;

    procedure Log(Level: TWsLogLevel; const Msg: string);
    procedure State(const S: string);

    procedure Run;
    function NextBackoffMs(const Prev: Integer): Integer;
    function NowUtcStr: string;
  public
    constructor Create(const AUrl: string);
    destructor Destroy; override;

    procedure Start;
    procedure Stop(TimeoutMs: Cardinal = 5000);

    property OnLog: TWsLogEvent read FOnLog write FOnLog;
    property OnText: TWsTextEvent read FOnText write FOnText;
    property OnState: TWsStateEvent read FOnState write FOnState;

    property MinBackoffMs: Integer read FMinBackoffMs write FMinBackoffMs;
    property MaxBackoffMs: Integer read FMaxBackoffMs write FMaxBackoffMs;
    property HeartbeatSec: Integer read FHeartbeatSec write FHeartbeatSec;
    property MaxMessageBytes: Integer read FMaxMessageBytes write FMaxMessageBytes;
  end;

implementation

uses
  System.NetEncoding;

constructor TDelphiWebSocketClient.Create(const AUrl: string);
begin
  inherited Create;
  FUrl := AUrl;
  FStopEvent := TEvent.Create(nil, True, False, '');

  FMinBackoffMs := 500;
  FMaxBackoffMs := 15000;
  FHeartbeatSec := 20;      // App-Heartbeat gegen Idle-Timeouts
  FMaxMessageBytes := 2*1024*1024; // Schutz gegen Payload-Explosion
end;

destructor TDelphiWebSocketClient.Destroy;
begin
  Stop;
  FStopEvent.Free;
  inherited;
end;

procedure TDelphiWebSocketClient.Start;
begin
  if Assigned(FWorker) then Exit;

  FStopEvent.ResetEvent;
  FWorker := TThread.CreateAnonymousThread(
    procedure
    begin
      Run;
    end);
  FWorker.FreeOnTerminate := False;
  FWorker.Start;
end;

procedure TDelphiWebSocketClient.Stop(TimeoutMs: Cardinal);
var
  W: TThread;
begin
  FStopEvent.SetEvent;

  W := FWorker;
  if Assigned(W) then
  begin
    if W.WaitFor(TimeoutMs) = wrTimeout then
      Log(llWarn, 'Stop: Worker reagiert nicht innerhalb Timeout (Connect/Receive könnte blockieren)');
    FreeAndNil(FWorker);
  end;
end;

procedure TDelphiWebSocketClient.Log(Level: TWsLogLevel; const Msg: string);
begin
  if Assigned(FOnLog) then
    TThread.Queue(nil,
      procedure
      begin
        FOnLog(Level, NowUtcStr + ' ' + Msg);
      end);
end;

procedure TDelphiWebSocketClient.State(const S: string);
begin
  if Assigned(FOnState) then
    TThread.Queue(nil,
      procedure
      begin
        FOnState(S);
      end);
end;

function TDelphiWebSocketClient.NowUtcStr: string;
begin
  Result := FormatDateTime('yyyy-mm-dd"T"hh:nn:ss.zzz"Z"', TTimeZone.Local.ToUniversalTime(Now));
end;

function TDelphiWebSocketClient.NextBackoffMs(const Prev: Integer): Integer;
var
  N: Integer;
begin
  if Prev <= 0 then Exit(FMinBackoffMs);
  N := Prev * 2;
  if N < FMinBackoffMs then N := FMinBackoffMs;
  if N > FMaxBackoffMs then N := FMaxBackoffMs;
  Result := N;
end;

procedure TDelphiWebSocketClient.Run;
var
  WS: TClientWebSocket;
  Backoff: Integer;
  LastHeartbeat: UInt64;
  Buf: TBytes;
  Received: TWebSocketReceiveResult;
  SB: TStringBuilder;
  AssembledBytes: Integer;

  function StopRequested: Boolean;
  begin
    Result := (FStopEvent.WaitFor(0) = wrSignaled);
  end;

  procedure SafeClose;
  begin
    try
      if WS.State = TWebSocketState.Open then
        WS.Close(TWebSocketCloseStatus.NormalClosure, 'client shutdown');
    except
      on E: Exception do
        Log(llDebug, 'Close: ' + E.ClassName + ': ' + E.Message);
    end;
  end;

  procedure DispatchText(const S: string);
  begin
    if Assigned(FOnText) then
      TThread.Queue(nil,
        procedure
        begin
          FOnText(S);
        end);
  end;

begin
  Backoff := 0;
  LastHeartbeat := 0;

  while not StopRequested do
  begin
    WS := TClientWebSocket.Create;
    try
      State('connecting');
      Log(llInfo, 'Connect: ' + FUrl);

      try
        // Connect ist synchron und kann bei DNS/TLS/Proxy blockieren.
        // Deshalb: immer im Worker, nie im Mainthread.
        WS.Connect(FUrl);
      except
        on E: Exception do
        begin
          State('connect_failed');
          Log(llWarn, 'Connect failed: ' + E.ClassName + ': ' + E.Message);
          Backoff := NextBackoffMs(Backoff);
          if FStopEvent.WaitFor(Backoff) = wrSignaled then Break;
          Continue;
        end;
      end;

      State('open');
      Backoff := 0;

      SetLength(Buf, 16 * 1024);
      SB := TStringBuilder.Create;
      AssembledBytes := 0;
      try
        while (WS.State = TWebSocketState.Open) and (not StopRequested) do
        begin
          // App-Heartbeat: bewusst als Text, weil Ping/Pong nicht überall gleich gut nutzbar ist.
          if (FHeartbeatSec > 0) then
          begin
            if (LastHeartbeat = 0) or
               (TThread.GetTickCount64 - LastHeartbeat >= UInt64(FHeartbeatSec) * 1000) then
            begin
              try
                WS.Send('{"type":"ping"}');
                LastHeartbeat := TThread.GetTickCount64;
                Log(llDebug, 'Heartbeat sent');
              except
                on E: Exception do
                begin
                  Log(llWarn, 'Heartbeat send failed: ' + E.Message);
                  Break;
                end;
              end;
            end;
          end;

          try
            Received := WS.Receive(Buf);
          except
            on E: Exception do
            begin
              Log(llWarn, 'Receive failed: ' + E.ClassName + ': ' + E.Message);
              Break;
            end;
          end;

          case Received.Kind of
            TWebSocketMessageKind.Text:
              begin
                SB.Append(TEncoding.UTF8.GetString(Buf, 0, Received.BytesReceived));
                Inc(AssembledBytes, Received.BytesReceived);

                // Schutz gegen „unendliche“ Nachrichten (z. B. Serverbug, Protokollbruch)
                if (FMaxMessageBytes > 0) and (AssembledBytes > FMaxMessageBytes) then
                begin
                  Log(llWarn, 'Message too large; closing connection');
                  Break;
                end;

                if Received.EndOfMessage then
                begin
                  DispatchText(SB.ToString);
                  SB.Clear;
                  AssembledBytes := 0;
                end;
              end;

            TWebSocketMessageKind.Binary:
              begin
                // In vielen Integrationen ist Text/JSON der Standard.
                // Wenn Binary gebraucht wird: separates Event + eigener Assembler + Limits.
                Log(llDebug, 'Binary frame: ' + Received.BytesReceived.ToString + ' bytes');
              end;

            TWebSocketMessageKind.Close:
              begin
                Log(llInfo, 'Server requested close');
                Break;
              end;
          end;

          TThread.Sleep(1);
        end;
      finally
        SB.Free;
      end;

      SafeClose;
      State('closed');

    finally
      WS.Free;
    end;

    if StopRequested then Break;

    Backoff := NextBackoffMs(Backoff);
    Log(llInfo, 'Reconnect in ' + Backoff.ToString + ' ms');
    if FStopEvent.WaitFor(Backoff) = wrSignaled then Break;
  end;

  State('stopped');
  Log(llInfo, 'Worker stopped');
end;

end.