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, SyncObjs, WebUtils, // SockClose, SockShutdown, SOCK_INVALID {$IFDEF MSWINDOWS} Windows, WinSock2 {$ELSE} BaseUnix, Sockets {$ENDIF}, 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); var buf: AnsiString; n: Integer; begin if FSocket = SOCK_INVALID then Exit; buf := AnsiString(S); if Length(buf) = 0 then Exit; {$IFDEF MSWINDOWS} n := send(FSocket, buf[1], Length(buf), 0); if n = SOCKET_ERROR then begin {$ELSE} n := fpSend(FSocket, @buf[1], Length(buf), 0); if n <= 0 then begin {$ENDIF} FMarkDelete := True; Terminate; 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; send: Boolean; begin if IdLimit = nil then begin SendStr(Msg); Exit; end; send := False; FIdLock.Acquire; try for i := 0 to FClientIds.Count-1 do if IdLimit.IndexOf(FClientIds[i]) >= 0 then begin send := True; Break; end; finally FIdLock.Release; end; if send 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.