unit UMQTTClient;

interface

uses
  System.SysUtils, System.Classes, System.SyncObjs, System.DateUtils,
  System.Generics.Collections,
  IdTCPClient, IdGlobal, IdExceptionCore, IdException, IdIOHandler;

type
  TMQTTMessageEvent = procedure(const ATopic, APayload: string) of object;
  TMQTTStatusEvent = procedure(AConnected: Boolean) of object;

  TMQTTRecvThread = class;

  TMQTTClient = class
  private
    FHost: string;
    FPort: Word;
    FClientID: string;
    FTCPClient: TIdTCPClient;
    FRecvThread: TMQTTRecvThread;
    FOnMessage: TMQTTMessageEvent;
    FOnStatus: TMQTTStatusEvent;
    FConnected: Boolean;
    FSubscribeTopic: string;
    FKeepAlive: Word;
    FPacketID: Word;
    FActive: Boolean;

    FSendQueue: TList<TIdBytes>;
    FSendLock: TCriticalSection;

    function NextPacketID: Word;
    function BuildConnectPacket: TIdBytes;
    function BuildSubscribePacket(const ATopic: string): TIdBytes;
    function BuildPingReqPacket: TIdBytes;
    function BuildDisconnectPacket: TIdBytes;
    function BuildPublishPacket(const ATopic, APayload: string): TIdBytes;
    function EncodeRemainingLength(ALength: Integer): TIdBytes;
    procedure SetConnected(AValue: Boolean);
    procedure FireMessage(const ATopic, APayload: string);
    procedure EnqueuePacket(const AData: TIdBytes);
  public
    constructor Create(const AHost: string; APort: Word; const AClientID: string = '');
    destructor Destroy; override;

    procedure Connect;
    procedure Disconnect;
    procedure Subscribe(const ATopic: string);
    procedure Publish(const ATopic, APayload: string);

    property Host: string read FHost write FHost;
    property Port: Word read FPort write FPort;
    property Connected: Boolean read FConnected;
    property OnMessage: TMQTTMessageEvent read FOnMessage write FOnMessage;
    property OnStatus: TMQTTStatusEvent read FOnStatus write FOnStatus;
  end;

  TMQTTRecvThread = class(TThread)
  private
    FOwner: TMQTTClient;
    function ReadRemainingLength: Integer;
    procedure ProcessPublish(ARemainingLen: Integer);
    function FlushSendQueue: Boolean;
  protected
    procedure Execute; override;
  public
    constructor Create(AOwner: TMQTTClient);
  end;

implementation

{ TMQTTClient }

constructor TMQTTClient.Create(const AHost: string; APort: Word; const AClientID: string);
begin
  inherited Create;
  FHost := AHost;
  FPort := APort;
  Randomize;
  if AClientID <> '' then
    FClientID := AClientID + IntToStr(Random(99999))
  else
    FClientID := 'SmartFarmHMI_' + IntToStr(Random(99999));
  FKeepAlive := 60;
  FPacketID := 0;
  FConnected := False;
  FActive := False;
  FSubscribeTopic := '';

  FSendQueue := TList<TIdBytes>.Create;
  FSendLock := TCriticalSection.Create;

  FTCPClient := TIdTCPClient.Create(nil);
  FTCPClient.ConnectTimeout := 5000;
  // ReadTimeout: 읽기 대기 시간 (500ms 폴링 주기)
  FTCPClient.ReadTimeout := 500;
end;

destructor TMQTTClient.Destroy;
begin
  Disconnect;
  FTCPClient.Free;
  FSendQueue.Free;
  FSendLock.Free;
  inherited;
end;

function TMQTTClient.NextPacketID: Word;
begin
  Inc(FPacketID);
  if FPacketID = 0 then
    FPacketID := 1;
  Result := FPacketID;
end;

function TMQTTClient.EncodeRemainingLength(ALength: Integer): TIdBytes;
var
  EncodedByte: Byte;
  Len: Integer;
begin
  SetLength(Result, 0);
  Len := ALength;
  repeat
    EncodedByte := Len mod 128;
    Len := Len div 128;
    if Len > 0 then
      EncodedByte := EncodedByte or $80;
    SetLength(Result, Length(Result) + 1);
    Result[High(Result)] := EncodedByte;
  until Len = 0;
end;

function TMQTTClient.BuildConnectPacket: TIdBytes;
var
  VarHeader, Payload, Remaining, Packet: TIdBytes;
  ClientIDBytes: TIdBytes;
  RemLen: Integer;
begin
  SetLength(VarHeader, 10);
  VarHeader[0] := 0;
  VarHeader[1] := 4;
  VarHeader[2] := Ord('M');
  VarHeader[3] := Ord('Q');
  VarHeader[4] := Ord('T');
  VarHeader[5] := Ord('T');
  VarHeader[6] := 4;
  VarHeader[7] := 2;
  VarHeader[8] := Hi(FKeepAlive);
  VarHeader[9] := Lo(FKeepAlive);

  ClientIDBytes := ToBytes(FClientID, IndyTextEncoding_UTF8);
  SetLength(Payload, 2 + Length(ClientIDBytes));
  Payload[0] := Hi(Word(Length(ClientIDBytes)));
  Payload[1] := Lo(Word(Length(ClientIDBytes)));
  if Length(ClientIDBytes) > 0 then
    CopyTIdBytes(ClientIDBytes, 0, Payload, 2, Length(ClientIDBytes));

  RemLen := Length(VarHeader) + Length(Payload);
  Remaining := EncodeRemainingLength(RemLen);

  SetLength(Packet, 1 + Length(Remaining) + RemLen);
  Packet[0] := $10;
  CopyTIdBytes(Remaining, 0, Packet, 1, Length(Remaining));
  CopyTIdBytes(VarHeader, 0, Packet, 1 + Length(Remaining), Length(VarHeader));
  CopyTIdBytes(Payload, 0, Packet, 1 + Length(Remaining) + Length(VarHeader), Length(Payload));

  Result := Packet;
end;

function TMQTTClient.BuildSubscribePacket(const ATopic: string): TIdBytes;
var
  TopicBytes, VarHeader, Payload, Remaining, Packet: TIdBytes;
  PID: Word;
  RemLen: Integer;
begin
  PID := NextPacketID;

  SetLength(VarHeader, 2);
  VarHeader[0] := Hi(PID);
  VarHeader[1] := Lo(PID);

  TopicBytes := ToBytes(ATopic, IndyTextEncoding_UTF8);
  SetLength(Payload, 2 + Length(TopicBytes) + 1);
  Payload[0] := Hi(Word(Length(TopicBytes)));
  Payload[1] := Lo(Word(Length(TopicBytes)));
  CopyTIdBytes(TopicBytes, 0, Payload, 2, Length(TopicBytes));
  Payload[2 + Length(TopicBytes)] := 0;

  RemLen := Length(VarHeader) + Length(Payload);
  Remaining := EncodeRemainingLength(RemLen);

  SetLength(Packet, 1 + Length(Remaining) + RemLen);
  Packet[0] := $82;
  CopyTIdBytes(Remaining, 0, Packet, 1, Length(Remaining));
  CopyTIdBytes(VarHeader, 0, Packet, 1 + Length(Remaining), Length(VarHeader));
  CopyTIdBytes(Payload, 0, Packet, 1 + Length(Remaining) + Length(VarHeader), Length(Payload));

  Result := Packet;
end;

function TMQTTClient.BuildPingReqPacket: TIdBytes;
begin
  SetLength(Result, 2);
  Result[0] := $C0;
  Result[1] := $00;
end;

function TMQTTClient.BuildDisconnectPacket: TIdBytes;
begin
  SetLength(Result, 2);
  Result[0] := $E0;
  Result[1] := $00;
end;

function TMQTTClient.BuildPublishPacket(const ATopic, APayload: string): TIdBytes;
var
  TopicBytes, PayloadBytes, Remaining, Packet: TIdBytes;
  RemLen, Offset: Integer;
begin
  TopicBytes := ToBytes(ATopic, IndyTextEncoding_UTF8);
  PayloadBytes := ToBytes(APayload, IndyTextEncoding_UTF8);

  RemLen := 2 + Length(TopicBytes) + Length(PayloadBytes);
  Remaining := EncodeRemainingLength(RemLen);

  SetLength(Packet, 1 + Length(Remaining) + RemLen);
  Packet[0] := $30;
  Offset := 1;
  CopyTIdBytes(Remaining, 0, Packet, Offset, Length(Remaining));
  Inc(Offset, Length(Remaining));
  Packet[Offset] := Hi(Word(Length(TopicBytes)));
  Packet[Offset + 1] := Lo(Word(Length(TopicBytes)));
  Inc(Offset, 2);
  CopyTIdBytes(TopicBytes, 0, Packet, Offset, Length(TopicBytes));
  Inc(Offset, Length(TopicBytes));
  if Length(PayloadBytes) > 0 then
    CopyTIdBytes(PayloadBytes, 0, Packet, Offset, Length(PayloadBytes));

  Result := Packet;
end;

procedure TMQTTClient.EnqueuePacket(const AData: TIdBytes);
begin
  FSendLock.Enter;
  try
    FSendQueue.Add(AData);
  finally
    FSendLock.Leave;
  end;
end;

procedure TMQTTClient.SetConnected(AValue: Boolean);
begin
  // 상태가 실제로 변경될 때만 이벤트 발생 (UI 플리커 방지)
  if FConnected = AValue then
    Exit;

  FConnected := AValue;
  if Assigned(FOnStatus) then
  begin
    TThread.Queue(nil,
      procedure
      begin
        if Assigned(FOnStatus) then
          FOnStatus(AValue);
      end);
  end;
end;

procedure TMQTTClient.FireMessage(const ATopic, APayload: string);
var
  LTopic, LPayload: string;
begin
  LTopic := ATopic;
  LPayload := APayload;
  if Assigned(FOnMessage) then
  begin
    TThread.Queue(nil,
      procedure
      begin
        if Assigned(FOnMessage) then
          FOnMessage(LTopic, LPayload);
      end);
  end;
end;

procedure TMQTTClient.Connect;
begin
  if FActive then
    Exit;
  FActive := True;
  FRecvThread := TMQTTRecvThread.Create(Self);
end;

procedure TMQTTClient.Disconnect;
begin
  FActive := False;

  if FRecvThread <> nil then
  begin
    FRecvThread.Terminate;
    FRecvThread.WaitFor;
    FreeAndNil(FRecvThread);
  end;

  try
    if (FTCPClient.IOHandler <> nil) and (not FTCPClient.IOHandler.ClosedGracefully) then
      FTCPClient.IOHandler.Write(BuildDisconnectPacket);
  except
  end;

  try
    FTCPClient.Disconnect;
  except
  end;

  FConnected := False;
  if Assigned(FOnStatus) then
    FOnStatus(False);
end;

procedure TMQTTClient.Subscribe(const ATopic: string);
begin
  FSubscribeTopic := ATopic;
  if FConnected then
    EnqueuePacket(BuildSubscribePacket(ATopic));
end;

procedure TMQTTClient.Publish(const ATopic, APayload: string);
begin
  if FConnected then
    EnqueuePacket(BuildPublishPacket(ATopic, APayload));
end;

{ TMQTTRecvThread }

constructor TMQTTRecvThread.Create(AOwner: TMQTTClient);
begin
  inherited Create(False);
  FreeOnTerminate := False;
  FOwner := AOwner;
end;

function TMQTTRecvThread.ReadRemainingLength: Integer;
var
  Multiplier: Integer;
  EncodedByte: Byte;
begin
  Result := 0;
  Multiplier := 1;
  repeat
    EncodedByte := FOwner.FTCPClient.IOHandler.ReadByte;
    Result := Result + (EncodedByte and $7F) * Multiplier;
    Multiplier := Multiplier * 128;
  until (EncodedByte and $80) = 0;
end;

procedure TMQTTRecvThread.ProcessPublish(ARemainingLen: Integer);
var
  TopicLen: Word;
  TopicBytes, PayloadBytes: TIdBytes;
  Topic, Payload: string;
  PayloadLen: Integer;
begin
  TopicLen := FOwner.FTCPClient.IOHandler.ReadByte shl 8;
  TopicLen := TopicLen or FOwner.FTCPClient.IOHandler.ReadByte;

  FOwner.FTCPClient.IOHandler.ReadBytes(TopicBytes, TopicLen);
  Topic := BytesToString(TopicBytes, IndyTextEncoding_UTF8);

  PayloadLen := ARemainingLen - 2 - TopicLen;
  if PayloadLen > 0 then
  begin
    FOwner.FTCPClient.IOHandler.ReadBytes(PayloadBytes, PayloadLen);
    Payload := BytesToString(PayloadBytes, IndyTextEncoding_UTF8);
  end
  else
    Payload := '';

  FOwner.FireMessage(Topic, Payload);
end;

function TMQTTRecvThread.FlushSendQueue: Boolean;
var
  Packets: TList<TIdBytes>;
  Pkt: TIdBytes;
begin
  Result := True;
  Packets := nil;

  FOwner.FSendLock.Enter;
  try
    if FOwner.FSendQueue.Count > 0 then
    begin
      Packets := TList<TIdBytes>.Create;
      Packets.AddRange(FOwner.FSendQueue);
      FOwner.FSendQueue.Clear;
    end;
  finally
    FOwner.FSendLock.Leave;
  end;

  if Packets <> nil then
  begin
    try
      for Pkt in Packets do
      begin
        try
          FOwner.FTCPClient.IOHandler.Write(Pkt);
        except
          Result := False;
          Break;
        end;
      end;
    finally
      Packets.Free;
    end;
  end;
end;

procedure TMQTTRecvThread.Execute;
var
  ConnAckBuf: TIdBytes;
  PacketType: Byte;
  RemainingLen: Integer;
  DiscardBuf: TIdBytes;
  LastPingTime: TDateTime;
  PingIntervalSec: Integer;
  GotData: Boolean;
begin
  PingIntervalSec := FOwner.FKeepAlive div 2;
  if PingIntervalSec < 10 then
    PingIntervalSec := 10;

  while not Terminated do
  begin
    // === Connection Phase ===
    try
      FOwner.FTCPClient.Host := FOwner.FHost;
      FOwner.FTCPClient.Port := FOwner.FPort;
      FOwner.FTCPClient.Connect;

      FOwner.FTCPClient.IOHandler.ReadTimeout := 500; // 읽기 타임아웃 재설정

      // CONNECT 패킷 전송
      FOwner.FTCPClient.IOHandler.Write(FOwner.BuildConnectPacket);

      // 무한대기 읽기로 CONNACK 수신 (접속시에만)
      FOwner.FTCPClient.IOHandler.ReadTimeout := 10000;
      FOwner.FTCPClient.IOHandler.ReadBytes(ConnAckBuf, 4, False);
      FOwner.FTCPClient.IOHandler.ReadTimeout := 500;

      if (Length(ConnAckBuf) >= 4) and (ConnAckBuf[0] = $20) and (ConnAckBuf[3] = $00) then
      begin
        FOwner.SetConnected(True);
        LastPingTime := Now;

        // Subscribe
        if FOwner.FSubscribeTopic <> '' then
          FOwner.FTCPClient.IOHandler.Write(FOwner.BuildSubscribePacket(FOwner.FSubscribeTopic));
      end
      else
      begin
        FOwner.FTCPClient.Disconnect;
        FOwner.SetConnected(False);
        if not Terminated then Sleep(5000);
        Continue;
      end;
    except
      FOwner.SetConnected(False);
      try FOwner.FTCPClient.Disconnect; except end;
      if not Terminated then Sleep(5000);
      Continue;
    end;

    // === Receive Loop ===
    while not Terminated do
    begin
      // 1) 송신 큐 flush
      if not FlushSendQueue then
        Break; // 전송 실패 = 연결 끊김

      // 2) PINGREQ 주기적 전송
      if SecondsBetween(Now, LastPingTime) >= PingIntervalSec then
      begin
        LastPingTime := Now;
        try
          FOwner.FTCPClient.IOHandler.Write(FOwner.BuildPingReqPacket);
        except
          Break; // 전송 실패 = 연결 끊김
        end;
      end;

      // 3) 소켓에서 데이터 읽기 시도
      try
        if not FOwner.FTCPClient.IOHandler.CheckForDataOnSource(500) then
          Continue;

        PacketType := FOwner.FTCPClient.IOHandler.ReadByte;
      except
        on E: EIdReadTimeout do
        begin
          // 타임아웃 = 정상 (데이터 없음), 다음 루프로
          Continue;
        end;
        on E: EIdConnClosedGracefully do
        begin
          Break;
        end;
        on E: EIdException do
        begin
          Break;
        end;
        on E: Exception do
        begin
          Break;
        end;
      end;

      // 4) 패킷 나머지 읽기 (데이터가 있으므로 타임아웃 가능성 낮음)
      try
        RemainingLen := ReadRemainingLength;

        case (PacketType and $F0) of
          $30: // PUBLISH
            ProcessPublish(RemainingLen);

          $90: // SUBACK
          begin
            if RemainingLen > 0 then
              FOwner.FTCPClient.IOHandler.ReadBytes(DiscardBuf, RemainingLen);
          end;

          $D0: // PINGRESP
          begin
            // Keep-Alive 확인 완료
            if RemainingLen > 0 then
              FOwner.FTCPClient.IOHandler.ReadBytes(DiscardBuf, RemainingLen);
          end;

        else
          // 알 수 없는 패킷 — 남은 바이트 소비
          if RemainingLen > 0 then
            FOwner.FTCPClient.IOHandler.ReadBytes(DiscardBuf, RemainingLen);
        end;
      except
        // 패킷 파싱 중 오류 = 연결 끊김
        Break;
      end;
    end;

    // === 연결 끊김 처리 ===
    FOwner.SetConnected(False);
    try FOwner.FTCPClient.Disconnect; except end;

    if not Terminated then
      Sleep(5000);
  end;
end;

end.