ewsdr/CATTcp.pas

unit CATTcp;
{
  CATTcp.pas — TCP CAT сервер (аналог TCPIPcatServer.cs из Thetis).

  Архитектура:
   • Один listening socket (TCATTcpServer)
   • На каждое входящее соединение — отдельный поток (TCATTcpClientThread)
   • Команды вида "ZZFA00014250000;" обрабатываются через TCATEngine.Parse()
   • Поддержка ZZGA/ZZGR для идентификации клиентов (directed broadcast)
   • Welcome-строка при подключении

  Порт по умолчанию: 4992

  Платформы: Linux / Windows (через WebUtils.SockClose/SockShutdown).
}
{$IFDEF FPC}
  {$MODE Delphi}
{$ENDIF}

interface

uses
  Classes, SysUtils,
  WebUtils,   // SockClose, SockShutdown, SockSend, SOCK_INVALID
  {$IFDEF MSWINDOWS}
    Windows, WinSock2,
  {$ELSE}
    BaseUnix, Sockets,
  {$ENDIF}
  SyncObjs,
  CATEngine;

const
  CAT_TCP_DEFAULT_PORT = 4992;
  CAT_TCP_BUF_SIZE     = 4096;
  CAT_TCP_WELCOME      = '#EWSDR CAT TCP Server;';

type
  TCATTcpServer = class;

  { TCATTcpClientThread — один поток на TCP клиента }
  TCATTcpClientThread = class(TThread)
  private
    FServer:     TCATTcpServer;
    FSocket:     TSocket;
    FBuf:        string;
    FClientIds:  TStringList;
    FIdLock:     TCriticalSection;
    FMarkDelete: Boolean;

    procedure ProcessBuffer;
    procedure SendStr(const S: string);
    procedure HandleCmd(const Cmd: string);
    procedure AddClientId(const Id: string);
    procedure RemoveClientId(const Id: string);
  protected
    procedure Execute; override;
  public
    constructor Create(AServer: TCATTcpServer; ASocket: TSocket);
    destructor  Destroy; override;
    procedure   SendData(const Msg: string; const IdLimit: TStringList = nil);
    function    IsMarkedForDelete: Boolean;
    procedure   Disconnect;
  end;

  { TCATTcpServer }
  TCATTcpServer = class
  private
    FEngine:       TCATEngine;
    FPort:         Integer;
    FListenSock:   TSocket;
    FServerThread: TThread;
    FPurgeThread:  TThread;
    FClients:      TList;
    FClientsLock:  TCriticalSection;
    FRunning:      Boolean;
    FSendWelcome:  Boolean;
    FLastError:    string;
    FStopEvent:    TEvent;

    procedure ServerLoop;
    procedure PurgeLoop;
    function  InitListen: Boolean;
  public
    constructor Create(AEngine: TCATEngine; APort: Integer = CAT_TCP_DEFAULT_PORT);
    destructor  Destroy; override;
    procedure   Start;
    procedure   Stop;
    procedure   Broadcast(const Msg: string; const IdLimit: TStringList = nil);
    function    ClientCount: Integer;
    property    Port:        Integer read FPort       write FPort;
    property    SendWelcome: Boolean read FSendWelcome write FSendWelcome;
    property    Running:     Boolean read FRunning;
    property    LastError:   string  read FLastError;
  end;

implementation

{ ── internal thread wrappers ─────────────────────────────────────────────── }

type
  TServerThread = class(TThread)
  private FServer: TCATTcpServer;
  protected procedure Execute; override;
  public constructor Create(S: TCATTcpServer);
  end;
  TPurgeThread = class(TThread)
  private FServer: TCATTcpServer;
  protected procedure Execute; override;
  public constructor Create(S: TCATTcpServer);
  end;

constructor TServerThread.Create(S: TCATTcpServer);
begin FServer := S; FreeOnTerminate := False; inherited Create(True); end;
procedure TServerThread.Execute; begin FServer.ServerLoop; end;

constructor TPurgeThread.Create(S: TCATTcpServer);
begin FServer := S; FreeOnTerminate := False; inherited Create(True); end;
procedure TPurgeThread.Execute; begin FServer.PurgeLoop; end;

{ ── TCATTcpClientThread ───────────────────────────────────────────────────── }

constructor TCATTcpClientThread.Create(AServer: TCATTcpServer; ASocket: TSocket);
begin
  FServer     := AServer;
  FSocket     := ASocket;
  FBuf        := '';
  FMarkDelete := False;
  FClientIds  := TStringList.Create;
  FIdLock     := TCriticalSection.Create;
  FreeOnTerminate := False;
  inherited Create(True);
end;

destructor TCATTcpClientThread.Destroy;
begin
  FClientIds.Free;
  FIdLock.Free;
  inherited;
end;

procedure TCATTcpClientThread.AddClientId(const Id: string);
var lo: string;
begin
  lo := LowerCase(Id);
  FIdLock.Acquire;
  try
    if FClientIds.IndexOf(lo) < 0 then FClientIds.Add(lo);
  finally FIdLock.Release; end;
end;

procedure TCATTcpClientThread.RemoveClientId(const Id: string);
var lo: string; idx: Integer;
begin
  lo := LowerCase(Id);
  FIdLock.Acquire;
  try
    idx := FClientIds.IndexOf(lo);
    if idx >= 0 then FClientIds.Delete(idx);
  finally FIdLock.Release; end;
end;

procedure TCATTcpClientThread.SendStr(const S: string);
// TCP не обязан отдавать весь буфер за один send: длинный ответ (IF, ZZEB,
// список режимов) мог уехать обрезанным — и молча, потому что усечение здесь
// не ошибка. Дописываем остаток, пока он есть.
// Пишем через WebUtils.SockSend, а не голым fpSend: он ставит MSG_NOSIGNAL.
// Без него запись в закрытый клиентом сокет — SIGPIPE, а тот по умолчанию
// убивает весь процесс; с ним send просто вернёт -1, и клиент выбрасывается.
var
  buf:  AnsiString;
  sent, n: Integer;
begin
  if FSocket = SOCK_INVALID then Exit;
  buf := AnsiString(S);
  if Length(buf) = 0 then Exit;
  sent := 0;
  while sent < Length(buf) do
  begin
    n := SockSend(FSocket, @buf[sent + 1], Length(buf) - sent, 0);
    if n <= 0 then
    begin
      FMarkDelete := True;
      Terminate;
      Exit;
    end;
    Inc(sent, n);
  end;
end;

procedure TCATTcpClientThread.HandleCmd(const Cmd: string);
const
  ZZGA_PFX = 'ZZGA';
  ZZGR_PFX = 'ZZGR';
var
  cmd2, guid, resp: string;
begin
  cmd2 := Trim(Cmd);
  if cmd2 = '' then Exit;
  if (Length(cmd2) >= 4) and (UpperCase(Copy(cmd2, 1, 4)) = ZZGA_PFX) then begin
    if Length(cmd2) >= 40 then begin // 4 + 36
      guid := LowerCase(Copy(cmd2, 5, 36));
      AddClientId(guid);
      SendStr(ZZGA_PFX + guid + ';');
    end else
      SendStr('?;');
  end else if (Length(cmd2) >= 4) and (UpperCase(Copy(cmd2, 1, 4)) = ZZGR_PFX) then begin
    if Length(cmd2) >= 40 then begin
      guid := LowerCase(Copy(cmd2, 5, 36));
      RemoveClientId(guid);
      SendStr(ZZGR_PFX + guid + ';');
    end else
      SendStr('?;');
  end else begin
    resp := FServer.FEngine.Parse(cmd2);
    if resp <> '' then SendStr(resp);
  end;
end;

procedure TCATTcpClientThread.ProcessBuffer;
var tpos: Integer; msg: string;
begin
  repeat
    tpos := Pos(';', FBuf);
    if tpos = 0 then Break;
    msg  := Copy(FBuf, 1, tpos);
    Delete(FBuf, 1, tpos);
    HandleCmd(msg);
  until tpos = 0;
  if Length(FBuf) > 1024 then FBuf := '';
end;

procedure TCATTcpClientThread.Execute;
var
  rawbuf: array[0..CAT_TCP_BUF_SIZE-1] of Byte;
  n:      Integer;
  chunk:  AnsiString;
begin
  if FServer.SendWelcome then
    SendStr(CAT_TCP_WELCOME);

  while not Terminated do begin
    {$IFDEF MSWINDOWS}
    n := recv(FSocket, rawbuf[0], CAT_TCP_BUF_SIZE, 0);
    if n = SOCKET_ERROR then begin FMarkDelete := True; Break; end;
    {$ELSE}
    n := fpRecv(FSocket, @rawbuf[0], CAT_TCP_BUF_SIZE, 0);
    if n <= 0 then begin FMarkDelete := True; Break; end;
    {$ENDIF}
    SetLength(chunk, n);
    Move(rawbuf[0], chunk[1], n);
    FBuf := FBuf + string(chunk);
    ProcessBuffer;
  end;

  SockShutdown(FSocket);
  SockClose(FSocket);
  FSocket := SOCK_INVALID;
end;

procedure TCATTcpClientThread.SendData(const Msg: string; const IdLimit: TStringList);
var
  i:      Integer;
  doSend: Boolean;
begin
  if IdLimit = nil then begin
    SendStr(Msg);
    Exit;
  end;
  doSend := False;
  FIdLock.Acquire;
  try
    for i := 0 to FClientIds.Count-1 do
      if IdLimit.IndexOf(FClientIds[i]) >= 0 then begin
        doSend := True;
        Break;
      end;
  finally FIdLock.Release; end;
  if doSend then SendStr(Msg);
end;

function TCATTcpClientThread.IsMarkedForDelete: Boolean;
begin Result := FMarkDelete; end;

procedure TCATTcpClientThread.Disconnect;
begin
  Terminate;
  SockShutdown(FSocket);
end;

{ ── TCATTcpServer ─────────────────────────────────────────────────────────── }

constructor TCATTcpServer.Create(AEngine: TCATEngine; APort: Integer);
{$IFDEF MSWINDOWS}
var wsa: TWSAData;
{$ENDIF}
begin
  inherited Create;
  FEngine      := AEngine;
  FPort        := APort;
  FListenSock  := SOCK_INVALID;
  FClients     := TList.Create;
  FClientsLock := TCriticalSection.Create;
  FStopEvent   := TEvent.Create(nil, True, False, '');
  FRunning     := False;
  FSendWelcome := True;
  FLastError   := '';
  {$IFDEF MSWINDOWS}
  WSAStartup($0202, wsa);
  {$ENDIF}
end;

destructor TCATTcpServer.Destroy;
begin
  Stop;
  FClients.Free;
  FClientsLock.Free;
  FStopEvent.Free;
  {$IFDEF MSWINDOWS}
  WSACleanup;
  {$ENDIF}
  inherited;
end;

function TCATTcpServer.InitListen: Boolean;
var
  Addr: {$IFDEF MSWINDOWS}TSockAddrIn{$ELSE}TInetSockAddr{$ENDIF};
  One:  Integer;
begin
  Result := False;
  {$IFDEF MSWINDOWS}
  FListenSock := socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
  {$ELSE}
  FListenSock := fpSocket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
  {$ENDIF}
  if FListenSock = SOCK_INVALID then begin
    FLastError := 'Cannot create socket';
    Exit;
  end;

  One := 1;
  FillChar(Addr, SizeOf(Addr), 0);
  Addr.sin_family := AF_INET;
  Addr.sin_port   := htons(FPort);

  {$IFDEF MSWINDOWS}
  setsockopt(FListenSock, SOL_SOCKET, SO_REUSEADDR, PChar(@One), SizeOf(One));
  Addr.sin_addr.S_addr := INADDR_ANY;
  if bind(FListenSock, @Addr, SizeOf(Addr)) = SOCKET_ERROR then begin
    FLastError := 'bind failed';
    SockClose(FListenSock); FListenSock := SOCK_INVALID; Exit;
  end;
  if listen(FListenSock, 8) = SOCKET_ERROR then begin
    FLastError := 'listen failed';
    SockClose(FListenSock); FListenSock := SOCK_INVALID; Exit;
  end;
  {$ELSE}
  fpSetSockOpt(FListenSock, SOL_SOCKET, SO_REUSEADDR, @One, SizeOf(One));
  Addr.sin_addr.s_addr := htonl(INADDR_ANY);
  if fpBind(FListenSock, @Addr, SizeOf(Addr)) <> 0 then begin
    FLastError := 'bind failed on port ' + IntToStr(FPort);
    SockClose(FListenSock); FListenSock := SOCK_INVALID; Exit;
  end;
  if fpListen(FListenSock, 8) <> 0 then begin
    FLastError := 'listen failed';
    SockClose(FListenSock); FListenSock := SOCK_INVALID; Exit;
  end;
  {$ENDIF}
  Result := True;
end;

procedure TCATTcpServer.Start;
begin
  if FRunning then Exit;
  if not InitListen then Exit;
  FRunning := True;
  FServerThread := TServerThread.Create(Self);
  TServerThread(FServerThread).Start;
  FPurgeThread := TPurgeThread.Create(Self);
  TPurgeThread(FPurgeThread).Start;
end;

procedure TCATTcpServer.Stop;
var
  i:   Integer;
  cli: TCATTcpClientThread;
begin
  if not FRunning then Exit;
  FRunning := False;

  // Будим PurgeLoop немедленно, чтобы он не ждал 5 секунд
  FStopEvent.SetEvent;

  // Закрываем listen socket — разблокирует fpAccept в ServerLoop
  if FListenSock <> SOCK_INVALID then begin
    SockShutdown(FListenSock);
    SockClose(FListenSock);
    FListenSock := SOCK_INVALID;
  end;

  // Разрываем всех клиентов до ожидания потоков
  FClientsLock.Acquire;
  try
    for i := 0 to FClients.Count-1 do
      TCATTcpClientThread(FClients[i]).Disconnect;
  finally FClientsLock.Release; end;

  if Assigned(FServerThread) then begin
    FServerThread.Terminate;
    FServerThread.WaitFor;
    FreeAndNil(FServerThread);
  end;
  if Assigned(FPurgeThread) then begin
    FPurgeThread.Terminate;
    FPurgeThread.WaitFor;
    FreeAndNil(FPurgeThread);
  end;

  // Теперь безопасно ждём завершения клиентских потоков и освобождаем
  FClientsLock.Acquire;
  try
    for i := 0 to FClients.Count-1 do begin
      cli := TCATTcpClientThread(FClients[i]);
      cli.WaitFor;
      cli.Free;
    end;
    FClients.Clear;
  finally FClientsLock.Release; end;

  FStopEvent.ResetEvent;
end;

procedure TCATTcpServer.ServerLoop;
var
  cli:        TCATTcpClientThread;
  clientSock: TSocket;
  clientAddr: {$IFDEF MSWINDOWS}TSockAddrIn{$ELSE}TInetSockAddr{$ENDIF};
  addrLen:    {$IFDEF MSWINDOWS}Integer{$ELSE}TSockLen{$ENDIF};
begin
  addrLen := SizeOf(clientAddr);
  while FRunning do begin
    {$IFDEF MSWINDOWS}
    clientSock := accept(FListenSock, @clientAddr, @addrLen);
    if clientSock = INVALID_SOCKET then Break;
    {$ELSE}
    clientSock := fpAccept(FListenSock, @clientAddr, @addrLen);
    if clientSock = SOCK_INVALID then Break;
    {$ENDIF}
    if not FRunning then begin SockClose(clientSock); Break; end;

    cli := TCATTcpClientThread.Create(Self, clientSock);
    FClientsLock.Acquire;
    try FClients.Add(cli);
    finally FClientsLock.Release; end;
    cli.Start;
  end;
end;

procedure TCATTcpServer.PurgeLoop;
var
  i:   Integer;
  cli: TCATTcpClientThread;
  del: TList;
begin
  del := TList.Create;
  try
    while FRunning do begin
      if FStopEvent.WaitFor(5000) <> wrTimeout then Break;
      del.Clear;
      FClientsLock.Acquire;
      try
        for i := FClients.Count-1 downto 0 do begin
          cli := TCATTcpClientThread(FClients[i]);
          if cli.IsMarkedForDelete then begin
            del.Add(cli);
            FClients.Delete(i);
          end;
        end;
      finally FClientsLock.Release; end;
      for i := 0 to del.Count-1 do begin
        cli := TCATTcpClientThread(del[i]);
        cli.WaitFor;
        cli.Free;
      end;
    end;
  finally del.Free; end;
end;

procedure TCATTcpServer.Broadcast(const Msg: string; const IdLimit: TStringList);
var
  i:   Integer;
  cli: TCATTcpClientThread;
begin
  FClientsLock.Acquire;
  try
    for i := 0 to FClients.Count-1 do begin
      cli := TCATTcpClientThread(FClients[i]);
      if not cli.IsMarkedForDelete then
        cli.SendData(Msg, IdLimit);
    end;
  finally FClientsLock.Release; end;
end;

function TCATTcpServer.ClientCount: Integer;
var i: Integer;
begin
  Result := 0;
  FClientsLock.Acquire;
  try
    for i := 0 to FClients.Count-1 do
      if not TCATTcpClientThread(FClients[i]).IsMarkedForDelete then Inc(Result);
  finally FClientsLock.Release; end;
end;

end.