dntcprequests.pas

一个国外比较早的IOCP控件

// The contents of this file are used with permission, subject to
// the Mozilla Public License Version 1.1 (the "License"); you may
// not use this file except in compliance with the License. You may
// obtain a copy of the License at
// http://www.mozilla.org/MPL/MPL-1.1.html
//
// Software distributed under the License is distributed on an
// "AS IS" basis, WITHOUT WARRANTY OF ANY KIND, either express or
// implied. See the License for the specific language governing
// rights and limitations under the License.
unit DnTcpRequests;
interface
uses
  Classes, SysUtils, Windows, Winsock, Winsock2,
  DnTcpReactor, DnConst, DnInterfaces, DnRtl;

type

  IDnTcpCloseHandler = interface
  ['{AB1279A1-BBC9-11d5-BDB9-0000212296FE}']
    procedure DoClose(Context: TDnThreadContext; Channel: IDnChannel; Key: Pointer);
    procedure DoCloseError(Context: TDnThreadContext; Channel: IDnChannel; Key: Pointer;
                                  ErrorCode: Cardinal);
  end;

  TDnTcpCloseRequest = class (TDnTcpRequest)
  protected
    FWSABuf:      WSABUF;
    FTempBuffer:  String;
    FRead:        Cardinal;
    FFlags:       Cardinal;
    FHandler:     IDnTcpCloseHandler;
    FBrutal:      Boolean;
    procedure     SetTransferred(Transferred: Cardinal); override;
    
    //IDnIORequest
    procedure Execute; override;
    function  IsComplete: Boolean; override;
    procedure ReExecute; override;
    function  RequestType: TDnIORequestType; override;
    function  IsCPUNeeded: Boolean; override;

    //IDnIOResponse
    procedure CallHandler(Context: TDnThreadContext); override;

  public
    constructor Create(Channel: IDnChannel; Key: Pointer; Handler: IDnTcpCloseHandler;
                        Brutal: Boolean = False);
    destructor  Destroy; override;
  end;

  IDnTcpReadHandler = interface
  ['{AB1279A2-BBC9-11d5-BDB9-0000212296FE}']
    procedure DoRead(Context: TDnThreadContext; Channel: IDnChannel; Key: Pointer;
                                  Buf: PChar; BufSize: Cardinal);
    procedure DoReadError(Context: TDnThreadContext; Channel: IDnChannel; Key: Pointer;
                                  ErrorCode: Cardinal);
    procedure DoReadClose(Context: TDnThreadContext; Channel: IDnChannel; Key: Pointer);
  end;

  TDnTcpReadRequest = class (TDnTcpRequest)
  protected
    FWSABuf:  WSABUF;
    FRead:    Cardinal;
    FToRead:  Cardinal;
    FFlags:   Cardinal;
    FHandler: IDnTcpReadHandler;
    FMustAll: Boolean;
    FStrBuf:  String;
    
    procedure SetTransferred(Transferred: Cardinal); override;

    //IDnIORequest
    procedure Execute; override;
    function IsComplete: Boolean; override;
    procedure ReExecute; override;
    function  RequestType: TDnIORequestType; override;
    function  IsCPUNeeded: Boolean; override;
    //IDnIOResponse
    procedure CallHandler(Context: TDnThreadContext); override;
  public
    constructor Create( Channel: IDnChannel; Key: Pointer;
                        Handler: IDnTcpReadHandler; Buf: PChar;
                        BufSize: Cardinal; MustAll:  Boolean = True);// overload;
    constructor CreateString( Channel:  IDnChannel; Key: Pointer;
                        Handler: IDnTcpReadHandler; BufSize: Integer;
                        MustAll:  Boolean = True); 
    destructor Destroy; override;
    procedure Init( Channel: IDnChannel; Key: Pointer;
                    Handler: IDnTcpReadHandler; Buf: PChar;
                    BufSize: Cardinal; MustAll: Boolean = True);

  end;

  IDnTcpLineHandler = interface
  ['{AB1279A4-BBC9-11d5-BDB9-0000212296FE}']
    procedure DoLine( Context: TDnThreadContext; Channel: IDnChannel; Key: Pointer;
                      ReceivedLine: String; EolFound: Boolean );
    procedure DoLineError(Context: TDnThreadContext; Channel: IDnChannel; Key: Pointer;
                          ErrorCode: Cardinal);
    procedure DoLineClose(Context: TDnThreadContext; Channel: IDnChannel; Key: Pointer);
  end;

  TDnTcpLineRequest = class (TDnTcpRequest)
  protected
    FWSABuf:    WSABUF;
    FRead:      Cardinal;
    FToRead:    Integer;
    FWasRead:   Integer;
    FTotalWasRead:
                Integer;
    FFlags:     Cardinal;
    FHandler:   IDnTcpLineHandler;
    FMaxSize:   Integer;
    FRecv:      String;
    FEolFound:  Boolean;
    FEolSign:   PChar;

    FBufInitialSize: Integer;
    FBufGranularity: Integer;
    
    function  CheckForEol(Line: PChar; Len: Integer): Integer;
    procedure SetTransferred(Transferred: Cardinal); override;
    function  IssueWSARecv( s : TSocket; lpBuffers : LPWSABUF; dwBufferCount : DWORD; var lpNumberOfBytesRecvd : DWORD; var lpFlags : DWORD;
              lpOverlapped : LPWSAOVERLAPPED; lpCompletionRoutine : LPWSAOVERLAPPED_COMPLETION_ROUTINE ): Integer; stdcall;
    procedure Reset(Channel: IDnChannel; Key: Pointer;
                        Handler: IDnTcpLineHandler; MaxSize: Cardinal);
    //IDnIORequest
    procedure Execute; override;
    function  IsComplete: Boolean; override;
    procedure ReExecute; override;
    function  RequestType: TDnIORequestType; override;
    function  IsCPUNeeded: Boolean; override;

    //IDnIOResponse
    procedure CallHandler(Context: TDnThreadContext); override;
  public
    constructor Create( Channel: IDnChannel; Key: Pointer;
                        Handler: IDnTcpLineHandler; MaxSize: Cardinal );
    destructor Destroy; override;
  end;

  IDnTcpWriteHandler = interface
  ['{AB1279A5-BBC9-11d5-BDB9-0000212296FE}']
    procedure DoWrite(Context: TDnThreadContext; Channel: IDnChannel; Key: Pointer;
                      Buf: PChar; BufSize: Cardinal);
    procedure DoWriteError( Context: TDnThreadContext; Channel: IDnChannel; Key: Pointer;
                            ErrorCode: Cardinal );
  end;


  TDnTcpWriteRequest = class (TDnTcpRequest)
  protected
    FWSABuf:        WSABUF;
    FWritten:       Cardinal;
    FToWrite:       Cardinal;
    FFlags:         Cardinal;
    FHandler:       IDnTcpWriteHandler;
    FTempStorage:   String;

    procedure SetTransferred(Transferred: Cardinal); override;
    //IDnIORequest
    procedure Execute; override;
    function  IsComplete: Boolean; override;
    procedure ReExecute; override;
    function  RequestType: TDnIORequestType; override;
    function  IsCPUNeeded: Boolean; override;

    //IDnIOResponse
    procedure CallHandler(Context: TDnThreadContext); override;
  public
    constructor Create( Channel: IDnChannel; Key: Pointer; Handler: IDnTcpWriteHandler;
                        Buf: PChar; BufSize: Cardinal); overload;
    constructor CreateString( Channel: IDnChannel; Key: Pointer; Handler: IDnTcpWriteHandler;
                        Buf: String); overload;
    destructor Destroy; override;
  end;



implementation
var
  CRLFZero:PChar = #13#10#0;


//----------------------------------------------------------------------------
//----------------------------------------------------------------------------

constructor TDnTcpCloseRequest.Create(Channel: IDnChannel; Key: Pointer;
                                      Handler: IDnTcpCloseHandler; Brutal: Boolean);
begin
  inherited Create(Channel, Key);
  FHandler := Handler;
  SetLength(FTempBuffer, 1024);
  FWSABuf.len := 1024;
  FWSABuf.buf := @FTempBuffer[1];
  FBrutal := Brutal;
end;

function TDnTcpCloseRequest.RequestType: TDnIORequestType;
begin
  if FBrutal then
    Result := rtBrutalClose
  else
    Result := rtClose;
end;

procedure TDnTcpCloseRequest.SetTransferred(Transferred: Cardinal);
begin
  FRead := Transferred;
  FWSABuf.buf := PChar(FTempBuffer);
  FWSABuf.len := 1024;
end;

procedure TDnTcpCloseRequest.Execute;
var ResCode: Integer;
    ChannelImpl: TDnTcpChannel;
begin
  inherited Execute;
  ChannelImpl := TDnTcpChannel.CheckImpl(FChannel);
  Winsock2.shutdown(ChannelImpl.SocketHandle, SD_SEND); //disable sending
  InterlockedIncrement(PendingRequests);
  if not FBrutal then
  begin
    FRead := 0;
    ResCode := Winsock2.WSARecv(ChannelImpl.SocketHandle, @FWSABuf, 1,  FRead, FFlags, @FContext, Nil);
    if ResCode <> 0 then
    begin
      ResCode := WSAGetLastError;
      if ResCode <> WSA_IO_PENDING then
        Self.PostError(ResCode, 0);
    end;
  end else
    PostQueuedCompletionStatus(ChannelImpl.Reactor.PortHandle, 0,
                                Cardinal(Pointer(ChannelImpl)), @FContext);
end;

procedure TDnTcpCloseRequest.ReExecute;
begin
  Execute;
end;

function TDnTcpCloseRequest.IsComplete: Boolean;
var ChannelImpl: TDnTcpChannel;
begin
  inherited IsComplete;
  ChannelImpl := TDnTcpChannel.CheckImpl(FChannel);
  Result := (FRead = 0) or (FErrorCode <> 0);
  if Result then
  begin
    ChannelImpl.CloseSocketHandle;
  end;
end;

function TDnTcpCloseRequest.IsCPUNeeded: Boolean;
begin
  Result := False;
end;

procedure TDnTcpCloseRequest.CallHandler(Context: TDnThreadContext);
var ChannelImpl: TDnTcpChannel;
begin
  try
    ChannelImpl := TDnTcpChannel.CheckImpl(FChannel);
    if FErrorCode <> 0 then
      FHandler.DoCloseError(Context, FChannel, FKey, FErrorCode)
    else
      FHandler.DoClose(Context, FChannel, FKey);
    //ChannelImpl.CloseSocketHandle;
    //drop channel
    ChannelImpl.Reactor.RemoveChannel(FChannel);
  finally
    //InterlockedDecrement(PendingRequests);
  end;
end;


destructor  TDnTcpCloseRequest.Destroy;
begin
  FHandler := Nil;
  inherited Destroy;
end;
//-------------------------------------------------------------------------------------------------

constructor TDnTcpReadRequest.Create( Channel: IDnChannel; Key: Pointer;
                                      Handler: IDnTcpReadHandler; Buf: PChar;
                                      BufSize: Cardinal; MustAll: Boolean = True );
begin
  inherited Create(Channel, Key);
  SetLength(FStrBuf, 0);
  Init(Channel, Key, Handler, Buf, BufSize, MustAll);
end;

constructor TDnTcpReadRequest.CreateString( Channel: IDnChannel; Key: Pointer;
                                      Handler: IDnTcpReadHandler; BufSize: Integer;
                                      MustAll: Boolean = True );
begin
  inherited Create(Channel, Key);
  SetLength(FStrBuf, BufSize);
  Init(Channel, Key, Handler, @FStrBuf[1], Length(FStrBuf), MustAll);
end;

procedure TDnTcpReadRequest.Init(Channel: IDnChannel; Key: Pointer;
                                  Handler: IDnTcpReadHandler; Buf: PChar;
                                  BufSize: Cardinal; MustAll: Boolean = True);
begin
  FWSABuf.Len := BufSize;
  FWSABuf.Buf := Buf;
  FRead := 0;
  FToRead := BufSize;
  FFlags := 0;
  //FTotalSize := BufSize;
  FStartBuffer := @Buf;
  FHandler := Handler;
  FMustAll := MustAll;
end;

destructor TDnTcpReadRequest.Destroy;
begin
  FHandler := Nil;
  inherited Destroy;
end;

function  TDnTcpReadRequest.RequestType: TDnIORequestType;
begin
  Result := rtRead;
end;

procedure TDnTcpReadRequest.SetTransferred(Transferred: Cardinal);
begin
  FRead := Transferred;
  Inc(FWSABuf.buf, FRead);
  Dec(FWSABuf.len, FRead);
end;

procedure TDnTcpReadRequest.Execute;
var ChannelImpl: TDnTcpChannel;
    ResCode: Integer;
begin
  inherited Execute;
  ChannelImpl := TDnTcpChannel.CheckImpl(FChannel);
  //check the channel read cache
  FRead := ChannelImpl.ExtractFromCache(FWSABuf.buf, FToRead);
  //inc pending requests count
  InterlockedIncrement(PendingRequests);
  if FRead = FToRead then
    PostQueuedCompletionStatus(ChannelImpl.Reactor.PortHandle, FRead,
                                Cardinal(Pointer(ChannelImpl)), @FContext)
  else
  begin //not read yet...
    Inc(FWSABuf.buf, FRead);
    Dec(FWSABuf.len, FRead);
    ResCode := Winsock2.WSARecv(ChannelImpl.SocketHandle, @FWSABuf, 1,  FRead, FFlags, @FContext, Nil);
    //ResCode := Integer(ReadFileEx(ChannelImpl.SocketHandle, @FWSABuf, FRead, @FContext, Nil));
    if ResCode <> 0 then
    begin
      ResCode := WSAGetLastError;
      if ResCode <> WSA_IO_PENDING then
        Self.PostError(ResCode, FRead)