unit TCIServer; { TCIServer.pas — транспорт TCI: WebSocket-сервер (роль сервера играем мы, как ExpertSDR3; клиенты — логгеры, скиммеры, программы цифровых видов). Что делает: • слушает TCP-порт (умолчание 40001), принимает HTTP-Upgrade на WebSocket по любому пути (клиенты ходят на ws://host:40001/); • режет входящие текстовые фреймы на команды и отдаёт их наверх (OnCommand) прямо в потоке клиента — маршалинг в поток контроллера делает TCIAdapter, как это устроено у CAT; • рассылает строки всем клиентам (Broadcast) — сервер TCI обязан синхронизировать всех подключённых (§3.5); • тикает OnTick (умолчание 20 мс) — по нему адаптер шлёт показания измерителей с индивидуальным для каждого клиента периодом. Отправка НИКОГДА не блокирует того, кто зовёт Send/Broadcast: строка кладётся в очередь клиента, а в сокет её пишет тик-поток (FlushClients). Иначе медленный клиент останавливал бы UI-поток на секунду за раз — уведомления рождаются в OnState, то есть внутри Changed() контроллера. Владение объектом клиента: создаёт accept-поток, освобождает ТОЛЬКО тик-поток (ReapClients) и только после того, как клиентский поток честно вышел. Никто больше клиентов не освобождает — поэтому указатель, взятый под FClientLock, остаётся валидным, пока тик-поток не сделает следующий проход. Бинарные фреймы (потоки IQ/аудио, §3.4) пока не обрабатываются: этап 2, см. doc/TCI.md. Приходящие от клиента binary-фреймы молча отбрасываются. Сокеты и WS-фреймы переиспользованы из веб-подсистемы (WebUtils/WsClient): тот же код handshake и та же схема «поток на клиента + accept-поток», что в WebServer. Авторизации у TCI нет by design. Порт слушается там, где сказано в настройках; умолчание — 127.0.0.1, чтобы наружу он не торчал без спроса. } {$IFDEF FPC} {$MODE Delphi} {$LONGSTRINGS ON} {$ENDIF} interface uses Classes, SysUtils, WebUtils, WsClient, TCIProtocol {$IFDEF WINDOWS}, Windows, WinSock2{$ELSE}, Sockets{$ENDIF}, SyncObjs; // ← после платформенных юнитов (конфликт идентификатора Create) const TCI_MAX_CLIENTS = 8; TCI_TICK_MS = 20; // период OnTick (сенсоры троттлятся адаптером) TCI_WS_GUID = '258EAFA5-E914-47DA-95CA-C5AB0DC85B11'; TCI_SEND_TIMEOUT = 300; // мс на SockSend, иначе клиент считается мёртвым TCI_OUT_MAX = 4000; // потолок очереди отправки на клиента (строк) TCI_OUT_CHUNK = 3800; // склейка очереди в один фрейм, символов TCI_MSG_MAX = 65536; // потолок собираемого из фрагментов сообщения TCI_STOP_WAIT_MS = 10000; // сколько ждём выхода клиентских потоков в Stop type TTCIServer = class; { Один подключённый клиент: WS-сокет, его личные подписки и очередь отправки. Подписки на сенсоры в TCI индивидуальны (RX_SENSORS_ENABLE «отправляется только клиентом»), поэтому живут здесь, а не в адаптере. Параметры потоков (§4.3) — тоже клиентские, их держит адаптер по ссылке на этот объект. } TTCIClient = class private FWs: TWsClient; FReady: Boolean; // пачка инициализации отправлена FRxSensors: Boolean; FRxSensorsMs: Integer; FRxSensorsAt: QWord; // тик последней отправки FTxSensors: Boolean; FTxSensorsMs: Integer; FTxSensorsAt: QWord; // Параметры бинарных потоков (§3.4): по спецификации это настройки // КЛИЕНТА, а не устройства — свои у каждого подключения. FIQRate: Integer; FAudioRate: Integer; FAudioSamples: Integer; FAudioChannels: Integer; FAudioSampleType: string; FTxBuffering: Integer; // Очередь отправки: пишут любые потоки, читает тик-поток. FOutLock: TCriticalSection; FOut: array of string; FOutCount: Integer; FDead: Boolean; // сокет уже не пишется — гасим соединение FKilled: Boolean; // shutdown сокета уже сделан FClosed: Boolean; // клиентский поток вышел (можно освобождать) public constructor Create(AWs: TWsClient); destructor Destroy; override; { Строку в очередь клиенту. False — соединение уже мертво. Не блокирует. } function Send(const S: string): Boolean; { Слить очередь в сокет. Зовёт только тик-поток. False — клиент умер. } function Flush: Boolean; { Пометить мёртвым и разбудить его поток (shutdown сокета). } procedure Kill; property Ws: TWsClient read FWs; property Ready: Boolean read FReady write FReady; property Dead: Boolean read FDead; property RxSensors: Boolean read FRxSensors write FRxSensors; property RxSensorsMs: Integer read FRxSensorsMs write FRxSensorsMs; property RxSensorsAt: QWord read FRxSensorsAt write FRxSensorsAt; property TxSensors: Boolean read FTxSensors write FTxSensors; property TxSensorsMs: Integer read FTxSensorsMs write FTxSensorsMs; property TxSensorsAt: QWord read FTxSensorsAt write FTxSensorsAt; property IQRate: Integer read FIQRate write FIQRate; property AudioRate: Integer read FAudioRate write FAudioRate; property AudioSamples: Integer read FAudioSamples write FAudioSamples; property AudioChannels: Integer read FAudioChannels write FAudioChannels; property AudioSampleType: string read FAudioSampleType write FAudioSampleType; property TxBuffering: Integer read FTxBuffering write FTxBuffering; end; TTCIClientEvent = procedure(Client: TTCIClient) of object; TTCICommandEvent = procedure(Client: TTCIClient; const Cmd: string) of object; TTCIServer = class private FListenSock: TSocket; FClients: array[0..TCI_MAX_CLIENTS-1] of TTCIClient; FClientCount: Integer; FClientLock: TCriticalSection; FAcceptThread: TThread; FTickThread: TThread; FThreadCount: LongInt; // живых клиентских потоков (Interlocked*) FRunning: Boolean; FStopping: Boolean; FPort: Word; FBindIP: string; FOnCommand: TTCICommandEvent; FOnConnect: TTCIClientEvent; FOnDisconnect: TTCIClientEvent; FOnTick: TThreadMethod; function InitListen: Boolean; procedure ReapClients; // освободить клиентов, чьи потоки вышли procedure FlushClients; // слить очереди в сокеты (вне FClientLock) public constructor Create; destructor Destroy; override; { Настройка слушателя. Применяется при следующем Start. False — адрес не разобран (порт не откроется). } function Configure(APort: Word; const ABindIP: string): Boolean; function Start: Boolean; procedure Stop; function Running: Boolean; { Всем клиентам, прошедшим инициализацию. Skip — кого пропустить (обычно автора изменения не пропускаем: сервер отвечает и ему тоже, это и есть подтверждение установки). Только кладёт в очереди. } procedure Broadcast(const S: string; Skip: TTCIClient = nil); { Обход клиентов под локом — для рассылки с индивидуальным периодом. Proc обязана быть быстрой: она держит FClientLock. } procedure EnumClients(Proc: TTCIClientEvent); function ClientCount: Integer; { Внутреннее (зовётся потоками сервера). } procedure AcceptLoop; procedure TickLoop; procedure HandleClient(Client: TTCIClient); procedure ThreadDone; // клиентский поток отработал property Port: Word read FPort; property BindIP: string read FBindIP; { Идёт остановка: адаптер не должен начинать новых вызовов в поток контроллера — тот, кто нас останавливает, обычно и есть поток контроллера, и Synchronize из клиентского потока в него не вернётся. } property Stopping: Boolean read FStopping; property OnCommand: TTCICommandEvent read FOnCommand write FOnCommand; property OnConnect: TTCIClientEvent read FOnConnect write FOnConnect; property OnDisconnect: TTCIClientEvent read FOnDisconnect write FOnDisconnect; property OnTick: TThreadMethod read FOnTick write FOnTick; end; { IPv4 из строки в сетевом порядке. Строгий: ровно четыре десятичных октета 0..255. '' и '0.0.0.0' — это INADDR_ANY (слушать везде), и только они: «ошибка разбора = слушаем всё» в протоколе без авторизации недопустима. } function TCIParseIPv4(const S: string; out Addr: LongWord): Boolean; implementation type TTCIAcceptThread = class(TThread) private FServer: TTCIServer; protected procedure Execute; override; public constructor Create(AServer: TTCIServer); end; TTCITickThread = class(TThread) private FServer: TTCIServer; protected procedure Execute; override; public constructor Create(AServer: TTCIServer); end; TTCIClientThread = class(TThread) private FServer: TTCIServer; FClient: TTCIClient; protected procedure Execute; override; public constructor Create(AServer: TTCIServer; AClient: TTCIClient); end; constructor TTCIAcceptThread.Create(AServer: TTCIServer); begin inherited Create(True); FServer := AServer; FreeOnTerminate := False; end; procedure TTCIAcceptThread.Execute; begin FServer.AcceptLoop; end; constructor TTCITickThread.Create(AServer: TTCIServer); begin inherited Create(True); FServer := AServer; FreeOnTerminate := False; end; procedure TTCITickThread.Execute; begin FServer.TickLoop; end; constructor TTCIClientThread.Create(AServer: TTCIServer; AClient: TTCIClient); begin inherited Create(True); FServer := AServer; FClient := AClient; FreeOnTerminate := True; end; procedure TTCIClientThread.Execute; begin try try FServer.HandleClient(FClient); except // Разбор фрейма рухнул — соединение всё равно закрываем штатно, иначе // клиент остался бы висеть в массиве до остановки сервера. end; finally // Освобождать себя нельзя: объект переиспользуется рассылкой из чужих // потоков. Помечаем «поток вышел» — освободит тик-поток (ReapClients). FClient.FClosed := True; FServer.ThreadDone; end; end; function TCIParseIPv4(const S: string; out Addr: LongWord): Boolean; var Oct: array[0..3] of LongWord; N, i, Start, V: Integer; Part: string; c: Char; begin Addr := 0; // INADDR_ANY Result := False; if (Trim(S) = '') or (Trim(S) = '0.0.0.0') then Exit(True); N := 0; Start := 1; for i := 1 to Length(S) + 1 do if (i > Length(S)) or (S[i] = '.') then begin if N > 3 then Exit(False); Part := Copy(S, Start, i - Start); if (Part = '') or (Length(Part) > 3) then Exit(False); for c in Part do if (c < '0') or (c > '9') then Exit(False); V := StrToIntDef(Part, -1); if (V < 0) or (V > 255) then Exit(False); Oct[N] := LongWord(V); Inc(N); Start := i + 1; end; if N <> 4 then Exit(False); // Сетевой порядок байт: первый октет — младший байт in_addr. Addr := Oct[0] or (Oct[1] shl 8) or (Oct[2] shl 16) or (Oct[3] shl 24); Result := True; end; { ═══════════════════════════════════════════════════════════════════════════ TTCIClient ═══════════════════════════════════════════════════════════════════════════ } constructor TTCIClient.Create(AWs: TWsClient); begin inherited Create; FWs := AWs; FReady := False; FRxSensors := False; FRxSensorsMs := 200; FTxSensors := False; FTxSensorsMs := 200; FOutLock := TCriticalSection.Create; FOutCount := 0; SetLength(FOut, 64); // Умолчания параметров потоков — как в §4.3 (клиент их обычно переопределяет). FIQRate := 48000; FAudioRate := 48000; FAudioSamples := 2048; FAudioChannels := 2; FAudioSampleType := 'float32'; FTxBuffering := 50; end; destructor TTCIClient.Destroy; begin FOutLock.Free; inherited; end; function TTCIClient.Send(const S: string): Boolean; begin Result := False; if (S = '') or FDead or (FWs = nil) then Exit; FOutLock.Enter; try if FDead then Exit; // Очередь переполнилась: клиент не читает сокет быстрее, чем мы пишем. // Копить дальше нечестно (память + отставшее состояние), рвём соединение. if FOutCount >= TCI_OUT_MAX then begin FDead := True; Exit; end; if FOutCount >= Length(FOut) then SetLength(FOut, Length(FOut) * 2); FOut[FOutCount] := S; Inc(FOutCount); Result := True; finally FOutLock.Leave; end; end; function TTCIClient.Flush: Boolean; var Batch: array of string; N, i: Integer; Chunk: string; begin Result := not FDead; // Мёртвым клиента могла пометить и очередь (переполнилась в чужом потоке — // там гасить сокет нельзя, лок чужой). Добиваем здесь: без shutdown его // поток так и висел бы в recv, а объект никогда бы не освободился. if FDead then begin Kill; Exit; end; FOutLock.Enter; try N := FOutCount; if N > 0 then begin SetLength(Batch, N); for i := 0 to N - 1 do begin Batch[i] := FOut[i]; FOut[i] := ''; end; FOutCount := 0; end; finally FOutLock.Leave; end; if N = 0 then Exit; // Склейка: несколько команд в одном фрейме протокол разрешает (§3.1), а // syscall'ов и заголовков становится в разы меньше. Chunk := ''; for i := 0 to N - 1 do begin if (Chunk <> '') and (Length(Chunk) + Length(Batch[i]) > TCI_OUT_CHUNK) then begin if not FWs.SendText(Chunk) then begin Kill; Exit(False); end; Chunk := ''; end; Chunk := Chunk + Batch[i]; end; if Chunk <> '' then if not FWs.SendText(Chunk) then begin Kill; Exit(False); end; end; procedure TTCIClient.Kill; begin FOutLock.Enter; try FDead := True; FOutCount := 0; if FKilled then Exit; // shutdown уже был — второй раз незачем FKilled := True; finally FOutLock.Leave; end; if FWs <> nil then begin FWs.State := wsClosed; SockShutdown(FWs.Socket); // будим поток клиента, висящий в recv end; end; { ═══════════════════════════════════════════════════════════════════════════ TTCIServer — жизненный цикл ═══════════════════════════════════════════════════════════════════════════ } constructor TTCIServer.Create; {$IFDEF WINDOWS} var WSAData: TWSAData; {$ENDIF} begin inherited Create; {$IFDEF WINDOWS} WSAStartup($0202, WSAData); // refcounted: свой вызов на каждый сервер {$ENDIF} FListenSock := SOCK_INVALID; FClientCount := 0; FClientLock := TCriticalSection.Create; FPort := TCI_DEFAULT_PORT; FBindIP := '127.0.0.1'; end; destructor TTCIServer.Destroy; begin Stop; FClientLock.Free; {$IFDEF WINDOWS} WSACleanup; {$ENDIF} inherited; end; function TTCIServer.Configure(APort: Word; const ABindIP: string): Boolean; var Dummy: LongWord; begin FPort := APort; FBindIP := ABindIP; Result := (APort <> 0) and TCIParseIPv4(ABindIP, Dummy); end; function TTCIServer.Running: Boolean; begin Result := FRunning; end; function TTCIServer.InitListen: Boolean; var Addr: {$IFDEF WINDOWS}TSockAddrIn{$ELSE}TInetSockAddr{$ENDIF}; One: Integer; IP: LongWord; begin Result := False; // Кривой адрес — отказ. Молча свалиться в INADDR_ANY нельзя: в TCI нет // авторизации, и открытый наружу порт отдаёт управление передатчиком. if not TCIParseIPv4(FBindIP, IP) then Exit; if FPort = 0 then Exit; {$IFDEF WINDOWS} FListenSock := socket(AF_INET, SOCK_STREAM, IPPROTO_TCP); {$ELSE} FListenSock := fpSocket(AF_INET, SOCK_STREAM, IPPROTO_TCP); {$ENDIF} if FListenSock = SOCK_INVALID then Exit; One := 1; {$IFDEF WINDOWS} setsockopt(FListenSock, SOL_SOCKET, SO_REUSEADDR, @One, SizeOf(One)); FillChar(Addr, SizeOf(Addr), 0); Addr.sin_family := AF_INET; Addr.sin_port := htons(FPort); Addr.sin_addr.S_addr := IP; if bind(FListenSock, @Addr, SizeOf(Addr)) = SOCKET_ERROR then Exit; if listen(FListenSock, 5) = SOCKET_ERROR then Exit; {$ELSE} fpSetSockOpt(FListenSock, SOL_SOCKET, SO_REUSEADDR, @One, SizeOf(One)); FillChar(Addr, SizeOf(Addr), 0); Addr.sin_family := AF_INET; Addr.sin_port := htons(FPort); Addr.sin_addr.s_addr := IP; if fpBind(FListenSock, @Addr, SizeOf(Addr)) <> 0 then Exit; if fpListen(FListenSock, 5) <> 0 then Exit; {$ENDIF} Result := True; end; function TTCIServer.Start: Boolean; begin Result := False; if FRunning then Exit; if not InitListen then begin if FListenSock <> SOCK_INVALID then begin SockClose(FListenSock); FListenSock := SOCK_INVALID; end; Exit; end; FStopping := False; FRunning := True; FAcceptThread := TTCIAcceptThread.Create(Self); TTCIAcceptThread(FAcceptThread).Start; FTickThread := TTCITickThread.Create(Self); TTCITickThread(FTickThread).Start; Result := True; end; procedure TTCIServer.Stop; var i, Waited: Integer; begin if not FRunning then Exit; FStopping := True; // адаптер перестаёт звать Invoke (см. property Stopping) FRunning := False; // Шаг 1: гасим listen-сокет. SockShutdown обязателен до close — иначе // fpAccept в accept-потоке не разблокируется (см. WebServer.Stop). if FListenSock <> SOCK_INVALID then begin SockShutdown(FListenSock); SockClose(FListenSock); FListenSock := SOCK_INVALID; end; // Шаг 2: будим клиентские потоки, висящие в recv. FClientLock.Enter; try for i := 0 to FClientCount - 1 do if FClients[i] <> nil then FClients[i].Kill; finally FClientLock.Leave; end; // Шаг 3: свои потоки (клиентские — FreeOnTerminate, ждём их отдельно). if FAcceptThread <> nil then begin FAcceptThread.WaitFor; FreeAndNil(FAcceptThread); end; if FTickThread <> nil then begin FTickThread.WaitFor; FreeAndNil(FTickThread); end; // Шаг 4: ждём выхода клиентских потоков. Прокачивая очередь Synchronize: // Stop зовёт поток контроллера (UI), а клиентский поток может как раз в нём // висеть на FController.Invoke. Без прокачки это гарантированный взаимный // клин, а по его истечении — освобождение объекта из-под живого потока. Waited := 0; while (FThreadCount > 0) and (Waited < TCI_STOP_WAIT_MS) do begin if GetCurrentThreadId = MainThreadID then CheckSynchronize(5) else Sleep(5); Inc(Waited, 5); end; // Шаг 5: зачистка. Если поток всё же не вышел (не должно случаться: сокеты // закрыты, очередь прокачана), объект НЕ освобождаем — утечка на выходе // несравнимо дешевле обращения к освобождённой памяти из живого потока. FClientLock.Enter; try for i := 0 to FClientCount - 1 do if (FClients[i] <> nil) and FClients[i].FClosed then begin FClients[i].Ws.Free; FreeAndNil(FClients[i]); end; FClientCount := 0; finally FClientLock.Leave; end; end; { ═══════════════════════════════════════════════════════════════════════════ Приём соединений ═══════════════════════════════════════════════════════════════════════════ } procedure TTCIServer.AcceptLoop; var CSock: TSocket; Addr: {$IFDEF WINDOWS}TSockAddrIn{$ELSE}TInetSockAddr{$ENDIF}; ALen: {$IFDEF WINDOWS}Integer{$ELSE}TSockLen{$ENDIF}; Client: TTCIClient; T: TTCIClientThread; Full: Boolean; begin while FRunning do begin ALen := SizeOf(Addr); {$IFDEF WINDOWS} CSock := accept(FListenSock, @Addr, @ALen); {$ELSE} CSock := fpAccept(FListenSock, @Addr, @ALen); {$ENDIF} if CSock = SOCK_INVALID then begin if FRunning then Sleep(10); Continue; end; if not FRunning then begin SockClose(CSock); Break; end; SockSetSndTimeout(CSock, TCI_SEND_TIMEOUT); Client := TTCIClient.Create(TWsClient.Create(CSock)); Full := False; FClientLock.Enter; try // Слот берём под локом: место в массиве освобождает тик-поток. if FClientCount >= TCI_MAX_CLIENTS then Full := True else begin FClients[FClientCount] := Client; Inc(FClientCount); end; finally FClientLock.Leave; end; if Full then begin Client.Ws.Free; // закрывает сокет Client.Free; Continue; end; InterLockedIncrement(FThreadCount); T := TTCIClientThread.Create(Self, Client); T.Start; end; end; procedure TTCIServer.ReapClients; // Освобождение клиентов — единственное место во всей программе. Зовёт только // тик-поток, поэтому указатель, взятый кем угодно под FClientLock, живёт до // следующего прохода тика (а вне лока указателей никто не держит). var i, j, N: Integer; Doomed: array[0..TCI_MAX_CLIENTS-1] of TTCIClient; begin N := 0; FClientLock.Enter; try i := 0; while i < FClientCount do if (FClients[i] <> nil) and FClients[i].FClosed then begin Doomed[N] := FClients[i]; Inc(N); for j := i to FClientCount - 2 do FClients[j] := FClients[j + 1]; FClients[FClientCount - 1] := nil; Dec(FClientCount); end else Inc(i); finally FClientLock.Leave; end; for i := 0 to N - 1 do begin if Assigned(FOnDisconnect) then FOnDisconnect(Doomed[i]); Doomed[i].Ws.Free; // закрывает сокет Doomed[i].Free; end; end; procedure TTCIServer.FlushClients; // Запись в сокеты — вне FClientLock: медленный клиент не должен держать лок, // иначе Broadcast из потока контроллера снова начнёт ждать сеть. var Snap: array[0..TCI_MAX_CLIENTS-1] of TTCIClient; i, N: Integer; begin N := 0; FClientLock.Enter; try for i := 0 to FClientCount - 1 do if (FClients[i] <> nil) and not FClients[i].FClosed then begin Snap[N] := FClients[i]; Inc(N); end; finally FClientLock.Leave; end; for i := 0 to N - 1 do Snap[i].Flush; end; { ═══════════════════════════════════════════════════════════════════════════ Клиентский поток: handshake + разбор WS-фреймов ═══════════════════════════════════════════════════════════════════════════ } procedure TTCIServer.HandleClient(Client: TTCIClient); var Ws: TWsClient; R, HeaderEnd: Integer; Header, HeaderLC, Key, AcceptKey, Response, Text: string; Raw: array[0..4095] of Byte; RawLen, Rest: Integer; B0, B1: Byte; Masked, Fin, Pending: Boolean; PayLen, Need, i, j, Consumed, KPos, KEnd: Integer; Hi32: LongWord; Mask: array[0..3] of Byte; Payload: array of Byte; Opcode, MsgOp: Byte; Frag: string; Cmds: TStringList; begin Ws := Client.Ws; RawLen := 0; // ── HTTP-запрос: ждём конца заголовков ─────────────────────────────────── Header := ''; HeaderEnd := 0; repeat R := SockRecv(Ws.Socket, @Raw[RawLen], SizeOf(Raw) - RawLen, 0); if R <= 0 then begin Ws.State := wsClosed; Break; end; Inc(RawLen, R); SetLength(Header, RawLen); Move(Raw[0], Header[1], RawLen); HeaderEnd := System.Pos(#13#10#13#10, Header); until (HeaderEnd > 0) or (RawLen >= SizeOf(Raw)); if (Ws.State = wsClosed) or (HeaderEnd = 0) then Exit; Consumed := HeaderEnd + 3; // длина заголовков вместе с CRLFCRLF Header := Copy(Header, 1, Consumed); HeaderLC := LowerCase(Header); // Путь не проверяем: клиенты ходят на '/', но протокол его не оговаривает. if System.Pos('upgrade: websocket', HeaderLC) = 0 then begin Response := 'HTTP/1.1 426 Upgrade Required'#13#10 + 'Content-Length: 0'#13#10'Connection: close'#13#10#13#10; Ws.SendRaw(Response[1], Length(Response)); Exit; end; Key := ''; KPos := System.Pos('sec-websocket-key: ', HeaderLC); if KPos > 0 then begin Key := Copy(Header, KPos + 19, 100); KEnd := System.Pos(#13, Key); if KEnd > 0 then Key := Copy(Key, 1, KEnd - 1); Key := Trim(Key); end; // Пустой ключ = не WebSocket-клиент (или сломанный): Accept без ключа // формально считается валидным, и такое «соединение» потом молча висит. if Key = '' then begin Response := 'HTTP/1.1 400 Bad Request'#13#10 + 'Content-Length: 0'#13#10'Connection: close'#13#10#13#10; Ws.SendRaw(Response[1], Length(Response)); Exit; end; AcceptKey := Base64EncodeBytes(SHA1(Key + TCI_WS_GUID), 20); Response := 'HTTP/1.1 101 Switching Protocols'#13#10 + 'Upgrade: websocket'#13#10 + 'Connection: Upgrade'#13#10 + 'Sec-WebSocket-Accept: ' + AcceptKey + #13#10#13#10; if not Ws.SendRaw(Response[1], Length(Response)) then Exit; Ws.State := wsOpen; // Хвост первого пакета: клиент вправе прислать первый WS-фрейм в том же // сегменте, что и заголовки. Выбросить его — потерять первую команду. Rest := RawLen - Consumed; if Rest > 0 then Move(Raw[Consumed], Ws.BufData[0], Rest); Ws.BufLen := Rest; // Пачка инициализации + текущее состояние (§3.1) — дело адаптера. if Assigned(FOnConnect) then FOnConnect(Client); // ── Цикл WS-сообщений ──────────────────────────────────────────────────── Cmds := TStringList.Create; Frag := ''; MsgOp := 0; Pending := Rest > 0; // хвост handshake разбираем до первого recv try while FRunning and (Ws.State = wsOpen) and not Client.Dead do begin if not Pending then begin R := Ws.Recv; if R <= 0 then Break; end; Pending := False; while Ws.BufLen >= 2 do begin B0 := Ws.BufData[0]; B1 := Ws.BufData[1]; Fin := (B0 and $80) <> 0; Opcode := B0 and $0F; Masked := (B1 and $80) <> 0; PayLen := B1 and $7F; // RSV1..3 без согласованных расширений обязаны быть нулями. if (B0 and $70) <> 0 then begin Ws.State := wsClosed; Break; end; Need := 2; if PayLen = 126 then Inc(Need, 2) else if PayLen = 127 then Inc(Need, 8); if Masked then Inc(Need, 4); if Ws.BufLen < Need then Break; i := 2; if PayLen = 126 then begin PayLen := (Ws.BufData[2] shl 8) or Ws.BufData[3]; Inc(i, 2); end else if PayLen = 127 then begin // 64-битная длина: старшие четыре байта обязаны быть нулём, иначе // значение не помещается в Integer и превращается в отрицательное. Hi32 := (LongWord(Ws.BufData[2]) shl 24) or (LongWord(Ws.BufData[3]) shl 16) or (LongWord(Ws.BufData[4]) shl 8) or LongWord(Ws.BufData[5]); if Hi32 <> 0 then begin Ws.State := wsClosed; Break; end; Hi32 := (LongWord(Ws.BufData[6]) shl 24) or (LongWord(Ws.BufData[7]) shl 16) or (LongWord(Ws.BufData[8]) shl 8) or LongWord(Ws.BufData[9]); if Hi32 > LongWord(SizeOf(Raw)) then begin Ws.State := wsClosed; Break; end; PayLen := Integer(Hi32); Inc(i, 8); end; // Клиент ОБЯЗАН маскировать (RFC 6455 §5.1). Незамаскированный кадр — // либо не клиент, либо попытка прогнать через нас чужой трафик. if not Masked then begin Ws.State := wsClosed; Break; end; // Управляющие кадры: только короткие и только целиком (§5.5). if (Opcode >= $08) and ((PayLen > 125) or (not Fin)) then begin Ws.State := wsClosed; Break; end; // Фрейм крупнее приёмного буфера TWsClient никогда не соберётся — // BufLen упрётся в потолок и цикл встанет намертво. Рвём соединение: // команд такой длины у TCI нет, а бинарные потоки от клиента (TX-аудио) // мы пока не принимаем. if Need + PayLen > SizeOf(Raw) then begin Ws.State := wsClosed; Break; end; if Ws.BufLen < Need + PayLen then Break; Mask[0] := Ws.BufData[i]; Mask[1] := Ws.BufData[i+1]; Mask[2] := Ws.BufData[i+2]; Mask[3] := Ws.BufData[i+3]; Inc(i, 4); SetLength(Payload, PayLen); if PayLen > 0 then begin Move(Ws.BufData[i], Payload[0], PayLen); for j := 0 to PayLen - 1 do Payload[j] := Payload[j] xor Mask[j and 3]; end; Consumed := i + PayLen; if Ws.BufLen > Consumed then Move(Ws.BufData[Consumed], Ws.BufData[0], Ws.BufLen - Consumed); Ws.BufLen := Ws.BufLen - Consumed; case Opcode of $00, $01, $02: // данные: продолжение / текст / binary begin if Opcode = $00 then begin // Продолжение без начала — рассинхрон, дальше читать нечего. if MsgOp = 0 then begin Ws.State := wsClosed; Break; end; end else begin // Новое сообщение поверх недособранного — тоже рассинхрон. if MsgOp <> 0 then begin Ws.State := wsClosed; Break; end; MsgOp := Opcode; Frag := ''; end; // Копим только текст: binary — это TX-аудио от клиента, этап 2. if MsgOp = $01 then begin if Length(Frag) + PayLen > TCI_MSG_MAX then begin Ws.State := wsClosed; Break; end; if PayLen > 0 then begin SetLength(Text, PayLen); Move(Payload[0], Text[1], PayLen); Frag := Frag + Text; end; end; if Fin then begin if (MsgOp = $01) and Assigned(FOnCommand) and (Frag <> '') then begin TCISplit(Frag, Cmds); for j := 0 to Cmds.Count - 1 do FOnCommand(Client, Cmds[j]); end; MsgOp := 0; Frag := ''; end; end; $08: // close begin Ws.State := wsClosed; Break; end; $09: // ping → pong if PayLen > 0 then Ws.SendWsFrame($0A, Payload[0], PayLen) else Ws.SendWsFrame($0A, PayLen, 0); $0A: ; // pong — ничего не ждём else // Незнакомый opcode: по RFC соединение обязано закрыться. Ws.State := wsClosed; Break; end; end; end; finally Cmds.Free; end; end; { ═══════════════════════════════════════════════════════════════════════════ Рассылка и обход ═══════════════════════════════════════════════════════════════════════════ } procedure TTCIServer.Broadcast(const S: string; Skip: TTCIClient); var i: Integer; begin if S = '' then Exit; FClientLock.Enter; try // Send только кладёт строку в очередь клиента — лок держится микросекунды, // сколько бы клиент ни тормозил. В сокеты пишет тик-поток. for i := 0 to FClientCount - 1 do if (FClients[i] <> nil) and (FClients[i] <> Skip) and FClients[i].Ready then FClients[i].Send(S); finally FClientLock.Leave; end; end; procedure TTCIServer.EnumClients(Proc: TTCIClientEvent); var i: Integer; begin if not Assigned(Proc) then Exit; FClientLock.Enter; try for i := 0 to FClientCount - 1 do if (FClients[i] <> nil) and not FClients[i].FClosed then Proc(FClients[i]); finally FClientLock.Leave; end; end; procedure TTCIServer.ThreadDone; begin InterLockedDecrement(FThreadCount); end; function TTCIServer.ClientCount: Integer; begin Result := FClientCount; end; procedure TTCIServer.TickLoop; begin while FRunning do begin Sleep(TCI_TICK_MS); if not FRunning then Break; ReapClients; // отключившиеся — освобождаем только здесь if (FClientCount > 0) and Assigned(FOnTick) then FOnTick; FlushClients; // очереди → сокеты end; end; end.