iggstreamclient.pas

通信控件

              if (POut <> nil) and (POut.Status = bsWait) then begin
                DesRoute := GetRoute(POut.InAttachID);
                AroundDealOutBlock(POut, DesRoute.RouteIP, DesRoute.RoutePort);
              end else if (POut <> nil) and (POut.Status = bsComplete) then begin
                FOutPool.Cancel(PMsgItem.Msg.OutID);
                FMsgPool.Remove(PMsgItem.MsgSequ);
              end;
            end;
            Dec(PMsgItem.LifeSecond);
            if (PMsgItem.LifeSecond = 0) then
              PMsgItem.Status := msOvertime;
          end;
          msSendSucc:
          begin
            //OnSendMessageSucc();
            FMsgPool.Remove(PMsgItem^.MsgSequ);
          end;
          msRouteFail:
          begin
            //OnSendMessageFail();
            FMsgPool.Remove(PMsgItem^.MsgSequ);
          end;
          msOvertime:
          begin
            //if Assigned(FOnSendMsgTimeoutNotify) then
              //FOnSendMsgTimeoutNotify(PMsgItem.Msg.Describe, PMsgItem.Msg.DescribeSize,
              //      PMsgItem.Msg.Data, PMsgItem.Msg.DataSize, PMsgItem.Msg.CommandV);
            if PMsgItem.Status = msWaitBlock then
              FOutPool.Cancel(PMsgItem.Msg.OutID)
            else begin
              FRoutePool.Cancel(PMsgItem.Msg.RecvID);
              FRoutePool.Route(FSelfID);
            end;
            FMsgPool.Remove(PMsgItem.MsgSequ);
          end;
        end;
        //Dec(PMsgItem.LifeSecond);
        if Index = ExcIndex then Break;
      end;
      CheckInBlock();
    except
      Result := -1;
    end;
  end;

begin
  if not Check then
    Result := MsgPoolMessage(PTag)
  else
    Result := CheckMsgPool();
end;

function TIClient.DoOutPool(var PTag: TPacketTag; Check: Boolean): Integer;

  function OutPoolMessage(var PTag: TPacketTag): Integer;

    procedure OnOutSubBlock(PPacket: PSbSubBlockPacket; IP: Integer; Port: Integer);
    var
      PItem: POutItem; DesRoute: TRoute;
      PRsPacket: PRsSubBlockPacket;
      I, L, DataBit32, Ss: Integer;
    begin
      try
      PItem := FOutPool.RequestBlock(PPacket.Data.OutID, PPacket.Data.InID,
                  PPacket.Header.dwAttachID, PPacket.Data.CompleteBit32, PPacket.Data.First = 1);
      if (PItem <> nil) and (PItem.Status = bsOut) then
      begin
        if (IP = FServerIP) then begin
          DesRoute := GetRoute(PItem.InAttachID);
          if (DesRoute.Revert = 2) then begin
            IP    := DesRoute.RouteIP;
            Port  := DesRoute.RoutePort;
          end;
        end else begin
          DesRoute.LoginIP    := IP;
          DesRoute.LoginPort  := Port;
        end;
        PRsPacket := PRsSubBlockPacket(PktBox.FPThdBuf);
        PRsPacket.Header := BPKHeader(FSelfID, PSPKHeader(PItem.Block.Packet).wSequ,
                  SizeOf(TRsSubBlockPacket), RS_SOCK_REQBLK, DesRoute.LoginIP, DesRoute.LoginPort);
        PRsPacket.Data.OutID        := PItem.Block.OutID;
        PRsPacket.Data.InID         := PItem.Block.InID;
        PRsPacket.Data.SubDataSize  := PItem.Block.SubDataSize;
        PRsPacket.Data.LastBlock    := 0;
        PRsPacket.Data.Reserved     := 0;

        DataBit32 := PItem.Block.DataBit32;
        L := 0;
        for I := 0 to 31 do begin
          if ((DataBit32 and $80000000) = 0) then begin
            L := 31 - I; Break;
          end;
          DataBit32 := DataBit32 shl 1;
        end;
        DataBit32 := PItem.Block.DataBit32;
        Ss := PItem.Block.DataSize-(PItem.Block.SubDataAmount-1)*PItem.Block.SubDataSize;
        for I := 0 to PItem.Block.SubDataAmount-1 do begin
          if (DataBit32 and $1) = 0 then begin
            PRsPacket.Data.SubDataSequ := I;
            if (L <> I) then begin
              Move((PItem.Block.Data + I*PItem.Block.SubDataSize)^, (PktBox.FPThdBuf+36)^, PItem.Block.SubDataSize);
               FUDPServer.SendTo(PRsPacket, PRsPacket.Header.wDataOffset+PItem.Block.SubDataSize, IP, Port);
            end else begin
              PRsPacket.Data.SubDataSize := ss;
              Move((PItem.Block.Data + I*PItem.Block.SubDataSize)^, (PktBox.FPThdBuf+36)^, ss);
              PRsPacket.Data.LastBlock := L;
              FUDPServer.SendTo(PRsPacket, PRsPacket.Header.wDataOffset+ss, IP, Port);
            end;

          end;
          DataBit32 := DataBit32 shr 1;
        end;
      end else if PItem.Status = bsComplete then
      begin
        FMsgPool.Remove(PSPKHeader(PItem.Block.Packet).wSequ);
        FOutPool.Cancel(PItem.Block.OutID);
      end;
      except
        exit;
      end;
    end;

    procedure  OnOutCommand(PPacket: PBlockCommPacket; IP: Integer; Port: Integer);
    begin
      case PPacket^.Data.Command of
        $01:;
      end;
    end;

  var
    PBH: PBPKHeader;
    IP, Port: Integer;
  begin
    Result := 0;
    PBH := PBPKHeader(PTag.Data);
    IP := PTag.SockAddr.sin_addr.S_addr;
    Port := PTag.SockAddr.sin_port;

    if PBH.wCommand = SB_SOCK_REQBLK then begin
      OnOutSubBlock(PSbSubBlockPacket(PTag.Data), IP, Port);
    end else
    if PBH.wCommand = SB_SOCK_IN_CMD then begin
      OnOutCommand(PBlockCommPacket(PTag.Data), IP, Port);
    end;
  end;

begin
  //if Assigned(PTag) then
    OutPoolMessage(PTag);
end;

function TIClient.DoInPool(var PTag: TPacketTag; Check: Boolean): Integer;

  procedure OnInSubBlock(PPacket: PRsSubBlockPacket; PktSize: Integer; IP: Integer; Port: Integer);
  var
    PItem: PInItem;
    I, SubDataSize, SubDataBit32: Integer;
  begin
    PItem := FInPool.CheckValid(PPacket.Data.InID, PPacket.Data.OutID);
    if (PItem <> nil) and (PItem.Status = bsIn) then begin
      SubDataBit32 := $1;
      SubDataBit32 := SubDataBit32 shl PPacket.Data.SubDataSequ;
      if (SubDataBit32 and PItem.Block.DataBit32 = 0) then
      begin
        SubDataSize := PktSize - 36;
        Move((PTag.Data+PPacket.Header.wDataOffset)^,
            (PItem.Block.Data + PPacket.Data.SubDataSequ*PItem.Block.SubDataSize)^, SubDataSize);
        Inc(PItem.Block.CompleteSize, SubDataSize);
        PItem.Block.DataBit32 := PItem.Block.DataBit32 or SubDataBit32;
        I := $FFFFFFFF;
        if (PItem.Block.DataBit32 = I) then begin
          PItem.Status := bsComplete;
        end;
        if (PPacket.Data.LastBlock > 0) or (PItem.Status = bsComplete) then begin
          PItem.InTime := Now();
          AroundDealInBlock(PItem, IP, Port);
        end;
      end;
    end;
  end;

  procedure OnInBlockTask(PPacket: PBlockTaskPacket; IP: Integer; Port: Integer);
  var
    InID: Integer;
    PItem: PInItem;
    InPacket: PChar;
  begin
    InPacket := PChar(PPacket)+SizeOf(TBlockTaskPacket);
    InID := FInPool.Put(PPacket.Data.Route.AttachID, PPacket.Data.BlockID, InPacket, PPacket.Data.PacketSize, PPacket.Data.DataSize,
                   PPacket.Data.DataBit32, PPacket.Data.SubDataSize, PPacket.Data.SubDataAmount);
    if (InID > 0) then
    begin
      PItem := FInPool.CheckValid(InID);
      if (PItem <> nil) then begin
        PItem.Status := bsIn;
        AroundDealInBlock(PItem, IP, Port);
      end;
    end;
  end;

var
  PBH: PBPKHeader;
  IP, Port: Integer;
begin
  Result := 0;
  PBH := PBPKHeader(PTag.Data);
  IP := PTag.SockAddr.sin_addr.S_addr;
  Port := PTag.SockAddr.sin_port;
  if (PBH.wCommand = RS_SOCK_REQBLK) then begin
    OnInSubBlock(PRsSubBlockPacket(PTag.Data), PTag.DataSize, IP, Port);
  end else
  if (PBH.wCommand = SB_SOCK_REQ) then begin
    OnInBlockTask(PBlockTaskPacket(PTag.Data), IP, Port);
  end else
  if (PBH.wCommand = SB_SOCK_OUT_CMD) then begin
  end;
end;

procedure TIClient.AroundDealOutBlock(var PItem: POutItem; IP: Integer=0; Port: Integer=0);
var
  PPacket: PBlockTaskPacket;
  PWH: PWPKHeader;
  DesRoute: TRoute;
begin
  PWH     := PWPKHeader(PItem.Block.Packet);
  PPacket := PBlockTaskPacket(PktBox.FPBlkBuf2);
  PPacket.Header := BPKHeader(FSelfID, PWH.SPK.wSequ, SizeOf(TBlockTaskPacket), SB_SOCK_REQ);
  PPacket.Data.InAttachID     := PItem.InAttachID;
  PPacket.Data.BlockID        := PItem.Block.OutID;
  Move(PItem.Block.Packet^, (PktBox.FPBlkBuf2+SizeOf(TBlockTaskPacket))^, PItem.Block.PacketSize);
  PPacket.Data.PacketSize     := PItem.Block.PacketSize;
  PPacket.Data.DataSize       := PItem.Block.DataSize;
  PPacket.Data.DataBit32      := PItem.Block.DataBit32;
  PPacket.Data.SubDataSize    := PItem.Block.SubDataSize;
  PPacket.Data.SubDataAmount  := PItem.Block.SubDataAmount;

  DesRoute := GetRoute(PItem.InAttachID);
  PPacket.Data.Route.AttachID   := FSelfID;
  PPacket.Data.Route.LoginIP    := FLoginIP;
  PPacket.Data.Route.LoginPort  := FLoginPort;
  PPacket.Data.Route.LocalIP    := FLocalIP;
  PPacket.Data.Route.LocalPort  := FLocalPort;
  PPacket.Data.Route.Revert     := DesRoute.Revert;
  //PPacket.Data.Route.SendRoute  := DesRoute.RecvRoute;
  //PPacket.Data.Route.RecvRoute  := DesRoute.SendRoute;
  if (DesRoute.Revert = 2) then begin
    FUDPServer.SendTo(PPacket, PPacket.Header.wDataOffset+PItem.Block.PacketSize, DesRoute.RouteIP, DesRoute.RoutePort);
  end else if (DesRoute.LoginIP <> 0) then begin
    PPacket.Header.dwIP := DesRoute.LoginIP;
    PPacket.Header.wPort := DesRoute.LoginPort;
    FUDPServer.SendTo(PPacket, PPacket.Header.wDataOffset+PItem.Block.PacketSize, FServerIP, FServerPort);
  end;
end;

procedure TIClient.AroundDealInBlock(var PItem: PInItem; IP: Integer; Port: Integer);
var
  PWH: PWPKHeader; gData: Pointer;
  PPacket: PSbSubBlockPacket;
  DesRoute: TRoute;
  Tag: TPacketTag;
begin
  if (PItem.Status = bsComplete) then
  begin
    PWH := PWPKHeader(PItem.Block.Packet);
    gData := Data(PWH);
    case PWH.SPK.bpkType of
      pkWPKH: begin
        //if Assigned(FOnRecvMessage) then
        //  FOnRecvMessage(Self, gData, PWH.SPK.wpkSize-PWH.SPK.wDataOffset, PItem.Block.Data, PItem.Block.DataSize, PWH.SPK.wCommand, PWH.SPK.wRespond);
       // if Assigned(FOnRecvMsgProc) then
        //  FOnRecvMsgProc(PWH.SPK.dwSendID, gData^, PWH.SPK.wpkSize - PWH.SPK.wDataOffset,
         //               PItem.Block.Data, PItem.Block.DataSize, PWH.SPK.wCommand, PWH.SPK.wRespond);
      end;
      pkMPKH: begin
         Tag.Data := PItem.Block.Packet;
         Tag.DataSize := PItem.Block.PacketSize;
         DoBufferPool(Tag, FALSE);
      end;
    end;
  end;

  if (IP = FServerIP) then
  begin
    DesRoute := GetRoute(PItem.OutAttachID);
    if (DesRoute.Revert = 2) then begin
      IP := DesRoute.RouteIP;
      Port := DesRoute.RoutePort;
    end;
  end else begin
    DesRoute.LoginIP := IP;
    DesRoute.LoginPort := Port;
  end;

  PPacket := PSbSubBlockPacket(PktBox.FPBlkBuf1);
  try
    if (PItem.Block.Packet <> nil) and Assigned(PItem.Block.Packet) then
    PPacket.Header := BPKHeader(FSelfID, PSPKHeader(PItem.Block.Packet).wSequ,
          SizeOf(TSbSubBlockPacket), SB_SOCK_REQBLK, DesRoute.LoginIP, DesRoute.LoginPort);

  PPacket.Data.OutID          := PItem.Block.OutID;
  PPacket.Data.InID           := PItem.Block.InID;
  PPacket.Data.CompleteBit32  := PItem.Block.DataBit32;
  PPacket.Data.CompleteSize   := PItem.Block.CompleteSize;
  if (PPacket.Data.CompleteSize <> 0) then
    PPacket.Data.First := 0
  else
    PPacket.Data.First := 1;
  FUDPServer.SendTo(PPacket, PPacket.Header.wDataOffset, IP, Port);

  if (PItem.Status = bsComplete) then
    FInPool.Cancel(PItem.Block.InID);
  except
    exit;
  end;  
end;

function TIClient.SendRecMsgProc(RecvID: DWord; var Rec; RecSize: Integer; Command: Word; Respond: Word; Rule: Integer): Integer;
begin
  Result := FMsgPool.Put(RecvID, Command, Respond, Rec, RecSize, nil, 0, Rule);
  if (Result <> 0) then
  begin
    Result := Send(RecvID, Result);
  end;
end;

function TIClient.SendMsgProc(RecvID: DWord; var Rec; RecSize: Integer; Data: Pointer; DataSize: Integer;
  Command: Word; Respond: Word; Rule: Integer): Integer;
begin
  Result := FMsgPool.Put(RecvID, Command, Respond, Rec, RecSize, Data, DataSize, Rule);
  if Result <> 0 then
  begin
    Result := Send(RecvID, Result);
  end;
end;

function TIClient.Send(RecvID: Integer; Sequ: Integer): Integer;

  function UseBlock(var PItem: PMessageItem; var Route: TRoute): Boolean;
  var
    PWH: PWPKHeader;
    POut: POutItem;
  begin
    Result := (PItem.Status = msSend) and (PItem.Msg.Reserved = 1);
    if Result then
    begin
      PWH  := PWPKHeader(PItem.Msg.Rec);
      PWH^ := WPKHeader(PItem.Msg.RecvID, PItem.MsgSequ, PItem.Msg.RecSize, PItem.Msg.CommandV, PItem.Msg.RespondV,
                        Route.LoginIP, Route.LoginPort, pkWPKH);
      PWH.SPK.wReserved[0] := PItem.Msg.DataSize;
      PItem.Msg.OutID := FOutPool.Put(PItem.Msg.RecvID, PItem.Msg.Rec, PWH.SPK.wPkSize, PItem.Msg.Data, PItem.Msg.DataSize);
      if (PItem.Msg.OutID > 0) then begin
        FMsgPool.SetStatus(PItem.MsgSequ, msWaitBlock);
        POut := FOutPool.Find(PItem.Msg.OutID);
        if (POut <> nil) then
          AroundDealOutBlock(POut, 0, 0);
      end;
    end;
  end;

var
  PItem: PMessageItem;
  SendRoute: TRoute;
  PSH: PSPKHeader;
begin
  Result := 0;
  SendRoute := GetRoute(RecvID);
  if FMsgPool.SendGet(PItem, Sequ) and (PItem <> nil) and (not(UseBlock(PItem, SendRoute))) then
  begin
    PSH := Packeted(PItem, SendRoute, OthBUF_);
    if (PSH <> nil) then begin
      Result := FUDPServer.SendTo(PSH, PSH.wpkSize, SendRoute.RouteIP, SendRoute.RoutePort);
      if (RecvID = 0) then
        FMsgPool.Remove(PItem.MsgSequ);
    end;
  end;
end;

function TIClient.PackedBuffer(var PItem: PMessageItem; var Route: TRoute): Pointer;
var
  PWH: PWPKHeader;
  DataSize: Integer;
  P: PChar;
begin
  PWH := PWPKHeader(PktBox.FPOthBuf); P := PktBox.FPOthBuf;
  try
    DataSize := PItem.Msg.RecSize + PItem.Msg.DataSize;
    PWH^ := WPKHeader(PItem.Msg.RecvID, PItem.MsgSequ, DataSize, PItem.Msg.CommandV, 0,
                      Route.LoginIP, Route.LoginPort, pkMPKH);
    PWH.SPK.wReserved[0] := PItem.Msg.DataSize;
    P := P + PWH.SPK.wDataOffset;
    Move(PItem.Msg.Rec^, P^, DataSize);
    Result := PWH;
  except
    Result := nil;
  end;
end;

function TIClient.SendBufferMsgProc(RecvID: DWord; var Rec; RecSize: Integer; Cmd: Word;
  Rule: Word; Data: Pointer; DataSize: Integer): Integer;

  function UseBlock(var PItem: PMessageItem; var Route: TRoute): Boolean;
  var
    PWH: PWPKHeader;
    POut: POutItem;
  begin
    Result := (PItem.Status = msSend) and (PItem.Msg.Reserved = 1);
    if Result then
    begin
      PWH := PWPKHeader(PItem.Msg.Rec);
      PWH^ := WPKHeader(PItem.Msg.RecvID, PItem.MsgSequ, PItem.Msg.RecSize, PItem.Msg.CommandV, 0,
                        Route.LoginIP, Route.LoginPort, pkMPKH);
      PWH.SPK.wReserved[0] := PItem.Msg.DataSize;
      PItem.Msg.OutID := FOutPool.Put(PItem.Msg.RecvID, PItem.Msg.Rec, PWH.SPK.wPkSize, PItem.Msg.Data, PItem.Msg.DataSize, 0);
      if PItem.Msg.OutID > 0 then begin
        FMsgPool.SetStatus(PItem.MsgSequ, msWaitBlock);
        POut := FOutPool.Find(PItem.Msg.OutID);
        if (POut <> nil) and (POut.Status = bsWait) then
          AroundDealOutBlock(POut, 0, 0);
      end;
    end;
  end;

var
  PItem: PMessageItem;
  SendRoute: TRoute;
  PWH: PWPKHeader;
begin
  Result := FMsgPool.Put(RecvID, Cmd, 0, Rec, RecSize, Data, DataSize, Rule);
  if (Result > 0) then
  begin
    SendRoute := GetRoute(RecvID);
    if FMsgPool.SendGet(PItem, Result) and (PItem <> nil) and (not(UseBlock(PItem, SendRoute))) then
    begin
      PWH := PackedBuffer(PItem, SendRoute);
      if (PWH <> nil) then begin
        Result := FUDPServer.SendTo(PWH, PWH.SPK.wpkSize, SendRoute.RouteIP, SendRoute.RoutePort);
        FMsgPool.Remove(PItem.MsgSequ);
      end;
    end;
  end;
end;

end.