完成端口-Delphi应用（3）

eighteen

{/****************下面给出完成端口工作线程的主要代码有兴趣的朋友可以参考研究******************/}

  { 报警控制TCP服务器工作线程的定义 }
  TAlarmWorkThread = class(TThread)
    private
       FThreadPool      : TThreadPool;       //线程池指针
       FCompletePort    : THandle;           //完成端口
       FAlarmSocketLists: TThreadList;       //存放客户Socket相关信息的线程安全列表
       FTimeOutCounter  : TTimeOutCounter;
    private
       function FCloseConnAndFree(PHandleDataPPerHandelData):Integer;//关闭连接并释放相关资源
       function FGetClientID(APhandlePPerHandelData):integer;       //取得ClientID
       function FGetClientIDByAddr(Addr:String):Integer;              //通过目标地址名取得ClientID
       function FGetClientIODataByAddr(Addr:String)PPerHandelData;      //取得Client的IOdata
       function FCheckPacketHeader(var WimpHeader;Len:Byte):Integer;  //处理包头
       function FGetLenOfBodyByHeader(Cmd:Word):Integer;              //根据包头取得包体的长度
       function FReceiveData(PHandleData: LPPerHandelData;Block: PBlock = nil):Boolean; //接收包头
       function FMergePacked(var WimpPacket:TWimpMsg;const Cmd:Word;Var buf):ShortInt;         //把缓冲区中的数据合并为一个完整协议包
       function FSendPacket(PHandleData: LPPerHandelData; var HeadBuf,BodyBuf; HeadLen, BodyLen: Integer):Integer;      //发送一个协议包
       function FHandleRecPacket(var WimpPacket:TWimpMsg; const  CMD : Word; PHandleData: LPPerHandelData):ShortInt;    //处理已经接收完成的协议包
       function FRecAndMakePac(PHandleData: LPPerHandelData;Block: PBlock; Transfered: DWORD):Integer; //接收并拼凑协议包
       procedure FPostFEListAndParmToRC(PHandleData: LPPerHandelData; RCAddr: String);  //提交RC需要的FE设备列表和参数
       procedure FPostFEInfoAndParm(PHandleDataPPerHandelData;Addr:String;LoginType:Byte);   //提交FE的配置参数
       function FCheckisCompletePac(PHandleData: LPPerHandelData):Integer;
    public
      procedure close();
      constructor Create(ACompletePortointer;APool:TThreadPool;ASocketLists:TThreadList;ATimeOutCounter:TTimeOutCounter);
      procedure Execute; override;
      destructor Destroy; override;
  end;

 

procedure TAlarmWorkThread.close;
begin
  //给完成端口发消息，关闭工作线程
  PostQueuedCompletionStatus(FCompletePort,0,0,nil);
  Terminate;
end;

constructor TAlarmWorkThread.Create(ACompletePortointer;APool:TThreadPool;ASocketLists:TThreadList;ATimeOutCounter:TTimeOutCounter);
begin
   FreeOnTerminate    := True;
   FTimeOutCounter    := ATimeOutCounter;
   FCompletePort      := THANDLE(ACompletePort);
   FAlarmSocketLists  := ASocketLists;
   FThreadPool        := APool;
   inherited Create(False);
end;

destructor TAlarmWorkThread.Destroy;
begin
  inherited;
end;

procedure TAlarmWorkThread.Execute;
var
   BytesTransferred: DWORD ;
   PHandleData: LPPerHandelData;
   Block: PBlock;
   Flags: DWORD ;
   BodySize,BytesSend,ErrCode : Integer;
   Cmd : Word;
begin
  while (not self.Terminated) do
  begin
    //查询IOCP状态（数据读写操作是否完成）
    if (GetQueuedCompletionStatus(FCompletePort, BytesTransferred, DWORD(PHandleData), POverlapped(Block), INFINITE) = true) then
    begin
        //FThreadPool.PShowCommGauge();
        //工作线程被Post消息提示退出
        if (PHandleData = nil) then
        begin
           FThreadPool.PShowRealMsg(APP_MSGTYPE_THREAD,'控制信息','报警控制TCP服务器释放相关资源，成功关闭服务！');
           Exit;
        end;

        //如果链路断开，释放相关资源
        if (BytesTransferred = 0) then
        begin
           FThreadPool.PShowRealMsg(APP_MSGTYPE_THREAD,'控制信息','客户端【Addr=' + PHandleData.PSrcAddr + '】'+ PHandleData.PSrcName  + '断开，TCP服务器释放相关资源！');
           //清除该客户端相关资源
           FCloseConnAndFree(PHandleData);
           Continue;
        end;

        with Block^.Data do
        try
          case Block^.Data.OperType of
            APP_OPERTYPE_RECV:
            begin
              FRecAndMakePac(PHandleData,Block,BytesTransferred);
              Block.Data.IsUse := False;
              Block.IsUse := False;
              Block := nil;
              FReceiveData(PHandleData,Block);
              Continue;
            end;
          
            APP_OPERTYPE_SEND:
            begin
              Dec(wsaBuffer.len, BytesTransferred);
              if wsaBuffer.len <= 0 then
              begin
                { 发送完成，将Block置空，返回到FBlock的可使用的缓区中 }

                //FSendHandlePacByCmd(PHandleData,Block,BytesTransferred);
                Block.Data.IsUse := False;
                Block.IsUse := False;
                Block := nil;
              end else
              begin
                { 数据还没发送完成，继续发送 }
                Flags := 0;
                Inc(Pbyte(wsaBuffer.buf), BytesTransferred);
                FillChar(Overlapped, SizeOf(Overlapped), 0);
                if SOCKET_ERROR = WSASend( PHandleData.Socket, @wsaBuffer, 1, @BytesSend,Flags, @Overlapped, nil) then
                begin
                  ErrCode := WSAGetLastError;
                  if ErrCode <> ERROR_IO_PENDING then
                    begin
                      FThreadPool.PShowRealMsg(APP_MSGTYPE_THREAD,'控制信息','客户端【Addr=' + PHandleData.PSrcAddr + '】异常断开，TCP服务器释放相关资源！');
                      FCloseConnAndFree(PHandleData);//清除该客户端相关资源
                      Continue;
                    end;
                end;
              end;
            end;
          end;
      except
      end;
    end else begin
       if (PHandleData <> nil ) then
       begin
          FThreadPool.PShowRealMsg(APP_MSGTYPE_THREAD,'视频服务','客户端【Addr=' + PHandleData.PSrcAddr + '】异常断开，TCP服务器释放相关资源！');
          FCloseConnAndFree(PHandleData);//清除该客户端相关资源
       end;
    end;
  end;
end;

function TAlarmWorkThread.FRecAndMakePac(PHandleData: LPPerHandelData;Block: PBlock; Transfered: DWORD): Integer;
begin
  PHandleData.RingBuffer.WriteBuffer(Block.Data.Buffer,Transfered);
  FCheckisCompletePac(PHandleData);
end;

function TAlarmWorkThread.FCheckisCompletePac(PHandleData: LPPerHandelData): Integer;
var
  Tmp:string;
  PackedLen,BodySize: Word;
  PackedHead: TNetAlarmHead;
  dataSize : LongInt;
  RevBuffer: array [0..MAX_BUFFER_LEN] of Char;
  WimpPacket: TWimpMsg;
  cmd:Word;
begin
    dataSize := PHandleData.RingBuffer.GetDataCount();
    if dataSize < SizeOf(TNetAlarmHead) then Exit; //如果不够包头长就退出
    PHandleData.RingBuffer.Copybuffer(PackedHead,SizeOf(TNetAlarmHead));
    cmd := ntohs(PackedHead.Command_Id);
    PackedLen := FGetLenOfBodyByHeader(cmd); //包体长度
    if PackedLen < 0 then
    begin
      PHandleData.RingBuffer.Readbuffer(PackedHead,SizeOf(TNetAlarmHead));
      FCheckisCompletePac(PHandleData); //嵌套执行
    end else begin
      if dataSize >= (PackedLen + SizeOf(TNetAlarmHead)) then
      begin
        PHandleData.RingBuffer.Readbuffer(PackedHead,SizeOf(TNetAlarmHead));
        PHandleData.RingBuffer.Readbuffer(RevBuffer,PackedLen);
        FMergePacked(WimpPacket,cmd,RevBuffer);
        FHandleRecPacket(WimpPacket,cmd,PHandleData);
        FCheckisCompletePac(PHandleData); //嵌套执行
      end;
    end;
end;

function TAlarmWorkThread.FSendPacket(PHandleData: LPPerHandelData; var HeadBuf,BodyBuf; HeadLen, BodyLen: Integer): Integer;
var
  ErrCode: Integer;
  Flags, Transfer: Cardinal;
  Block: PBlock;
begin
  Result := 0;
  Block := PHandleData.MemoryBuffer.AllocBlock;
  with Block^.Data do
  begin
    Flags := 0;
    OperType := APP_OPERTYPE_SEND;
    FillChar(Buffer, SizeOf(Buffer), 0);
    Move(HeadBuf,Buffer,HeadLen);
    Move(BodyBuf,Buffer[HeadLen],BodyLen);
    FillChar(Overlapped, SizeOf(Overlapped), 0);
    wsaBuffer.buf := @buffer;
    wsaBuffer.len := HeadLen + BodyLen;
    if SOCKET_ERROR = WSASend(PHandleData.Socket, @wsaBuffer, 1, @Transfer, Flags, @Overlapped, nil) then
    begin
      ErrCode := WSAGetLastError;
      if ErrCode <> ERROR_IO_PENDING then
      begin
        Result := SOCKET_ERROR;
      end;
    end;
  end;
end;

 

function TAlarmWorkThread.FCloseConnAndFree(PHandleDataPPerHandelData): Integer;
var PIoDataID : Integer;
    Llt : TList;
begin
   Result := 0;
   PIoDataID := FGetClientID(PHandleData);
   FThreadPool.PUpdateOnLine(PHandleData.PSrcType,PHandleData.PSrcAddr,False);
   FThreadPool.PShowRealMsg(APP_MSGTYPE_THREAD,'控制信息','客户端【登录名=' + PHandleData.PSrcAddr + '】'+ PHandleData.PSrcName  + '断开，TCP服务器释放相关资源！');
   if PIoDataID >=0 then
   begin
      FThreadPool.PUpdateCount(PHandleData.PSrcType,APP_UPDATECOUNT_DELETE);
      Llt := FAlarmSocketLists.LockList();
      Llt.Delete(PIoDataID);
      FAlarmSocketLists.UnlockList;
      shutdown(PHandleData.Socket,FD_CLOSE);
      closesocket(PHandleData.Socket);
      FreeAndNil(PHandleData.MemoryBuffer);
      FreeAndNil(PHandleData.RingBuffer);
      GlobalFree(DWORD(PHandleData));
   end;
end;

function TAlarmWorkThread.FGetClientID(APhandle: LPPerHandelData): integer;
var i:Integer;
    Llt:TList;
begin
  Result := -1;
  try
    Llt := FAlarmSocketLists.LockList();
    for i:=0 to Llt.Count - 1 Do
    begin
      if LPPerHandelData(Llt.Items) = APhandle then
      begin
         Result := i;
         Break;
      end;
    end;
  finally
    FAlarmSocketLists.UnlockList;
  end;
end;