unit UDP;

interface
uses
  Windows, forms, XMLTag, Messages, Contnrs,
  ExtCtrls, IdComponent, IdSocketHandle, IdUDPServer, SysUtils, Classes;

const
  AnyQUDP_P2P_Port = 5224;        //P2P连接端口
  AnyQUDP_SERVER_Port = 5225;     //服务器UDP连接端口
  AnyQUDP_MAXSize = 1440;         //每个UDP包的最大大小
//  AnyQUDP_TimeOut = 30;

  WM_AnyQ_STARTUDPSEND = WM_USER+5424;

type

  TAnyQUDPHead = packed record
    TaskID: TGUID;            //16 byte
    DataNo: Cardinal;         // 4 byte
    DataPackageSize: DWord;       // 4 Byte
    DataSize: Word;           // 2 byte
    DataType: Byte;           // 1 Byte   为零时，表示数据，为非零时，表示应答
  end;                        //total 27 byte

  TAnyQUDPController = class;

  TAnyQRcvDataEvent = procedure(sender: TObject; AData: TStream) of Object;

  TAnyQBaseUDPTask = class
  private
    _Controller: TAnyQUDPController;
    _Timer: TTimer;
    _Tick: Cardinal;
    _Hosts: TStringList;
    _ConnectIndex: Integer;
//    _CurHost: string;
//    _CurPort: Integer;
    _SendBuffer: TMemoryStream;
    _RcvBuffer: TMemoryStream;
    _SendDataPackageSize: Integer;
    _DataSizeQueue: TQueue;
    _RcvDataPackageSize: Integer;
    _SendBuf: array of Byte;
    _SendSize: Integer;
    _TryCount: Integer;
    _Sending: Boolean;
    _Receiving: Boolean;

    FOnRcvData: TAnyQRcvDataEvent;
    FOnRcvStatus: TNotifyEvent;
    FOnTimeOut: TNotifyEvent;
    FOnFail: TNotifyEvent;
    procedure Sendnext;
    procedure ProcRcvData(AData: TStream);
    procedure Timeout(sender: TObject);
  protected
    SendNo: cardinal;
    RcvNo: Cardinal;

    procedure SendRawData(var Data; Size: Integer);
    procedure Resend;
    procedure RcvData(AData: TStream; ABinding: TIdSocketHandle); virtual;
  public
    _CurHost: string;
    _CurPort: Integer;

    TaskID: TGUID;
    JID: widestring;
    Flag: string;
    Status: string;
    PackageSize: Integer;
    NoCheck: Boolean;

    constructor Create(Controller: TAnyQUDPController; GUID: TGUID); virtual;
    destructor Destroy; override;

    procedure Start; virtual; abstract;
    procedure Continue; virtual; abstract;
    procedure Connect; overload;
    procedure Connect(hosts: Tstrings); overload;
    procedure SendData(const Data; Size: Integer; EndData: Boolean = True);
    procedure SendEndFlag;
    property OnRcvData: TAnyQRcvDataEvent read FOnRcvData write FOnRcvData;
    property OnRcvStatus: TNotifyEvent read FOnRcvStatus write FOnRcvStatus;
    property OnTimeOut: TNotifyEvent read FOnTimeOut write FOnTimeOut;
    property OnFail: TNotifyEvent read FOnFail write FOnFail;
    property Sending: Boolean read _Sending;
    property Receiving: Boolean read _Receiving;
  published

  end;

{
  TAnyQUDPTask = class
  private
    _Controller: TAnyQUDPController;
    _Timer: TTimer;
    _Tick: Cardinal;
    FOnRcvData: TAnyQRcvDataEvent;
    FOnConnected: TNotifyEvent;
    FOnConnectFail: TNotifyEvent;
    FOnSendTimeOut: TNotifyEvent;
    FOnRcvTimeOut: TNotifyEvent;
  protected
    SendNo: cardinal;
    RcvNo: Cardinal;
    Synchronized: Boolean;
    _buffer: array of Byte;
    _bufferAlloc: Integer;
    procedure SendSpecData(var Data; Size: Integer);
    procedure RcvTimeout(sender: TObject);
  public
    TaskID: TGUID;
    JID: widestring;
    DestHost: string;
    DestPort: Integer;
    SyncDataSend: Boolean;      //是否同步数据发送，音视频数据不需要同步，文件发送等需要。
    Connected: Boolean;
    ISAV: Boolean;
    UseThread: Boolean;

    constructor Create(Controller: TAnyQUDPController; GUID: TGUID); virtual;
    destructor Destroy; override;
    function Connect(hosts: Tstringlist): Boolean;
    procedure SendData(const Data; Size: Integer);
    procedure SendDataBuffered(const Data; Size: Integer; EndBuffer: Boolean = false);
    procedure SendEndFlag;

    procedure RcvData(AData: TStream; ABinding: TIdSocketHandle);
    property OnRcvData: TAnyQRcvDataEvent read FOnRcvData write FOnRcvData;
    property OnConnected: TNotifyEvent read FOnConnected write FOnConnected;
    property OnConnectFail: TNotifyEvent read FOnConnectFail write FOnConnectFail;
    property OnSendTimeOut: TNotifyEvent read FOnSendTimeOut write FOnSendTimeOut;
    property OnRcvTimeOut: TNotifyEvent read FOnRcvTimeOut write FOnRcvTimeOut;
  published

  end;
{}

  TAnyQUDPController = class
  private
    _server: TIdUDPServer;
    _tasklist: Tlist;
    _WANHost: string;
    _WANPort: Integer;
    _UDPInfoGetted: Boolean;
    _UDPInfoGetting: Boolean;
    _Timer: TTimer;
    procedure LiveTimeout(Sender: TObject);
  protected
    procedure AnyQUDPRead(Sender: TObject; AData: TStream;
      ABinding: TIdSocketHandle);
  public
    constructor Create(Port: Integer=AnyQUDP_P2P_Port);
    destructor Destroy; override;
    function FindTask(taskstr: string): TAnyQBaseUDPTask; overload;
    function FindTask(taskid: TGUID): TAnyQBaseUDPTask; overload;
    procedure Addtask(task: TAnyQBaseUDPTask);
    procedure Deltask(task: TAnyQBaseUDPTask);
    procedure GetUDPInfo(InfoList: TstringList; usethread: boolean = false);
    procedure SendData(host: string; port: Integer; var Data; Size: Integer);
  published

  end;

var
  UDPController: TAnyQUDPController;
  AUDPController: TAnyQUDPController;
  VUDPController: TAnyQUDPController;

procedure AnyQUdpReceive(tag: TXMLTag);

implementation

uses Session, ExSession, Transfer, InstantMail, whiteboard, ChatWin,
  TypInfo, math;

procedure Sendreject(tojid, id: string);
var
  iq: TXMLTag;
begin
  iq := TXMLTag.Create('iq');
  with iq do begin
    setAttribute('to', tojid);
    setAttribute('id', id);
    setAttribute('type', 'error');
    AddBasicTag('error', '用户拒绝请求').setAttribute('code', '406');
  end;
  MainSession.SendTag(iq);
end;

procedure TaskSendUnShare(tojid, id: string);
var
  iq: TXMLTag;
begin
  iq := TXMLTag.Create('iq');
  with iq do begin
    setAttribute('to', tojid);
    setAttribute('id', id);
    setAttribute('type', 'error');
    AddBasicTag('error', '您没有获取该用户共享文件的权限');
  end;
  MainSession.SendTag(iq);
end;

procedure AnyQUdpReceive(tag: TXMLTag);
var
  qTag, tmp_tag: TXMLTag;
  from, desc, ID: string;
  taskguid: TGuid;
  task: TAnyQBaseUDPTask;
  hosts: TStringList;
  childlist: TXMLTagList;
  i: integer;
  filename: string;
  filesize: Int64;
begin
  // Callback for receiving file transfers
  from := tag.GetAttribute('from');
  ID := tag.GetAttribute('id');
  qTag := tag.getFirstTag('query');
  if not qtag.TagExists('taskid') then exit;

  hosts := TStringList.Create;
  childlist := qtag.ChildTags;
  for i:= 0 to childlist.Count-1 do
    if childlist.Tags[i].Name = 'host' then
      hosts.Add(childlist.Tags[i].Data);
  childlist.Free;

  taskguid := StringToGUID(qtag.GetBasicText('taskid'));
  if qtag.TagExists('close') then  //对方拒绝请求
  begin
    for i:= Application.ComponentCount-1 downto 0 do
      if Application.Components[i] is TfrmTransfer then
        if IsEqualGUID(TfrmTransfer(Application.Components[i]).TaskID, taskguid) then
        begin
          TfrmTransfer(Application.Components[i]).CancleTask;
          Exit;
        end;

  end;

  task := UDPController.FindTask(taskguid);
  if task<>nil then   //任务已经存在，说明这可能是对方连接不上，请求反向连接
  begin
    task._Hosts.Assign(hosts);
    hosts.Free;
    task._ConnectIndex := 0;
    task.Connect;
    exit;
  end;

  desc := qTag.GetBasicText('desc');
  if not qtag.TagExists('type') then exit;


  task := TAnyQBaseUDPTask.Create(UDPController, taskguid);
  if qTag.GetBasicText('type')='file' then
  begin
    tmp_tag :=  qTag.GetFirstTag('desc');
    filename := tmp_tag.GetAttribute('filename');
    filesize := StrToInt64(tmp_tag.GetAttribute('filesize'));
    FileRcv(task, from, desc, filename, filesize, hosts);
    exit;
  end;

  if qTag.GetBasicText('type')='instantmail' then
  begin
    tmp_tag :=  qTag.GetFirstTag('desc');
    filename := tmp_tag.GetAttribute('filename');
    filesize := StrToInt64(tmp_tag.GetAttribute('filesize'));
    InstantmailReceive(task, from, desc, filename, filesize, hosts);
    exit;
  end;

  if qTag.GetBasicText('type')='powerpoint' then
  begin
    tmp_tag :=  qTag.GetFirstTag('desc');
    filename := tmp_tag.GetAttribute('filename');
    filesize := StrToInt64(tmp_tag.GetAttribute('filesize'));
    if MessageBox(Application.handle, pchar(from + '请求与您进行PowerPoint演讲，接受吗？'), '请求', MB_YESNO or MB_ICONQUESTION or MB_TOPMOST) = IDYes then
      StartpptChat(task, from, ID, hosts, fileName, fileSize)
    else
    begin
      Sendreject(from, id);
      FreeandNil(task);
    end;
    exit;
  end;


  if qTag.GetBasicText('type')='sharelist' then
  begin
    if MainSession.IsUnShared(from) then
    begin
      TaskSendUnshare(from, id);
      FreeandNil(task);
      exit;
    end;
    ShareListSend(task, from, ID, hosts);
    exit;
  end;
  if qTag.GetBasicText('type')='sharefile' then
  begin
    if MainSession.IsUnShared(from) then
    begin
      TaskSendUnshare(from, id);
      FreeandNil(task);
      exit;
    end;
    tmp_tag :=  qTag.GetFirstTag('desc');
    filename := tmp_tag.GetAttribute('filename');
    if MessageBox(Application.handle, pchar(from + '请求下载您共享的“'+filename+'”文件，同意吗？'), '请求同意', MB_YESNO or MB_ICONQUESTION or MB_TOPMOST) = IDYes then
      ShareFileSend(task, from, ID, desc, filename, hosts)
    else
    begin
      Sendreject(from, id);
      FreeandNil(task);
    end;
    exit;
  end;
  if qTag.GetBasicText('type')='audio' then
  begin
    if MessageBox(Application.handle, pchar(from + '请求与您进行语音对话，接受吗？'), '请求', MB_YESNO or MB_ICONQUESTION or MB_TOPMOST) = IDYes then
      StartAudioChat(task, from, ID, hosts)
    else
    begin
      Sendreject(from, id);
      FreeandNil(task);
    end;
    exit;
  end;
  if qTag.GetBasicText('type')='video' then
  begin
    if MessageBox(Application.handle, pchar(from + '请求与您进行视频对话，接受吗？'), '请求', MB_YESNO or MB_ICONQUESTION or MB_TOPMOST) = IDYes then
      StartVideoChat(task, from, ID, hosts)
    else
    begin
      Sendreject(from, id);
      FreeandNil(task);
    end;
    exit;
  end;
  if qTag.GetBasicText('type')='whiteboard' then
  begin
    if MessageBox(Application.handle, pchar(from + '请求与您进行白板交流，接受吗？'), '请求', MB_YESNO or MB_ICONQUESTION or MB_TOPMOST) = IDYes then
      StartWhiteboard(task, from, ID, hosts)
    else
    begin
      Sendreject(from, id);
      FreeandNil(task);
    end;
    exit;
  end; {}
end;

{ TAnyQUDPController }

procedure TAnyQUDPController.Addtask(task: TAnyQBaseUDPTask);
begin
  _tasklist.Add(task);
end;

constructor TAnyQUDPController.Create(Port: Integer);
//var
//  port: Integer;
begin
  _server := TIdUDPServer.Create(nil);
//  _server.DefaultPort := 0;
//  port := AnyQUDP_P2P_Port;
  _server.DefaultPort := port;
  _server.OnUDPRead := AnyQUDPRead;

//{$IFDEF AnyQDEBUG}
  while not _server.Active do
//{$ENDIF}
    try
      _server.Active := true;
//      _server.BeginWork(wmRead);
    except
//{$IFDEF AnyQDEBUG}
      Inc(port);
      _server.Free;
      _server := TIdUDPServer.Create(nil);
      _server.DefaultPort := port;
      _server.OnUDPRead := AnyQUDPRead;
//{$ENDIF}
    end;
    
  _UDPInfoGetted := False;
  _UDPInfoGetting := False;
  _tasklist := TList.Create;
  _Timer := TTimer.Create(nil);
  _Timer.Enabled := false;
  //过期时间为1分钟，1分钟后认为获得的WANPort失效，需要重新获取.
  //因为有一些服务器一定时间内没有连接的话会释放端口重新分配。
  _Timer.Interval := 30000;
  _Timer.OnTimer := LiveTimeout;
  _Timer.Enabled := True;
end;

procedure TAnyQUDPController.Deltask(task: TAnyQBaseUDPTask);
var
  index: Integer;
  i: Integer;
begin
  index := _tasklist.IndexOf(task);
  if index>=0 then
  begin
    _tasklist.Delete(index);
//    FreeAndNil(task);
  end;
  for i:= _tasklist.Count-1 downto 0 do
  begin
    if IsEqualGUID(TAnyQBaseUDPTask(_tasklist.Items[i]).TaskID, task.TaskID) then
    begin
      _tasklist.Delete(i);
      exit;
    end;
  end;
end;

destructor TAnyQUDPController.Destroy;
var
  i: integer;
begin
  _Timer.Enabled := False;
  _Timer.Free;
  _server.Active := False;
  for i:= _tasklist.Count-1 downto 0 do
  begin
    try
      if _tasklist.Items[i]<> nil then
        Tobject(_tasklist.Items[i]).Free;
      if i<_tasklist.Count then
        _tasklist.Delete(i);
    except
    end;
  end;
  _tasklist.Free;
//  _server.EndWork(wmRead);
  _server.Free;
  inherited;
end;

function TAnyQUDPController.FindTask(taskstr: string): TAnyQBaseUDPTask;
begin
  Result := FindTask(StringToGUID(taskstr));
end;

function TAnyQUDPController.FindTask(taskid: TGUID): TAnyQBaseUDPTask;
var
  i: integer;
begin
  Result := nil;
  for i:= 0 to _tasklist.Count-1 do
  begin
    if IsEqualGUID(TAnyQBaseUDPTask(_tasklist.Items[i]).TaskID, taskid) then
    begin
      Result := TAnyQBaseUDPTask(_tasklist.Items[i]);
      exit;
    end;
  end;
end;

procedure TAnyQUDPController.GetUDPInfo(InfoList: TstringList; usethread: boolean = false);
var
  tick: Cardinal;
begin
  if InfoList<>nil then
  begin
    InfoList.Clear;
    InfoList.Add(MainSession.Stream.LocalIP+':'+inttostr(_server.DefaultPort));
  end;
{
  try
    if not _UDPInfoGetted then
      _server.Send(MainSession.Server, AnyQUDP_SERVER_Port, 'AnyQ GET UDP INFO');

    tick := GetTickCount;
    while not _UDPInfoGetted do
    begin
      //15秒内没有收到服务器返回的消息的话，认为连接不上
      if GetTickCount-tick > 15000 then
        Break;
      if not usethread then
        Application.ProcessMessages
      else
        Sleep(3);
    end;
  except
  end;
{}
  if _UDPInfoGetted then
  begin
    if InfoList=nil then
      exit;
    if _WANHost+':'+inttostr(_WANPort)<>InfoList[0] then
      InfoList.Add(_WANHost+':'+inttostr(_WANPort));
  end
  else
  begin
    if MainSession.Active then
    try
      _server.Send(MainSession.Server, AnyQUDP_SERVER_Port, 'SICQ GET UDP INFO');
      _UDPInfoGetting := true;
    except
    end;
  end;
end;

procedure TAnyQUDPController.LiveTimeout(Sender: TObject);
begin
  _Timer.Enabled := False;
  try
    if _UDPInfoGetting then
      _UDPInfoGetted := False;
    if MainSession.Active then
    try
      _server.Send(MainSession.Server, AnyQUDP_SERVER_Port, 'SICQ GET UDP INFO');
      _UDPInfoGetting := true;
    except
    end;
  finally
    _Timer.Enabled := True;
  end;
end;

procedure TAnyQUDPController.SendData(host: string; port: Integer; var Data;
  Size: Integer);
{$IFDEF AnyQDEBUG}
var
  f: Textfile;
{$ENDIF}
begin
{$IFDEF AnyQDEBUG}
  AssignFile(f, 'C:\Testlog'+inttostr(_server.DefaultPort)+'.txt');
  if FileExists('C:\Testlog'+inttostr(_server.DefaultPort)+'.txt') then
    Append(f)
  else
    Rewrite(f);

  Writeln(f, 'Send:'+host+':'+inttostr(port));
  Writeln(f, 'SendSize:'+inttostr(size));
  closefile(f);
{$ENDIF}
  try
    _server.SendBuffer(Host, Port, Data, Size);
  except
  end;  
end;

procedure TAnyQUDPController.AnyQUDPRead(Sender: TObject; AData: TStream;
  ABinding: TIdSocketHandle);
var
  UDPHead: TAnyQUDPHead;
  buf: array of byte;
  s: TStringStream;
  i: Integer;
  npos: Integer;
{$IFDEF AnyQDEBUG}
  f: textfile;
{$ENDIF}
begin
{$IFDEF AnyQDEBUG}
  AssignFile(f, 'C:\Testlog'+inttostr(_server.DefaultPort)+'.txt');
  if FileExists('C:\Testlog'+inttostr(_server.DefaultPort)+'.txt') then
    Append(f)
  else
    Rewrite(f);

  Writeln(f, 'RCV:'+abinding.PeerIP+':'+inttostr(abinding.PeerPort));
{$ENDIF}

  //自己发送给自己的包过滤掉
  if MainSession.Stream=nil then exit;
  if (MainSession.Stream.LocalIP = ABinding.PeerIP) and (ABinding.Port=ABinding.PeerPort) then
    exit;

  _Timer.Enabled := False;
  _Timer.Enabled := True;

  AData.Seek(0,0);
  if AData.Size >= sizeof(TAnyQUDPHead) then
  begin
    AData.Read(UDPHead, sizeof(TAnyQUDPHead));

{$IFDEF AnyQDEBUG}
    Writeln(f, 'TaskID:'+GUIDtostring(UDPhead.TaskID));
    Writeln(f, 'DataNo:'+Inttostr(UDPhead.DataNo));
    Writeln(f, 'DataSize:'+Inttostr(UDPhead.DataSize));
    Writeln(f, 'DataType:'+Inttostr(UDPhead.DataType));
    CloseFile(f);
{$ENDIF}

    for i:= 0 to _tasklist.Count-1 do
    begin
      if IsEqualGUID(TAnyQBaseUDPTask(_tasklist.Items[i]).TaskID, UDPHead.TaskID) then
      begin
        AData.Seek(0,0);
        TAnyQBaseUDPTask(_tasklist.Items[i]).RcvData(AData, ABinding);
        exit;
      end;
    end;
    exit;
  end;

  _UDPInfoGetted := true;
  _UDPInfoGetting := False;
  S := TStringStream.Create('');
  s.CopyFrom(AData, AData.Size);
  _WANHost := s.DataString;

{$IFDEF AnyQDEBUG}
  Writeln(f, s.DataString);
  CloseFile(f);
{$ENDIF}
                      
  s.Free;
  npos := pos(':', _WANHost);
  _WANPort := StrToInt(Copy(_WANHost, nPos+1, length(_WANHost)));
  delete(_WANHost, nPos, length(_WANHost));
end;

{ TAnyQUDPTask }
{
function TAnyQUDPTask.Connect(hosts: Tstringlist): Boolean;
var
  i: Integer;
  sHost: string;
  nPos: Integer;
  UDPHead: TAnyQUDPHead;
  tick : cardinal;
begin
  Connected := false;
  if _Controller._server.Active then
    for i := 0 to hosts.Count-1 do
    begin
      try
        sHost := hosts[i];
        nPos := pos(':', sHost);
        DestHost := copy(shost,1,nPos-1);
        Delete(sHost,1,nPos);
        DestPort := strtoint(sHost);
        UDPHead.TaskID := TaskID;
        UDPHead.DataNo := 0;
        UDPHead.DataSize := 0;
        UDPHead.DataType := 0;
        SendSpecData(UDPHead, sizeof(TAnyQUDPHead));
        if SyncDataSend then
        begin
          _Tick := GetTickCount;
//          _Timer.Enabled := false;
          if not _Timer.Enabled then
            _Timer.Enabled := true;
        end;
        tick := GetTickCount;
        while not Connected do
        begin
          if GetTickCount-tick>15000 then
            Break;
          Application.ProcessMessages;
//          if not usethread then
//            Application.ProcessMessages
//          else
//            Sleep(3);
        end;

        if Connected then
        begin
          Result := true;
          exit;
        end;
      except
      end;
    end;
  if Assigned(FOnConnectFail) then
    FOnConnectFail(self);
end;

constructor TAnyQUDPTask.Create(Controller: TAnyQUDPController; GUID: TGUID);
begin
  TaskID := GUID;
  _Controller := Controller;
  _Controller.Addtask(self);
  _Timer := TTimer.Create(nil);
  _Timer.Interval := 1000;
  _Timer.OnTimer := RcvTimeout;
  _Timer.Enabled :=False;
  UseThread := false;
  ISAV := false;
  SendNo := 0;
  RcvNo := 0;
  Connected := false;
  setlength(_Buffer, AnyQUDP_MAXSize-sizeof(TAnyQUDPHead));
  _BufferAlloc := 0;
end;

destructor TAnyQUDPTask.Destroy;
begin
  _Controller.Deltask(self);
  _Timer.Enabled := false;
  _Timer.Free;
  inherited;
end;

procedure TAnyQUDPTask.RcvData(AData: TStream; ABinding: TIdSocketHandle);
var
  UDPHead: TAnyQUDPHead;
  Buf: array of Byte;
begin
  if SyncDataSend then
    if _Timer.Enabled then
    begin
      _Tick := GetTickCount;
      if not _Timer.Enabled then
        _Timer.Enabled := true;
    end;
  AData.Seek(0,0);
  AData.Read(UDPHead, sizeof(TAnyQUDPHead));
  if UDPHead.DataNo = 0 then
  begin
    DestHost := ABinding.PeerIP;
    DestPort := ABinding.PeerPort;
    if UDPHead.DataType = 0 then
    begin
      UDPHead.DataType := 1;
      setlength(Buf, sizeof(TAnyQUDPHead));
      Move(UDPHead, BUF[0], sizeof(TAnyQUDPHead));
      SendSpecData(BUF[0], Length(BUF));
      Connected := True;
    end;
    if UDPHead.DataType = 1 then
      Connected := True;

    if Assigned(FOnConnected) then
      FOnConnected(self);
  end
  else
  begin
    if UDPHead.DataType = 1 then
      if SyncDataSend then
        Synchronized := UDPHead.DataNo=SendNo;
    if UDPHead.DataType = 0 then
    begin
      if SyncDataSend then
      begin
        RcvNo := UDPHead.DataNo;
        UDPHead.DataSize := 0;
        UDPHead.DataType := 1;
        setlength(Buf, sizeof(TAnyQUDPHead));
        Move(UDPHead, BUF[0], sizeof(TAnyQUDPHead));
        SendSpecData(BUF[0], Length(BUF));
      end;
      AData.Seek(0,0);
      if Assigned(FOnRcvData) then
        FOnRcvData(self, Adata);
    end;
  end;
end;

procedure TAnyQUDPTask.RcvTimeout(sender: TObject);
begin
  if GetTickCount - _Tick < 60000 then
    exit;
  if Assigned(FOnRcvTimeOut) then
    FOnRcvTimeOut(self);
end;

procedure TAnyQUDPTask.SendData(const Data; Size: Integer);
var
  buf: array of byte;
  UDPHead: TAnyQUDPHead;
  tick: Cardinal;
begin
  Inc(SendNo);
  if SendNo=0 then
    inc(SendNo);
  UDPHead.TaskID := TaskID;
  UDPHead.DataNo := SendNo;
  UDPHead.DataSize := Size;
  setlength(Buf, Size+sizeof(TAnyQUDPHead));
  Move(UDPHead, BUF[0], sizeof(TAnyQUDPHead));
  Move(DATA, BUF[sizeof(TAnyQUDPHead)], Size);
  if SyncDataSend then
    Synchronized := false;
  SendSpecData(Buf[0], Length(Buf));
  if SyncDataSend then
    if _Timer.Enabled then
    begin
      _Tick := GetTickCount;
      if not _Timer.Enabled then
        _Timer.Enabled := true;
    end;
  tick := GetTickCount;
  if SyncDataSend then
    while not Synchronized do
    begin
      if GetTickCount-tick>30000 then
      begin
        Connected := false;
        if Assigned(FOnSendTimeOut) then
          FOnSendTimeOut(self);
        exit;
      end;
      if not usethread then
        Application.ProcessMessages
      else
        Sleep(3);
    end;
end;

procedure TAnyQUDPTask.SendDataBuffered(const Data; Size: Integer;
  EndBuffer: Boolean);
var
  i: Integer;
  m: TMemoryStream;
  tick :cardinal;
begin
  m := TMemoryStream.Create;
  m.Write(Data,Size);
  m.Seek(0,0);
  while m.Position<m.Size do
  begin
    _bufferAlloc := _bufferAlloc + m.Read(_buffer[_bufferAlloc], AnyQUDP_MAXSize-sizeof(TAnyQUDPHead)-_bufferAlloc);
    if _bufferAlloc >= AnyQUDP_MAXSize-sizeof(TAnyQUDPHead) then
    begin
      SendData(_buffer[0], _bufferAlloc);
      if not SyncDataSend then
        if not ISAV then
          sleep(10)
        else
        begin
          sleep(5)
        end;

      _bufferAlloc := 0;
    end;
  end;
  if EndBuffer then
  begin
    SendData(_buffer[0], _bufferAlloc);
    _bufferAlloc := 0;
  end;
end;

procedure TAnyQUDPTask.SendEndFlag;
var
  buf: array of byte;
  UDPHead: TAnyQUDPHead;
  tick: Cardinal;
begin
  Inc(SendNo);
  if SendNo=0 then
    inc(SendNo);
  UDPHead.TaskID := TaskID;
  UDPHead.DataNo := SendNo;
  UDPHead.DataSize := 0;
  UDPHead.DataType := 0;
  setlength(Buf, sizeof(TAnyQUDPHead));
  Move(UDPHead, BUF[0], sizeof(TAnyQUDPHead));
  if SyncDataSend then
    Synchronized := false;
  SendSpecData(Buf[0], Length(Buf));
  tick := GetTickCount;
  if SyncDataSend then
    while not Synchronized do
    begin
      if GetTickCount-tick>30000 then
      begin
        Connected := false;
        abort;
      end;
      if not usethread then
        Application.ProcessMessages
      else
        Sleep(3);
    end;  
end;

procedure TAnyQUDPTask.SendSpecData(var Data; Size: Integer);
begin
  _Controller.SendData(DestHost,DestPort,Data,size);
end;

{ TAnyQBaseUDPTask }

procedure TAnyQBaseUDPTask.Connect;
var
  host: string;
  port: Integer;
  npos: Integer;
  UDPHead: TAnyQUDPHead;
begin
  Status := 'connecting';
  if _ConnectIndex >= _Hosts.Count then
  begin
    if Assigned(FOnFail) then
      FOnFail(self);
    exit;
  end;

  Host := _Hosts[_ConnectIndex];
  nPos := pos(':', Host);
  Port := strtoint(copy(host,nPos+1,length(host)-npos));
  Host := copy(host,1,nPos-1);
  _CurHost := Host;
  _CurPort := Port;

  UDPHead.TaskID := TaskID;
  UDPHead.DataNo := 0;
  UDPHead.DataSize := 0;
  UDPHead.DataPackageSize := 0;
  UDPHead.DataType := 0;
  SendRawData(UDPHead, sizeof(TAnyQUDPHead));
end;

constructor TAnyQBaseUDPTask.Create(Controller: TAnyQUDPController;
  GUID: TGUID);
begin
  TaskID := GUID;
  _DataSizeQueue := TQueue.Create;
  _Controller := Controller;
  _Controller.Addtask(self);
  _Timer := TTimer.Create(nil);
  _Timer.Interval := 1000;
  _Timer.OnTimer := Timeout;
  _Timer.Enabled :=False;
  _ConnectIndex := 0;
  _SendBuffer := TMemoryStream.Create;
  _RcvBuffer := TMemoryStream.Create;
  _TryCount := 0;
  _Hosts := TStringList.Create;
  _Sending := False;
  _Receiving := False;
  PackageSize := AnyQUDP_MAXSize;
  SendNo := 0;
  RcvNo := 0;
  NoCheck := false;
end;

destructor TAnyQBaseUDPTask.Destroy;
begin
  _SendBuffer.Free;
  _RcvBuffer.Free;
  _Controller.Deltask(self);
  _Timer.Enabled := false;
  _Timer.Free;
  _Timer := nil;
  _Hosts.Free;
  _DataSizeQueue.Free;
  inherited;
end;

procedure TAnyQBaseUDPTask.RcvData(AData: TStream;
  ABinding: TIdSocketHandle);
var
  UDPHead: TAnyQUDPHead;
  Buf: array of Byte;
  OldNo: Cardinal;
begin
  _Timer.Enabled := False;
  _TryCount := 0;
  AData.Seek(0,0);
  AData.Read(UDPHead, sizeof(TAnyQUDPHead));

  if UDPHead.DataNo = 0 then
  begin
    _CurHost := ABinding.PeerIP;
    _CurPort := ABinding.PeerPort;
    if UDPHead.DataType = 0 then
    begin
      UDPHead.DataType := 1;
      setlength(Buf, sizeof(TAnyQUDPHead));
      Move(UDPHead, BUF[0], sizeof(TAnyQUDPHead));
      SendRawData(BUF[0], Length(BUF));
      if (Status='') or (Status='connecting') then
        Status := 'connected';
    end;

    if UDPHead.DataType = 1 then
      if (Status='') or (Status='connecting') then
        Status := 'connected';

  end
  else
  begin
    if UDPHead.DataType = 1 then
    begin
      SendNext;

    end;
    if UDPHead.DataType = 0 then
    begin
      if UDPHead.DataSize=0 then
      begin
        if NoCheck then
          ProcRcvData(AData)
        else
          if (RcvNo+1)= UDPHead.DataNo then
            ProcRcvData(AData);
        Free;
        exit;
      end
      else
      begin
        OldNo := RcvNo;
        RcvNo := UDPHead.DataNo;
        UDPHead.DataSize := 0;
        UDPHead.DataPackageSize := 0;
        UDPHead.DataType := 1;
        setlength(Buf, sizeof(TAnyQUDPHead));
        Move(UDPHead, BUF[0], sizeof(TAnyQUDPHead));
        _CurHost := ABinding.PeerIP;
        _CurPort := ABinding.PeerPort;
        SendRawData(BUF[0], Length(BUF));
        if NoCheck then
          ProcRcvData(AData)
        else
          if (OldNo+1)= UDPHead.DataNo then
            ProcRcvData(AData);
      end;
    end;
  end;
  try
    if Assigned(FOnRcvStatus) then
      FOnRcvStatus(self);{}
  except
    if Assigned(FOnFail) then
      FOnFail(self);{}
  end;
end;

procedure TAnyQBaseUDPTask.ProcRcvData(AData: TStream);
var
  UDPHead: TAnyQUDPHead;
  Buf: array of Byte;
begin
  AData.Seek(0,0);
  AData.Read(UDPHead, sizeof(TAnyQUDPHead));
  if _RcvDataPackageSize=0 then
    _RcvBuffer.Clear;

  if not _receiving then
  begin
    _RcvDataPackageSize := UDPHead.DataPackageSize;
    _RcvBuffer.Clear;
  end;

  if NoCheck then
    if _RcvDataPackageSize <> UDPHead.DataPackageSize then
    begin
      _RcvDataPackageSize := UDPHead.DataPackageSize;
      _RcvBuffer.Clear;
    end;

  SetLength(Buf, UDPHead.DataSize);
  AData.Read(Buf[0], UDPHead.DataSize);
  _RcvBuffer.Write(Buf[0], UDPHead.DataSize);
  if _RcvBuffer.Size < _RcvDataPackageSize then
  begin
    _receiving := true;
    Status := 'receiving'
  end
  else
  begin
    _receiving := false;
    Status := 'received';
    _RcvDataPackageSize := 0;
    if _RcvBuffer.Size = UDPHead.DataPackageSize then
      if Assigned(FOnRcvData) then
        FOnRcvData(self, _RcvBuffer);
    exit;
  end;
  _Timer.Enabled := True;
end;

procedure TAnyQBaseUDPTask.SendData(const Data; Size: Integer;
  EndData: Boolean);
var
  buf: array of byte;
  UDPHead: TAnyQUDPHead;
  DataSize: Integer;
  OLdPos: Integer;
begin
  if not NoCheck then
  begin
    OLdPos := _SendBuffer.Position;
    _SendBuffer.Seek(0, soFromEnd);
    _SendBuffer.Write(Data,Size);
    _DataSizeQueue.Push(Pointer(Size));
    _SendBuffer.Seek(OLdPos, 0);
    if _sending then
      exit;
    if _SendDataPackageSize=0 then
      _SendDataPackageSize := Integer(_DataSizeQueue.pop);
  end
  else
  begin
    _SendBuffer.Clear;
    _SendBuffer.Write(Data,Size);
    _SendBuffer.Seek(0, 0);
    _SendDataPackageSize := Size;
//    if _sending then
//      exit;
  end;
   Status := 'sending';
  _sending := True;
  Inc(SendNo);
  if SendNo=0 then
    inc(SendNo);
  UDPHead.TaskID := TaskID;
  UDPHead.DataNo := SendNo;
  UDPHead.DataType := 0;
  UDPHead.DataPackageSize := _SendDataPackageSize;
  DataSize := min(PackageSize, _SendDataPackageSize-_SendBuffer.Position+sizeof(TAnyQUDPHead));
  setlength(Buf, DataSize);
  _SendBuffer.Read(BUF[sizeof(TAnyQUDPHead)], DataSize-sizeof(TAnyQUDPHead));
  UDPHead.DataSize := DataSize-sizeof(TAnyQUDPHead);
  Move(UDPHead, BUF[0], sizeof(TAnyQUDPHead));
  SendRawData(Buf[0], DataSize);
  sleep(2);
end;

procedure TAnyQBaseUDPTask.SendEndFlag;
var
  buf: array of byte;
  UDPHead: TAnyQUDPHead;
begin
  Status := 'ending';
  Inc(SendNo);
  if SendNo=0 then
    inc(SendNo);
  UDPHead.TaskID := TaskID;
  UDPHead.DataNo := SendNo;
  UDPHead.DataPackageSize := 0;
  UDPHead.DataSize := 0;
  UDPHead.DataType := 0;
  setlength(Buf, sizeof(TAnyQUDPHead));
  Move(UDPHead, BUF[0], sizeof(TAnyQUDPHead));
  SendRawData(Buf[0], Length(Buf));
end;

procedure TAnyQBaseUDPTask.Sendnext;
var
  buf: array of byte;
  UDPHead: TAnyQUDPHead;
  DataSize: Integer;
begin
  if _SendBuffer.Position<_SendBuffer.Size then
  begin
    if _SendBuffer.Position<_SendDataPackageSize then
    begin
      Status := 'sending';
      _Sending := true;
      Inc(SendNo);
      if SendNo=0 then
        inc(SendNo);
      UDPHead.TaskID := TaskID;
      UDPHead.DataNo := SendNo;
      UDPHead.DataPackageSize := _SendDataPackageSize;
      UDPHead.DataType := 0;
      DataSize := min(PackageSize, _SendDataPackageSize-_SendBuffer.Position+sizeof(TAnyQUDPHead));
      setlength(Buf, DataSize);
      _SendBuffer.Read(BUF[sizeof(TAnyQUDPHead)], DataSize-sizeof(TAnyQUDPHead));
      UDPHead.DataSize := DataSize-sizeof(TAnyQUDPHead);
      Move(UDPHead, BUF[0], sizeof(TAnyQUDPHead));
      SendRawData(Buf[0], DataSize);
    end
    else
    begin
      Move(Pointer(Integer(_SendBuffer.Memory)+_SendBuffer.Position)^, _SendBuffer.Memory^, _SendBuffer.Size - _SendBuffer.Position);
      _SendBuffer.Size := _SendBuffer.Size - _SendBuffer.Position;
  //    _SendBuffer.Clear;
      if _DataSizeQueue.Count>0 then
        _SendDataPackageSize := Integer(_DataSizeQueue.pop);
      _SendBuffer.Seek(0,0);
      Sendnext;
    end;
  end
  else
  begin
    Status := 'sended';
    _Sending := False;
    _SendBuffer.Clear;
    _SendDataPackageSize := 0;
  end;
end;

procedure TAnyQBaseUDPTask.SendRawData(var Data; Size: Integer);
begin
  SetLength(_SendBuf, Size);
  _SendSize := Size;
  Move(Data, _SendBuf[0], Size);
  _Controller.SendData(_CurHost,_CurPort,_SendBuf[0], _SendSize);
  _Tick := GetTickCount;
  _Timer.Enabled := True;
end;

procedure TAnyQBaseUDPTask.Timeout(sender: TObject);
begin
  if Status='connecting' then
  begin
    if GetTickCount-_Tick > 5000 then
    begin
      _ConnectIndex := _ConnectIndex+1;
      Connect;
    end;
  end
  else
//    if Status='receiving' then
    if _receiving then
    begin
      if NoCheck then
      begin
        if GetTickCount-_Tick > 60000 then
        begin
          if assigned(FOnFail) then
            FOnFail(Self);
        end;
      end
      else
        if GetTickCount-_Tick > 15000 then
        begin
          if assigned(FOnFail) then
            FOnFail(Self);
        end;
    end
    else
      if GetTickCount-_Tick > 5000 then
      begin
        if NoCheck then
        begin
          if _Sending then
            _Sending := False;
          exit;
        end;
        _TryCount := _TryCount+1;
        if _TryCount<3 then
          Resend
        else
          if _Sending then
            if assigned(FOnFail) then
              FOnFail(Self);
      end;
end;

procedure TAnyQBaseUDPTask.Resend;
begin
  _Controller.SendData(_CurHost,_CurPort,_SendBuf[0], _SendSize);
  _Tick := GetTickCount;
end;

procedure TAnyQBaseUDPTask.Connect(hosts: Tstrings);
begin
  _Hosts.Assign(Hosts);
  Connect;
end;

end.
