ewsdr/DXClusterClient.pas

unit DXClusterClient;

{
  DXClusterClient.pas — telnet-клиент DX-кластера (одно соединение).

  Один рабочий поток: резолв → connect (неблокирующий + select, чтобы не висеть
  минуту на мёртвом хосте) → логин позывным → чтение строк. Разобранные споты
  кладутся ПРЯМО в TDXSpotStore (он потокобезопасен), лог соединения — во
  внутреннее кольцо. В главный поток ничего не маршалим: UI сам замечает
  изменения по Version стора и по LogVersion — как оверлеи замечают смену вида
  по ключам кэша. Никаких LCL-зависимостей.

  Разрыв → реконнект с backoff (RECONNECT_MIN..RECONNECT_MAX). Stop() делает
  SockShutdown, что разблокирует recv в потоке (на Linux одного close мало —
  см. WebUtils.SockShutdown).

  Формат спота (общий для DXSpider / AR-Cluster / CC Cluster):
    DX de UA3XYZ:    14025.0  DL1ABC       CW 599 tnx qso        1832Z
  Всё, что не начинается с 'DX de ', в стор не идёт — только в лог.
}

{$IFDEF FPC}
  {$MODE Delphi}
  {$LONGSTRINGS ON}
{$ENDIF}

interface

uses
  Classes, SysUtils, SyncObjs, StrUtils, DateUtils,
  WebUtils,        // SockClose/SockShutdown/SockRecv/SockSend/SOCK_INVALID
  DXSpotStore
  // ★ Windows.pas здесь НЕТ намеренно: он объявляет TCriticalSection как
  // ЗАПИСЬ (алиас TRTLCriticalSection) и, стоя в uses после SyncObjs,
  // перекрывал класс — под win64 сборка падала на TCriticalSection.Create
  // («identifier idents no member Create»). Всё, что нужно сокетам, даёт
  // WinSock2; так же сделано в WsClient и HPSDRNetwork.
  {$IFDEF MSWINDOWS}
    , WinSock2
  {$ELSE}
    , BaseUnix, Sockets, netdb
  {$ENDIF};

const
  DX_DEFAULT_HOST  = 'cluster.dxfun.com';
  DX_DEFAULT_PORT  = 8000;
  DX_CONNECT_MS    = 8000;    // потолок ожидания connect
  DX_CONNECT_SLICE_MS = 200;  // квант ожидания connect (чтобы Stop не ждал 8 с)
  DX_RECV_MS       = 1000;    // квант recv: просыпаемся проверить Terminated/очередь
  DX_RECONNECT_MIN = 5;       // с
  DX_RECONNECT_MAX = 60;      // с
  DX_LOGIN_GRACE_S = 6;       // не увидели приглашение — шлём позывной сами
  DX_POSTLOGIN_S   = 2;       // пауза после позывного перед post-login командами
  DX_LOG_LINES     = 300;     // кольцо лога соединения
  DX_SEND_RETRIES  = 20;      // повторов send по таймауту буфера, потом разрыв

type
  TDXClusterState = (dxsOff, dxsConnecting, dxsLogin, dxsOnline, dxsRetry, dxsError);

  TDXClusterClient = class
  private
    FThread:    TThread;
    FLock:      TCriticalSection;   // защищает конфиг, лог, очередь, состояние
    FStopEvent: TEvent;

    // конфиг (копия под FLock — поток читает её через GetConfig)
    FHost:      string;
    FPort:      Integer;
    FLogin:     string;
    FPassword:  string;
    FPostLogin: string;             // строки через LineEnding

    FStore:     TDXSpotStore;       // не владеет

    // Сокет живого соединения — зеркало под FLock. Нужен ровно для того, чтобы
    // Stop() мог сделать SockShutdown и вывести поток из блокирующего recv:
    // иначе выход из программы ждал бы до конца кванта recv (и до конца
    // таймаута send, если кластер перестал забирать данные). Поток снимает
    // зеркало ДО close, поэтому под локом хэндл гарантированно ещё открыт.
    FActiveSock: TSocket;

    FState:     TDXClusterState;
    FStatusMsg: string;
    FSpotCount: Int64;              // сколько спотов принято за сессию

    FLog:       array[0..DX_LOG_LINES-1] of string;
    FLogHead:   Integer;            // индекс следующей записи
    FLogCount:  Integer;
    FLogVer:    Int64;

    FOutQueue:  TStringList;        // строки на отправку в кластер

    procedure SetState(St: TDXClusterState; const Msg: string);
    procedure PublishSock(S: TSocket);   // зеркало хэндла для Stop (см. FActiveSock)
  public
    constructor Create(AStore: TDXSpotStore);
    destructor  Destroy; override;

    procedure Configure(const AHost: string; APort: Integer;
      const ALogin, APassword, APostLogin: string);
    procedure Start;
    // Неблокирующая половина Stop: разбудить поток и порвать сокет, не дожидаясь
    // его конца. Зовётся в начале выхода из программы, чтобы поток доживал
    // ПАРАЛЛЕЛЬНО с разборкой радио, а join в Stop уже никого не ждал.
    procedure RequestStop;
    procedure Stop;              // RequestStop + join
    function  Running: Boolean;

    // Отправить сырую строку в кластер (окно списка: sh/dx, set/filter, …).
    procedure SendCommand(const S: string);

    // Лог соединения — снимок в порядке приёма (старые → новые).
    procedure GetLog(Dst: TStrings);
    function  LogVersion: Int64;
    procedure AddLog(const S: string);

    function State: TDXClusterState;
    function StateText: string;
    function StatusMessage: string;
    function SpotsReceived: Int64;

    property Store: TDXSpotStore read FStore;
  end;

// Разбор строки кластера. True — это спот (Spot заполнен).
function ParseDXSpot(const Line: string; out Spot: TDXSpot): Boolean;
function DXStateName(St: TDXClusterState): string;

implementation

type
  TDXClusterThread = class(TThread)
  private
    FOwner:  TDXClusterClient;
    FSocket: TSocket;
    FRxBuf:  string;             // хвост неполной строки

    // конфиг сессии (снимок на время соединения)
    FHost:     string;
    FPort:     Integer;
    FLogin:    string;
    FPassword: string;
    FPostLogin: string;

    FLoginSent:  Boolean;
    FPassSent:   Boolean;
    FPostSent:   Boolean;
    FLoginAt:    TDateTime;
    FConnectAt:  TDateTime;

    function  Resolve(const Host: string; out Addr: LongWord): Boolean;
    function  ConnectSock: Boolean;
    procedure CloseSock;
    function  SendAll(const Data: string): Boolean;
    function  SendLine(const S: string): Boolean;
    procedure FlushOutQueue;
    procedure HandleChunk(const Data: string);
    procedure HandleLine(const Line: string);
    procedure CheckPrompts(const Tail: string);
    function  Session: Boolean;    // одно соединение; False — выйти совсем
  protected
    procedure Execute; override;
  public
    constructor Create(AOwner: TDXClusterClient);
  end;

function DXStateName(St: TDXClusterState): string;
begin
  case St of
    dxsOff:        Result := 'OFF';
    dxsConnecting: Result := 'CONNECTING';
    dxsLogin:      Result := 'LOGIN';
    dxsOnline:     Result := 'ONLINE';
    dxsRetry:      Result := 'RETRY';
  else             Result := 'ERROR';
  end;
end;

function SockWouldBlock: Boolean;
// Последняя ошибка сокета означает «данных пока нет / истёк SO_RCVTIMEO /
// прерван сигналом» — соединение живо. Всё остальное (ECONNRESET, ENOTCONN,
// EPIPE…) — разрыв: без этой проверки поток крутил бы recv в пустом цикле,
// вместо того чтобы уйти на реконнект.
var E: Integer;
begin
{$IFDEF MSWINDOWS}
  E := WSAGetLastError;
  Result := (E = WSAEWOULDBLOCK) or (E = WSAETIMEDOUT) or (E = WSAEINTR);
{$ELSE}
  E := fpgeterrno;
  Result := (E = ESysEAGAIN) or (E = ESysEWOULDBLOCK) or (E = ESysEINTR);
{$ENDIF}
end;

{ ── Разбор строки спота ──────────────────────────────────────────────────── }

function ParseDXSpot(const Line: string; out Spot: TDXSpot): Boolean;
// 'DX de <spotter>: <кГц> <позывной> <комментарий> <HHMM>Z [грид]'
var
  S, Rest, FreqStr, TimeTok: string;
  P, i: Integer;
  KHz: Double;
  FS: TFormatSettings;
begin
  Result := False;
  FillChar(Spot, SizeOf(Spot), 0);
  Spot.Call := ''; Spot.Spotter := ''; Spot.Comment := ''; Spot.TimeUTC := '';

  S := Trim(Line);
  if Length(S) < 12 then Exit;
  if not SameText(Copy(S, 1, 6), 'DX de ') then Exit;

  Rest := Copy(S, 7, MaxInt);
  P := Pos(':', Rest);
  if P <= 1 then Exit;
  Spot.Spotter := Trim(Copy(Rest, 1, P - 1));
  Rest := Trim(Copy(Rest, P + 1, MaxInt));

  // частота (кГц) — первый токен
  P := Pos(' ', Rest);
  if P <= 1 then Exit;
  FreqStr := Copy(Rest, 1, P - 1);
  Rest := TrimLeft(Copy(Rest, P + 1, MaxInt));

  FS := DefaultFormatSettings;
  FS.DecimalSeparator := '.';
  FreqStr := StringReplace(FreqStr, ',', '.', [rfReplaceAll]);
  if not TryStrToFloat(FreqStr, KHz, FS) then Exit;
  if (KHz <= 0) or (KHz > 100000000.0) then Exit;   // мусор/битая строка
  Spot.FreqHz := KHz * 1000.0;

  // позывной DX — второй токен
  P := Pos(' ', Rest);
  if P > 0 then
  begin
    Spot.Call := Copy(Rest, 1, P - 1);
    Rest := TrimLeft(Copy(Rest, P + 1, MaxInt));
  end
  else
  begin
    Spot.Call := Rest;
    Rest := '';
  end;
  if Spot.Call = '' then Exit;

  // Время: последний токен вида HHMMZ. Всё до него — комментарий, всё после
  // (обычно грид спотера) приклеиваем к комментарию — терять жалко.
  Rest := TrimRight(Rest);
  P := 0;
  for i := Length(Rest) - 4 downto 1 do
    if (Rest[i] in ['0'..'9']) and (Rest[i+1] in ['0'..'9']) and
       (Rest[i+2] in ['0'..'9']) and (Rest[i+3] in ['0'..'9']) and
       (UpCase(Rest[i+4]) = 'Z') and
       ((i = 1) or (Rest[i-1] = ' ')) then
    begin
      P := i;
      Break;
    end;
  if P > 0 then
  begin
    TimeTok := Copy(Rest, P, 4);
    Spot.TimeUTC := TimeTok;
    Spot.Comment := Trim(Copy(Rest, 1, P - 1));
    Rest := Trim(Copy(Rest, P + 5, MaxInt));
    if Rest <> '' then Spot.Comment := Trim(Spot.Comment + ' ' + Rest);
  end
  else
    Spot.Comment := Trim(Rest);

  Spot.Mode  := DXModeFromComment(Spot.Comment);
  // Комментарий молчит — берём моду с участка бэндплана: в CW и SSB её сплошь
  // и рядом просто не пишут, а участок эти две моды разделяет надёжно.
  // Помечаем такой спот: угаданное не должно выглядеть как сказанное спотером.
  Spot.ModeGuessed := False;
  if Spot.Mode = dxmUnknown then
  begin
    Spot.Mode := DXModeFromFreq(Spot.FreqHz);
    Spot.ModeGuessed := Spot.Mode <> dxmUnknown;
  end;
  Spot.Stamp := Now;
  Result := True;
end;

{ ── TDXClusterClient ─────────────────────────────────────────────────────── }

constructor TDXClusterClient.Create(AStore: TDXSpotStore);
{$IFDEF MSWINDOWS}
var wsa: TWSAData;
{$ENDIF}
begin
  inherited Create;
  FStore     := AStore;
  FLock      := TCriticalSection.Create;
  FStopEvent := TEvent.Create(nil, True, False, '');
  FOutQueue  := TStringList.Create;
  FHost      := DX_DEFAULT_HOST;
  FPort      := DX_DEFAULT_PORT;
  FState     := dxsOff;
  FLogHead   := 0;
  FLogCount  := 0;
  FLogVer    := 0;
  FSpotCount := 0;
  FActiveSock := SOCK_INVALID;
  {$IFDEF MSWINDOWS}
  WSAStartup($0202, wsa);
  {$ENDIF}
end;

destructor TDXClusterClient.Destroy;
begin
  Stop;
  FOutQueue.Free;
  FStopEvent.Free;
  FLock.Free;
  {$IFDEF MSWINDOWS}
  WSACleanup;
  {$ENDIF}
  inherited Destroy;
end;

procedure TDXClusterClient.Configure(const AHost: string; APort: Integer;
  const ALogin, APassword, APostLogin: string);
begin
  FLock.Enter;
  try
    // Смена адреса/учётки — команды, набранные для ПРЕЖНЕГО кластера, теряют
    // смысл (и ушли бы на новый). Чистим очередь.
    if (FHost <> Trim(AHost)) or (FPort <> APort) or (FLogin <> Trim(ALogin)) then
      FOutQueue.Clear;
    FHost      := Trim(AHost);
    FPort      := APort;
    FLogin     := Trim(ALogin);
    FPassword  := APassword;
    FPostLogin := APostLogin;
  finally
    FLock.Leave;
  end;
end;

procedure TDXClusterClient.Start;
begin
  if FThread <> nil then Exit;
  FLock.Enter;
  try
    if (FHost = '') or (FLogin = '') then
    begin
      FState := dxsError;
      FStatusMsg := 'host or callsign not set';
      Exit;
    end;
  finally
    FLock.Leave;
  end;
  FStopEvent.ResetEvent;
  SetState(dxsConnecting, '');
  FThread := TDXClusterThread.Create(Self);
end;

procedure TDXClusterClient.RequestStop;
begin
  // Всё, что не успели отправить, к следующему соединению уже неактуально.
  FLock.Enter;
  try
    FOutQueue.Clear;
  finally
    FLock.Leave;
  end;
  if FThread = nil then Exit;
  FThread.Terminate;
  FStopEvent.SetEvent;
  // Терминации мало: поток стоит в блокирующем recv и проснулся бы только по
  // SO_RCVTIMEO (до DX_RECV_MS), а на застрявшем send — до SO_SNDTIMEO. Всё это
  // время join держал бы главный поток, т.е. окно висело бы на выходе и на
  // переподключении после правки настроек. SockShutdown рвёт обе стороны
  // немедленно (на Linux одного close мало — см. WebUtils.SockShutdown).
  // Закрывает сокет по-прежнему сам поток: закрыть чужой хэндл значило бы
  // гонку с его же recv.
  FLock.Enter;
  try
    if FActiveSock <> SOCK_INVALID then SockShutdown(FActiveSock);
  finally
    FLock.Leave;
  end;
end;

procedure TDXClusterClient.Stop;
var T: TThread;
begin
  RequestStop;
  T := FThread;
  if T = nil then
  begin
    SetState(dxsOff, '');
    Exit;
  end;
  FThread := nil;
  T.WaitFor;
  T.Free;
  SetState(dxsOff, '');
end;

function TDXClusterClient.Running: Boolean;
begin
  Result := FThread <> nil;
end;

procedure TDXClusterClient.PublishSock(S: TSocket);
// Зеркало хэндла для Stop. Поток зовёт с валидным хэндлом сразу после connect
// и с SOCK_INVALID ПЕРЕД close — так Stop под локом видит либо живой сокет,
// либо ничего, но не закрытый.
begin
  FLock.Enter;
  try
    FActiveSock := S;
  finally
    FLock.Leave;
  end;
end;

procedure TDXClusterClient.SetState(St: TDXClusterState; const Msg: string);
begin
  FLock.Enter;
  try
    FState := St;
    FStatusMsg := Msg;
    Inc(FLogVer);      // UI перечитает и статус тоже
  finally
    FLock.Leave;
  end;
end;

function TDXClusterClient.State: TDXClusterState;
begin
  FLock.Enter;
  try
    Result := FState;
  finally
    FLock.Leave;
  end;
end;

function TDXClusterClient.StateText: string;
begin
  Result := DXStateName(State);
end;

function TDXClusterClient.StatusMessage: string;
begin
  FLock.Enter;
  try
    Result := FStatusMsg;
  finally
    FLock.Leave;
  end;
end;

function TDXClusterClient.SpotsReceived: Int64;
begin
  FLock.Enter;
  try
    Result := FSpotCount;
  finally
    FLock.Leave;
  end;
end;

procedure TDXClusterClient.SendCommand(const S: string);
begin
  if Trim(S) = '' then Exit;
  FLock.Enter;
  try
    FOutQueue.Add(S);
  finally
    FLock.Leave;
  end;
end;

procedure TDXClusterClient.AddLog(const S: string);
begin
  FLock.Enter;
  try
    FLog[FLogHead] := S;
    FLogHead := (FLogHead + 1) mod DX_LOG_LINES;
    if FLogCount < DX_LOG_LINES then Inc(FLogCount);
    Inc(FLogVer);
  finally
    FLock.Leave;
  end;
end;

procedure TDXClusterClient.GetLog(Dst: TStrings);
var i, Start: Integer;
begin
  if Dst = nil then Exit;
  FLock.Enter;
  try
    Start := (FLogHead - FLogCount + DX_LOG_LINES) mod DX_LOG_LINES;
    for i := 0 to FLogCount - 1 do
      Dst.Add(FLog[(Start + i) mod DX_LOG_LINES]);
  finally
    FLock.Leave;
  end;
end;

function TDXClusterClient.LogVersion: Int64;
begin
  FLock.Enter;
  try
    Result := FLogVer;
  finally
    FLock.Leave;
  end;
end;

{ ── TDXClusterThread ─────────────────────────────────────────────────────── }

constructor TDXClusterThread.Create(AOwner: TDXClusterClient);
begin
  FOwner  := AOwner;
  FSocket := SOCK_INVALID;
  FreeOnTerminate := False;
  inherited Create(False);
end;

function TDXClusterThread.Resolve(const Host: string; out Addr: LongWord): Boolean;
// Возвращает адрес в СЕТЕВОМ порядке байт (готов для sin_addr).
{$IFNDEF MSWINDOWS}
var HE: THostEntry; HA: THostAddr;
{$ELSE}
var PH: PHostEnt; L: LongWord;
{$ENDIF}
begin
  Result := False;
  Addr := 0;
  if Host = '' then Exit;
{$IFDEF MSWINDOWS}
  L := inet_addr(PChar(Host));
  if L <> INADDR_NONE then          // литеральный IP (уже в сетевом порядке)
  begin
    Addr := L;
    Exit(True);
  end;
  PH := gethostbyname(PChar(Host));
  if (PH = nil) or (PH^.h_addr_list = nil) or (PH^.h_addr_list^ = nil) then Exit;
  Move(PH^.h_addr_list^^, Addr, 4);  // hostent отдаёт сетевой порядок
  Result := True;
{$ELSE}
  // Литеральный IP: TryStrToHostAddr отдаёт ХОСТОВЫЙ порядок — разворачиваем.
  if TryStrToHostAddr(Host, HA) then
  begin
    Addr := htonl(HA.s_addr);
    Exit(True);
  end;
  // DNS (netdb читает /etc/resolv.conf и /etc/hosts). Здесь адрес приходит
  // прямо из DNS-ответа, т.е. УЖЕ в сетевом порядке — второй раз не вертим.
  if ResolveHostByName(Host, HE) then
  begin
    Addr := HE.Addr.s_addr;
    Exit(True);
  end;
{$ENDIF}
end;

function TDXClusterThread.ConnectSock: Boolean;
// Неблокирующий connect + select: мёртвый хост не держит поток минуту на
// системном таймауте TCP, и Stop() отрабатывает быстро.
var
  Addr: {$IFDEF MSWINDOWS}TSockAddrIn{$ELSE}TInetSockAddr{$ENDIF};
  IP: LongWord;
  R, SoErr: Integer;
  FDS: TFDSet;
  TV: TTimeVal;
  One, Waited: Integer;
  {$IFDEF MSWINDOWS}ErrLen: Integer;{$ELSE}ErrLen: TSockLen;{$ENDIF}
begin
  Result := False;
  if not Resolve(FHost, IP) then
  begin
    FOwner.SetState(dxsRetry, 'cannot resolve ' + FHost);
    FOwner.AddLog('*** DNS: cannot resolve ' + FHost);
    Exit;
  end;

  {$IFDEF MSWINDOWS}
  FSocket := socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
  {$ELSE}
  FSocket := fpSocket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
  {$ENDIF}
  if FSocket = SOCK_INVALID then
  begin
    FOwner.SetState(dxsRetry, 'cannot create socket');
    Exit;
  end;

  FillChar(Addr, SizeOf(Addr), 0);
  Addr.sin_family := AF_INET;
  Addr.sin_port   := htons(FPort);
  Addr.sin_addr.s_addr := IP;

  SockSetNonBlock(FSocket, True);
  {$IFDEF MSWINDOWS}
  R := WinSock2.connect(FSocket, @Addr, SizeOf(Addr));
  {$ELSE}
  R := fpConnect(FSocket, @Addr, SizeOf(Addr));
  {$ENDIF}
  if R <> 0 then
  begin
    // ожидаемо: EINPROGRESS / WSAEWOULDBLOCK — ждём готовности на запись.
    // Ждём НЕ одним select на 8 с, а квантами по DX_CONNECT_SLICE_MS: иначе
    // Stop() (закрытие программы) висел бы до конца таймаута соединения.
    Waited := 0;
    R := 0;
    while (Waited < DX_CONNECT_MS) and (not Terminated) do
    begin
      TV.tv_sec  := DX_CONNECT_SLICE_MS div 1000;
      TV.tv_usec := (DX_CONNECT_SLICE_MS mod 1000) * 1000;
      {$IFDEF MSWINDOWS}
      FD_ZERO(FDS);
      FD_SET(FSocket, FDS);
      R := select(0, nil, @FDS, nil, @TV);
      {$ELSE}
      fpFD_ZERO(FDS);
      fpFD_SET(FSocket, FDS);
      R := fpSelect(FSocket + 1, nil, @FDS, nil, @TV);
      {$ENDIF}
      if R <> 0 then Break;                    // готов или ошибка select
      Inc(Waited, DX_CONNECT_SLICE_MS);
    end;
    if Terminated then
    begin
      CloseSock;
      Exit;
    end;
    if R <= 0 then
    begin
      CloseSock;
      FOwner.SetState(dxsRetry, 'connect timeout: ' + FHost);
      FOwner.AddLog('*** no answer from ' + FHost + ':' + IntToStr(FPort));
      Exit;
    end;
    SoErr := 0; ErrLen := SizeOf(SoErr);
    {$IFDEF MSWINDOWS}
    getsockopt(FSocket, SOL_SOCKET, SO_ERROR, PChar(@SoErr), ErrLen);
    {$ELSE}
    fpGetSockOpt(FSocket, SOL_SOCKET, SO_ERROR, @SoErr, @ErrLen);
    {$ENDIF}
    if SoErr <> 0 then
    begin
      CloseSock;
      FOwner.SetState(dxsRetry, 'connection refused');
      FOwner.AddLog('*** refused: ' + FHost + ':' + IntToStr(FPort));
      Exit;
    end;
  end;

  SockSetNonBlock(FSocket, False);
  SockSetRcvTimeout(FSocket, DX_RECV_MS);
  SockSetSndTimeout(FSocket, 5000);
  One := 1;
  {$IFDEF MSWINDOWS}
  setsockopt(FSocket, SOL_SOCKET, SO_KEEPALIVE, PChar(@One), SizeOf(One));
  {$ELSE}
  fpSetSockOpt(FSocket, SOL_SOCKET, SO_KEEPALIVE, @One, SizeOf(One));
  {$ENDIF}
  FOwner.PublishSock(FSocket);   // с этого момента Stop может рвать нас shutdown'ом
  Result := True;
end;

procedure TDXClusterThread.CloseSock;
begin
  if FSocket <> SOCK_INVALID then
  begin
    FOwner.PublishSock(SOCK_INVALID);   // до close: Stop не должен увидеть мёртвый хэндл
    SockShutdown(FSocket);
    SockClose(FSocket);
    FSocket := SOCK_INVALID;
  end;
end;

function TDXClusterThread.SendAll(const Data: string): Boolean;
// TCP вправе отправить меньше запрошенного — дописываем остаток. Таймаут
// (SO_SNDTIMEO) и EINTR — не ошибка соединения, пробуем ещё, но не бесконечно.
var
  Sent, N, Tries: Integer;
begin
  Result := False;
  if (FSocket = SOCK_INVALID) or (Data = '') then Exit;
  Sent  := 0;
  Tries := 0;
  while Sent < Length(Data) do
  begin
    if Terminated then Exit;
    N := SockSend(FSocket, @Data[Sent + 1], Length(Data) - Sent, 0);
    if N > 0 then
    begin
      Inc(Sent, N);
      Tries := 0;
    end
    else
    begin
      if not SockWouldBlock then Exit;      // настоящая ошибка сокета
      Inc(Tries);
      if Tries > DX_SEND_RETRIES then Exit; // не отдаёт буфер — считаем разрывом
    end;
  end;
  Result := True;
end;

function TDXClusterThread.SendLine(const S: string): Boolean;
begin
  Result := SendAll(S + #13#10);
  if not Result then
    FOwner.AddLog('*** send failed, dropping connection');
end;

procedure TDXClusterThread.FlushOutQueue;
var
  Pending: TStringList;
  i: Integer;
begin
  Pending := nil;
  FOwner.FLock.Enter;
  try
    if FOwner.FOutQueue.Count > 0 then
    begin
      Pending := TStringList.Create;
      Pending.Assign(FOwner.FOutQueue);
      FOwner.FOutQueue.Clear;
    end;
  finally
    FOwner.FLock.Leave;
  end;
  if Pending = nil then Exit;
  try
    for i := 0 to Pending.Count - 1 do
    begin
      if not SendLine(Pending[i]) then Break;   // сокет умер — остальное на реконнект
      FOwner.AddLog('> ' + Pending[i]);
    end;
  finally
    Pending.Free;
  end;
end;

procedure TDXClusterThread.CheckPrompts(const Tail: string);
// Приглашения логина И пароля обычно приходят БЕЗ перевода строки, поэтому
// ищем их и в незавершённом хвосте буфера, а не только в целых строках.
// Флаги FLoginSent/FPassSent держат отправку однократной: хвост проверяется
// заново с каждым пришедшим куском.
var U: string;
begin
  if Tail = '' then Exit;
  U := LowerCase(Tail);

  if not FLoginSent then
  begin
    if (Pos('login:', U) > 0) or (Pos('call:', U) > 0) or
       (Pos('callsign', U) > 0) or (Pos('your call', U) > 0) or
       (Pos('enter your', U) > 0) then
    begin
      if not SendLine(FLogin) then Exit;
      FOwner.AddLog('> ' + FLogin);
      FLoginSent := True;
      FLoginAt   := Now;
      FOwner.SetState(dxsLogin, '');
    end;
    Exit;
  end;

  if (FPassword <> '') and (not FPassSent) and (Pos('password', U) > 0) then
  begin
    if not SendLine(FPassword) then Exit;
    FOwner.AddLog('> ********');
    FPassSent := True;
  end;
end;

procedure TDXClusterThread.HandleLine(const Line: string);
var
  Spot: TDXSpot;
begin
  if Line = '' then Exit;
  FOwner.AddLog(Line);

  if ParseDXSpot(Line, Spot) then
  begin
    FOwner.FStore.Add(Spot);
    FOwner.FLock.Enter;
    try
      Inc(FOwner.FSpotCount);
    finally
      FOwner.FLock.Leave;
    end;
    if FOwner.State <> dxsOnline then FOwner.SetState(dxsOnline, '');
    Exit;
  end;

  CheckPrompts(Line);
end;

procedure TDXClusterThread.HandleChunk(const Data: string);
var
  P: Integer;
  Line: string;
begin
  FRxBuf := FRxBuf + Data;
  while True do
  begin
    P := 1;
    while (P <= Length(FRxBuf)) and not (FRxBuf[P] in [#10, #13]) do Inc(P);
    if P > Length(FRxBuf) then Break;          // перевода строки ещё нет
    Line := Copy(FRxBuf, 1, P - 1);
    // съедаем весь конец строки (CR, LF или CRLF)
    while (P <= Length(FRxBuf)) and (FRxBuf[P] in [#10, #13]) do Inc(P);
    Delete(FRxBuf, 1, P - 1);
    HandleLine(TrimRight(Line));
  end;

  // Хвост без перевода строки — возможно, это приглашение логина или пароля.
  if FRxBuf <> '' then CheckPrompts(FRxBuf);
  // Защита от мусора без переводов строк.
  if Length(FRxBuf) > 8192 then FRxBuf := '';
end;

function TDXClusterThread.Session: Boolean;
var
  Buf: array[0..4095] of Byte;
  N: Integer;
  Chunk: string;
  PL: TStringList;
  i: Integer;
begin
  Result := True;
  FRxBuf := '';
  FLoginSent := False;
  FPassSent  := False;
  FPostSent  := False;
  FConnectAt := Now;

  FOwner.SetState(dxsConnecting, FHost + ':' + IntToStr(FPort));
  FOwner.AddLog('*** connecting to ' + FHost + ':' + IntToStr(FPort));
  if not ConnectSock then Exit;

  FOwner.AddLog('*** connected');
  FOwner.SetState(dxsLogin, '');

  while not Terminated do
  begin
    N := SockRecv(FSocket, @Buf[0], SizeOf(Buf), 0);
    if N > 0 then
    begin
      SetLength(Chunk, N);
      Move(Buf[0], Chunk[1], N);
      HandleChunk(Chunk);
    end
    else if N = 0 then
    begin
      FOwner.AddLog('*** connection closed by peer');
      Break;
    end
    else if not SockWouldBlock then
    begin
      // Не таймаут кванта recv, а настоящая ошибка сокета — уходим на реконнект.
      FOwner.AddLog('*** connection lost (socket error)');
      Break;
    end;

    if Terminated then Break;

    // Приглашения не дождались — шлём позывной сами (часть кластеров молчит).
    if (not FLoginSent) and (SecondsBetween(Now, FConnectAt) >= DX_LOGIN_GRACE_S) then
    begin
      if not SendLine(FLogin) then Break;   // сокет умер — на реконнект
      FOwner.AddLog('> ' + FLogin + '   (no prompt seen)');
      FLoginSent := True;
      FLoginAt   := Now;
    end;

    // Post-login команды — один раз, чуть погодя после позывного.
    if FLoginSent and (not FPostSent) and
       (SecondsBetween(Now, FLoginAt) >= DX_POSTLOGIN_S) then
    begin
      FPostSent := True;
      if Trim(FPostLogin) <> '' then
      begin
        PL := TStringList.Create;
        try
          PL.Text := FPostLogin;
          for i := 0 to PL.Count - 1 do
            if Trim(PL[i]) <> '' then
            begin
              if not SendLine(Trim(PL[i])) then Break;
              FOwner.AddLog('> ' + Trim(PL[i]));
            end;
        finally
          PL.Free;
        end;
      end;
      if FOwner.State <> dxsOnline then FOwner.SetState(dxsOnline, '');
    end;

    FlushOutQueue;
  end;

  CloseSock;
  Result := not Terminated;
end;

procedure TDXClusterThread.Execute;
var
  Backoff: Integer;
begin
  Backoff := DX_RECONNECT_MIN;
  // Снимок конфига на всю жизнь потока: Configure во время работы применяется
  // следующим Start (так же ведёт себя веб-сервер при смене порта).
  FOwner.FLock.Enter;
  try
    FHost      := FOwner.FHost;
    FPort      := FOwner.FPort;
    FLogin     := FOwner.FLogin;
    FPassword  := FOwner.FPassword;
    FPostLogin := FOwner.FPostLogin;
  finally
    FOwner.FLock.Leave;
  end;

  while not Terminated do
  begin
    if not Session then Break;
    if Terminated then Break;

    FOwner.SetState(dxsRetry, 'reconnecting in ' + IntToStr(Backoff) + ' s');
    FOwner.AddLog('*** reconnecting in ' + IntToStr(Backoff) + ' s');
    if FOwner.FStopEvent.WaitFor(Backoff * 1000) = wrSignaled then Break;
    Backoff := Backoff * 2;
    if Backoff > DX_RECONNECT_MAX then Backoff := DX_RECONNECT_MAX;
  end;
  CloseSock;
end;

end.