dntcplistener.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.

{$I DnConfig.inc}
unit DnTcpListener;

interface
uses  Classes, SysUtils, Winsock2, Windows,
      DnConst, DnInterfaces, DnRtl,
      DnAbstractExecutor, DnAbstractLogger,
      DnTcpReactor;

type
  TDnClientTcpConnect = procedure (Context: TDnThreadContext; Channel: IDnChannel) of object;
  TDnCreateTcpChannel = procedure (Context: TDnThreadContext; Socket: TSocket; Addr: TSockAddrIn;
      Reactor: TDnTcpReactor; var ChannelImpl: TDnTcpChannel) of object;

  {$IFDEF ROOTISCOMPONENT}
  TDnTcpListener = class(TComponent, IDnIOResponse, IDnIOErrorValue, IUnknown)
  {$ELSE}
  TDnTcpListener = class(TDnObject, IDnIOResponse, IDnIOErrorValue)
  {$ENDIF}
  protected
    FActive:          Boolean;
    FNagle:           Boolean;
    FSocket:          TSocket;
    FAddress:         String;
    FAddr:            TSockAddrIn;
    FPort:            SmallInt;
    FBackLog:         Integer;
    FReactor:         TDnTcpReactor;
    FExecutor:        TDnAbstractExecutor;
    FLogger:          TDnAbstractLogger;
    FLogLevel:        TDnLogLevel;
    FKeepAlive:       Boolean;
    FOnClientConnect: TDnClientTcpConnect;
    FOnCreateChannel: TDnCreateTcpChannel;
    FGuard:           TDnMutex;
    
    FAcceptSocket:    TSocket;
    FAcceptBuffer:    String;
    FAcceptReceived:  Cardinal;
    FAcceptContext:   TDnReqContext;
    FAcceptChannel:   TDnTcpChannel;
    FAcceptThread:    TDnThreadContext;
    FErrorCode:       Integer;
    
    procedure SetAddress(Address: String);
    procedure SetActive(Value: Boolean);
    function  TurnOn: Boolean;
    function  TurnOff: Boolean;
    procedure CheckSocketError(Code: Cardinal; Msg: String);
    function  DoCreateChannel(Context: TDnThreadContext; Socket: TSocket;
                              Addr: TSockAddrIn): TDnTcpChannel;
    procedure DoLogMessage(S: String);
    procedure RunAcceptEx(Context: TDnThreadContext);
    
    procedure CallHandler(Context: TDnThreadContext);
    function  Channel: IDnIOTrackerHolder;
    procedure StoreError(ErrorCode: Integer);
    {$IFDEF ROOTISCOMPONENT}
    procedure Notification(AComponent: TComponent; Operation: TOperation); override;
    {$ENDIF}
  public
    constructor Create{$IFDEF ROOTISCOMPONENT}(AOwner: TComponent); override{$ENDIF};
    destructor  Destroy; override;
  published
    property Active: Boolean read FActive write SetActive;
    property Port: SmallInt read FPort write FPort;
    property Address: String read FAddress write SetAddress;

    property UseNagle: Boolean read FNagle write FNagle;
    property BackLog: Integer read FBackLog write FBackLog;
    property Reactor: TDnTcpReactor read FReactor write FReactor;
    property Executor: TDnAbstractExecutor read FExecutor write FExecutor;
    property Logger: TDnAbstractLogger read FLogger write FLogger;
    property LogLevel: TDnLogLevel read FLogLevel write FLogLevel;
    property KeepAlive: Boolean read FKeepAlive write FKeepAlive;
    property OnCreateChannel: TDnCreateTcpChannel read FOnCreateChannel write FOnCreateChannel;
    property OnIncoming: TDnClientTcpConnect read FOnClientConnect write FOnClientConnect;
  end;

  TDnAcceptExWrapper = class (TDnObject, IDnIOResponse)
  protected
    FListener: TDnTcpListener;
  public
    constructor Create(Listener: TDnTcpListener);
    destructor Destroy; override;
    procedure CallHandler(Context: TDnThreadContext);
    function Channel: IDnIOTrackerHolder;
  end;

procedure Register;

function AcceptEx(sListenSocket, sAcceptSocket: TSocket; lpOutputBuffer: Pointer; dwReceiveDataLength, dwLocalAddressLength, dwRemoteAddressLength: DWORD; var lpdwBytesReceived: DWORD;  lpOverlapped: POverlapped): BOOL; stdcall;
procedure GetAcceptExSockaddrs(lpOutputBuffer: Pointer; dwReceiveDataLength, dwLocalAddressLength, dwRemoteAddressLength: DWORD;  var LocalSockaddr: PSockAddr; var LocalSockaddrLength: Integer;  var RemoteSockaddr: PSockAddr; var RemoteSockaddrLength: Integer); stdcall;

implementation
function  AcceptEx;               external 'mswsock.dll' name 'AcceptEx';
procedure GetAcceptExSockaddrs;   external 'mswsock.dll' name 'GetAcceptExSockaddrs';

{$IFDEF ROOTISCOMPONENT}
constructor TDnTcpListener.Create(AOwner: TComponent);
begin
  inherited Create(AOwner);
{$ELSE}
constructor TDnTcpListener.Create;
begin
  inherited Create;
{$ENDIF}
  Self._AddRef;
  FActive := False;
  FPort := 7080;
  FAddress := '0.0.0.0';
  FOnCreateChannel := Nil;
  FOnClientConnect := Nil;
  FKeepAlive := False;
  FNagle := True;
  FSocket := INVALID_SOCKET;
  FAcceptSocket := INVALID_SOCKET;
  FBackLog := 5;
  FLogger := Nil;
  FLogLevel := llMandatory;
  FReactor := Nil;
  FExecutor := Nil;
  SetLength(FAcceptBuffer, 64);
  FAcceptReceived := 0;
  FGuard := TDnMutex.Create;
end;

destructor TDnTcpListener.Destroy;
begin
  if Active then
    Active := False;
  inherited Destroy;
end;

procedure TDnTcpListener.StoreError(ErrorCode: Integer);
begin
  FErrorCode := ErrorCode;
end;


{$IFDEF ROOTISCOMPONENT}
procedure TDnTcpListener.Notification(AComponent: TComponent; Operation: TOperation);
begin
  if Operation = opRemove then
  begin
    if AComponent = FExecutor then
      FExecutor := Nil
    else
    if AComponent = FLogger then
      FLogger := Nil
    else
    if AComponent = FReactor then
      FReactor := Nil;
  end;
end;
{$ENDIF}

procedure TDnTcpListener.DoLogMessage(S: String);
begin
  if FLogger<>Nil then
  try
    FLogger.LogMsg(FLogLevel, S);
  except
    ;
  end;
end;

procedure TDnTcpListener.CheckSocketError(Code: Cardinal; Msg: String);
begin
  if (FLogger <> Nil) and (Code = INVALID_SOCKET) then
  try
    FLogger.LogMsg(FLogLevel, Msg);
  except
    ; //suppress exception
  end
end;

procedure TDnTcpListener.SetActive(Value: Boolean);
begin
  FGuard.Acquire;
  if not FActive and Value then
    FActive := TurnOn
  else if FActive and not Value then
    FActive := TurnOff;
  FGuard.Release;
end;

function TDnTcpListener.TurnOn: Boolean;
var TempBool: LongBool;
begin
  FAcceptThread := Nil;
  FSocket := Winsock2.WSASocket(AF_INET, SOCK_STREAM, 0, Nil, 0, WSA_FLAG_OVERLAPPED);
  if FSocket = INVALID_SOCKET then
    raise EDnException.Create(ErrWin32Error, WSAGetLastError(), 'WSASocket');
  FillChar(FAddr, SizeOf(FAddr), 0);
  FAddr.sin_family := AF_INET;
  FAddr.sin_port := htons(FPort);
  FAddr.sin_addr.S_addr := inet_addr(PChar(FAddress));
  CreateIOCompletionPort(FSocket, FReactor.PortHandle, 0, 1);

  //Set SO_REUSEADDR
  TempBool := True;
  SetSockOpt(FSocket, SOL_SOCKET, SO_REUSEADDR, PChar(@TempBool), SizeOf(TempBool));
  //bind socket
  if Bind(FSocket, @FAddr, sizeof(FAddr)) = -1 then
    raise EDnException.Create(ErrWin32Error, WSAGetLastError, Format('Bind failed. Port is %d', [FPort]));
  if Winsock2.Listen(FSocket, FBackLog) = -1 then
    raise EDnException.Create(ErrWin32Error, WSAGetLastError(), Format('Listen failed. Port is %d.', [FPort]));

  FExecutor.PostEvent(TDnAcceptExWrapper.Create(Self));
  Result := True;
end;

procedure TDnTcpListener.RunAcceptEx(Context: TDnThreadContext);
var ResCode: Integer;
begin
  FErrorCode := 0;
  FillChar(FAcceptContext, sizeof(FAcceptContext), 0);
  {$IFDEF ROOTISCOMPONENT}
  FAcceptContext.FRequest := Pointer(IUnknown(Self));
  {$ELSE}
  FAcceptContext.FRequest := Pointer(Self);
  {$ENDIF}
  FAcceptContext.FReqRouting := False;
  if Assigned(FAcceptThread) then
    FAcceptThread.Release;
  if Assigned(Context) then
    Context.Grab;
  FAcceptSocket := Winsock2.WSASocketA(AF_INET, SOCK_STREAM, 0, Nil, 0, WSA_FLAG_OVERLAPPED);
  if FAcceptSocket = INVALID_SOCKET then
    raise EDnException.Create(ErrWin32Error, WSAGetLastError(), 'WSASocket');

  if AcceptEx(FSocket, FAcceptSocket, @FAcceptBuffer[1], 0, sizeof(TSockAddrIn)+16,
              sizeof(TSockAddrIn)+16, FAcceptReceived, POverlapped(@FAcceptContext)) = FALSE then
  begin
    if WSAGetLastError() <> ERROR_IO_PENDING then
    begin
      if Assigned(Context) then
        Context.Release;
      raise EDnException.Create(ErrWin32Error, WSAGetLastError(), 'AcceptEx');
    end;
    InterlockedIncrement(PendingRequests);
  end;
  FAcceptThread := Context;

end;

function  TDnTcpListener.DoCreateChannel(Context: TDnThreadContext; Socket: TSocket; Addr: TSockAddrIn): TDnTcpChannel;
var SockObj: TDnTcpChannel;
begin
  Result := Nil;
  try
    if Assigned(FOnCreateChannel) then
      FOnCreateChannel(Context, Socket, Addr, FReactor, SockObj);
  except
    on E: Exception do
          begin
            DoLogMessage(E.Message);
            SockObj := Nil;
          end;
  end;
  if not Assigned(Result) then
    SockObj := TDnTcpChannel.Create(FReactor, Socket, Addr);

  Result := SockObj;
end;

function TDnTcpListener.TurnOff: Boolean;
begin
  if FSocket <> INVALID_SOCKET then
  begin
    Shutdown(FSocket, SD_BOTH); //yes, I known that SD_BOTH is bad idea... But this is LISTENING socket :)
    CloseSocket(FSocket);
    FSocket := INVALID_SOCKET;
  end;
  Result := False;
end;

procedure TDnTcpListener.SetAddress(Address: String);
var addr: Cardinal;
begin
  addr := inet_addr(PChar(Address));
  if addr <> INADDR_NONE then
  begin
    FAddress := Address;
  end;
end;

procedure TDnTcpListener.CallHandler(Context: TDnThreadContext);
var LocalAddrP, RemoteAddrP: PSockAddr;
    LocalAddr, RemoteAddr: TSockAddrIn;
    LocalAddrLen, RemoteAddrLen: Integer;
    Channel: TDnTcpChannel;
    ResCode: Integer;
begin
  FGuard.Acquire;
  try
    if FActive and (FErrorCode = 0) then
    begin
      LocalAddrLen := Sizeof(TSockAddrIn); RemoteAddrLen := Sizeof(TSockAddrIn);
      ResCode := Winsock2.setsockopt(FAcceptSocket, SOL_SOCKET, SO_UPDATE_ACCEPT_CONTEXT, PChar(@Self.FSocket), sizeof(Self.FSocket));
      GetAcceptExSockaddrs(@FAcceptBuffer[1], 0, sizeof(TSockAddrIn)+16,
        sizeof(TSockAddrIn)+16, LocalAddrP, LocalAddrLen,
        RemoteAddrP, RemoteAddrLen); //}//there was a good idea but it is not working so just KISS :)
      LocalAddr := LocalAddrP^;
      RemoteAddr := RemoteAddrP^;
      Channel := DoCreateChannel(Context, FAcceptSocket, RemoteAddr);
      //post channel to reactor
      Channel._AddRef;
      FReactor.PostChannel(Channel);
      Channel._Release;
      if Assigned(FOnClientConnect) then
      try
        FOnClientConnect(Context, Channel);
      except
        on E: Exception do
          FLogger.LogMsg(FLogLevel, E.Message);
      end;
      InterlockedDecrement(PendingRequests);
      RunAcceptEx(Context);
    end;
  finally
    FGuard.Release;
  end;  
end;

function  TDnTcpListener.Channel: IDnIOTrackerHolder;
begin
  Result := Nil;
end;

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

constructor TDnAcceptExWrapper.Create(Listener: TDnTcpListener);
begin
  inherited Create;
  FRefCount := 0;
  FListener := Listener;
end;

destructor TDnAcceptExWrapper.Destroy;
begin
  inherited Destroy;
end;

procedure TDnAcceptExWrapper.CallHandler(Context: TDnThreadContext);
begin
  FListener.RunAcceptEx(Context);
end;

function TDnAcceptExWrapper.Channel: IDnIOTrackerHolder;
begin
  Result := Nil;
end;


procedure Register;
begin
  {$IFDEF ROOTISCOMPONENT}
  RegisterComponents('DNet', [TDnTcpListener]);
  {$ENDIF}
end;

end.