Delphi-IOCP研究笔记<七>===接收数据的解码器(Decoder)和数据处理

if PerIoData.IO_TYPE = IO_TYPE_Recv then
begin
            //加入到套接字对应的缓存中，处理逻辑
            lvClientContext.RecvBuffer(PerIoData.DataBuf.buf, PerIoData.Overlapped.InternalHigh);
            TIODataMemPool.instance.giveBackIOData(PerIoData);

procedure TClientContext.RecvBuffer(buf: PAnsiChar; len: Cardinal);
var
  lvObject:TObject;
begin
  FCS.Enter;
  try
    //加入到套接字对应的缓存
    FBuffers.AddBuffer(buf, len);

    //调用注册的解码器<进行解码>
    lvObject := TContextFactory.instance.FDecoder.Decode(FBuffers);
    if lvObject <> nil then
    try
      //解码成功，调用业务逻辑的处理方法
      dataReceived(lvObject);
    finally
      lvObject.Free;
    end;
  finally
    FCS.Leave;
  end;
end;

procedure TfrmMain.btnIOCPAPIRunClick(Sender: TObject);
var
  lvData:PInteger;

  lvdwThreadId:DWORD;
begin
  __ListenPort := StrToInt(edtPort.Text);

  //注册解码器
  TContextFactory.instance.registerDecoder(FDecoder);
    
  CreateThread(nil, 0, @D10_IOCPRun, Pointer(__ListenPort), 0, lvdwThreadId);
end;

function TJSonStreamDecoder.Decode(const inBuf: TBufferLink): TObject;
var
  lvJSonLength, lvStreamLength:Integer;
  lvData:String;
  lvBuffer:array of Char;
  lvBufData:PAnsiChar;
  lvStream:TMemoryStream;
  lvJsonStreamObject:TJSonStreamObject;
begin
  Result := nil;

  //如果缓存中的数据长度不够包头长度，解码失败<json字符串长度,流长度>
  if (inBuf.validCount < SizeOf(Integer) + SizeOf(Integer)) then
  begin
    Exit;
  end;

  //记录读取位置
  inBuf.markReaderIndex;
  inBuf.readBuffer(@lvJSonLength, SizeOf(Integer));
  inBuf.readBuffer(@lvStreamLength, SizeOf(Integer));

  //如果缓存中的数据不够json的长度和流长度<说明数据还没有收取完毕>解码失败
  if inBuf.validCount < (lvJSonLength + lvStreamLength) then
  begin
    //返回buf的读取位置
    inBuf.restoreReaderIndex;
    exit;
  end;


  //解码成功
  lvJsonStreamObject := TJSonStreamObject.Create;
  Result := lvJsonStreamObject;

  //读取json字符串
  if lvJSonLength > 0 then
  begin
    SetLength(lvData, lvJSonLength);
    ZeroMemory(@lvData[1], lvJSonLength);
    inBuf.readBuffer(@lvData[1], lvJSonLength);
    lvJsonStreamObject.JSon := SO(lvData);
  end;


  //读取流数据 
  if lvStreamLength > 0 then
  begin
    lvStream := TMemoryStream.Create;
    lvStream.Size := lvStreamLength;
    inBuf.readBuffer(lvStream.Memory, lvStreamLength);
    lvJsonStreamObject.setStream(lvStream);
  end;

end;

procedure TClientContext.dataReceived(const pvDataObject:TObject);
var
  lvJsonStream:TJSonStreamObject;
  lvFile:String;
begin
  lvJsonStream := TJSonStreamObject(pvDataObject);

  //客户端发送文件
  if lvJsonStream.JSon.I['cmdIndex'] = 102 then
  begin
    lvFile := ExtractFilePath(ParamStr(0)) + 'tempFile\';
    ForceDirectories(lvFile);
    lvFile := lvFile + lvJsonStream.JSon.S['file'];
    TMemoryStream(lvJsonStream.Stream).Position := 0;
    TMemoryStream(lvJsonStream.Stream).SaveToFile(lvFile);
  end;


  TLogClientWrapper.logINfo(lvJsonStream.JSon.AsJSon(True));

end;