dntcpconnect.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 DnTcpConnect;
interface
uses
  Windows, Classes, Winsock2, SysUtils, Contnrs,
  DnInterfaces, DnTcpReactor, DnRtl, DnAbstractExecutor, DnAbstractLogger,
  DnConst;

type
  IDnTcpConnectHandler = interface
  ['{8E98EE19-A73C-4981-90D3-F0D544ED8085}']
    procedure DoConnect(Context: TDnThreadContext; Channel: IDnChannel; Key: Pointer;
                        const IP: String; Port: Word);
    procedure DoConnectError(Context: TDnThreadContext; Channel: IDnChannel; Key: Pointer;
                              ErrorCode: Cardinal);
  end;


  TDnTcpConnectRequest = class;
  TDnTcpConnectWatcherThread = class;
  
  TDnTcpConnectWatcher = class(TObject)
  protected
    FConnects:      TObjectList;
    FNewConnects:   TObjectList;
    FGuard:         TDnMutex;
    FThreadGuard:   TDnMutex;
    FReactor:       TDnTcpReactor;
    FExecutor:      TDnAbstractExecutor;
    FLogger:        TDnAbstractLogger;
    FLogLevel:      TDnLogLevel;
    FActive:        Boolean;
    FThread:        TDnTcpConnectWatcherThread;
    
    procedure   SetActive(Value: Boolean);
    function    TurnOn: Boolean;
    function    TurnOff: Boolean;
    procedure   ConnectFinished(Request: TDnTcpConnectRequest);
    procedure   AbortRequest(Request: TDnTcpConnectRequest);
    procedure   AbortRequests;

  public
    constructor Create;
    destructor  Destroy; override;
    procedure   MakeConnect(Channel: IDnChannel; Key: Pointer; TimeOut: Cardinal;
                            Handler: IDnTcpConnectHandler);

    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    Active:     Boolean read FActive write SetActive;
  end;

  TDnTcpConnectWatcherThread = class (TDnThread)
  protected
    FWatcher: TDnTcpConnectWatcher;
    procedure ThreadRoutine; override;
    procedure CreateContext; override;
    procedure DestroyContext; override;
  public
    constructor Create(Watcher: TDnTcpConnectWatcher);
    destructor Destroy; override;
  end;

  TDnTcpConnectRequest = class (TDnObject, IDnIOResponse)
  protected
    FHandler:         IDnTcpConnectHandler;
    FConnectSignal:   THandle;
    FChannel:         IDnChannel;
    FKey:             Pointer;
    FErrorCode:       Integer;
    FConnectWatcher:  TDnTcpConnectWatcher;
    FTimeOut:         Cardinal;
    FStartTick:       Cardinal;
    
    //IDnIOResponse
    procedure CallHandler(Context: TDnThreadContext); virtual;
    function  Channel: IDnIOTrackerHolder;

    function IsComplete: Boolean;
  public
    constructor Create( Channel: IDnChannel; Key: Pointer; TimeOut: Cardinal;
                        Handler: IDnTcpConnectHandler;
                        ConnectWatcher: TDnTcpConnectWatcher);
    destructor  Destroy; override;
    procedure Execute;
  end;


implementation

constructor TDnTcpConnectWatcher.Create;
begin
  FConnects := Nil;
  FNewConnects := Nil;
  FReactor := Nil;
  FExecutor := Nil;
  FLogLevel := llMandatory;
  FLogger := Nil;
  FThread := Nil;
  FGuard := TDnMutex.Create;
  FThreadGuard := TDnMutex.Create;
end;

destructor  TDnTcpConnectWatcher.Destroy;
begin
  SetActive(False);
  FreeAndNil(FThreadGuard);
  FreeAndNil(FGuard);
  inherited Destroy;
end;

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

function TDnTcpConnectWatcher.TurnOn: Boolean;
begin
  if (FReactor = Nil) or (FLogger = Nil) or (FExecutor = Nil) then
    raise EDnException.Create(ErrInvalidConfig, 0, 'TDnTcpConnectWatcher');
  FConnects := TObjectList.Create(False);
  FNewConnects := TObjectList.Create(False);
  //FGuard := TDnMutex.Create;
  FThread := TDnTcpConnectWatcherThread.Create(Self);
  FThread.Resume;
  Result := True;
end;

function TDnTcpConnectWatcher.TurnOff: Boolean;
begin
  AbortRequests;

  if not FExecutor.Active then
  begin
    if Assigned(FConnects) then
      FConnects.OwnsObjects := True;
    if Assigned(FNewConnects) then
      FNewConnects.OwnsObjects := True;
  end;

  if Assigned(FConnects) then
    FreeAndNil(FConnects);
  if Assigned(FNewConnects) then
    FreeAndNil(FNewConnects);

  //FreeAndNil(FGuard);

  Result := False;
end;

procedure TDnTcpConnectWatcher.ConnectFinished(Request: TDnTcpConnectRequest);
var ChannelImpl: TDnTcpChannel;
    networkEvents: TWSANetworkEvents;
begin
  if Request = Nil then
    raise EDnException.Create(ErrInvalidParameter, 0, 'TDnTcpConnectWatcher.ConnectFinished');
  ChannelImpl := TDnTcpChannel.CheckImpl(Request.FChannel);
  //check for timeout
  if (Request.FTimeOut <> 0) and (CurrentTimeFromLaunch() - Request.FStartTick > Request.FTimeOut) then
  begin
    Request.FErrorCode := WSAETIMEDOUT;
    ChannelImpl._AddRef;
    WSAEventSelect(ChannelImpl.SocketHandle, Request.FConnectSignal, 0);
    FReactor.PostChannel(ChannelImpl);
    FExecutor.PostEvent(Request);
  end;
  if (Winsock2.WSAEnumNetworkEvents(ChannelImpl.SocketHandle, Request.FConnectSignal,
                                @networkEvents) = SOCKET_ERROR) then
    Request.FErrorCode := WSAGetLastError
  else
    Request.FErrorCode := networkEvents.iErrorCode[FD_CONNECT_BIT];
  WSAEventSelect(ChannelImpl.SocketHandle, Request.FConnectSignal, 0);
  //ChannelImpl._AddRef;
  //if Request.FErrorCode = 0 then
  FReactor.PostChannel(ChannelImpl);
  FExecutor.PostEvent(Request);
end;

procedure TDnTcpConnectWatcher.MakeConnect(Channel: IDnChannel; Key: Pointer;
    TimeOut: Cardinal; Handler: IDnTcpConnectHandler);
var ConnectRequest: TDnTcpConnectRequest;
begin
  if (Channel = Nil) or (Handler = Nil) then
    raise EDnException.Create(ErrInvalidParameter, 0, 'TDnTcpConnectReactor.MakeConnect');
  FGuard.Acquire;
  try
    ConnectRequest := TDnTcpConnectRequest.Create(Channel, Key, TimeOut, Handler, Self);
    ConnectRequest.Execute;
    InterlockedIncrement(PendingRequests);
    if ConnectRequest.FErrorCode <> WSAEWOULDBLOCK then
    begin
      if ConnectRequest.FErrorCode = 0 then
        FReactor.PostChannel(Channel); //bind new channel to IOCP //channel is already added
      FExecutor.PostEvent(ConnectRequest); //post event to executor
    end else
    begin
      ConnectRequest.FErrorCode := 0;
      FThreadGuard.Acquire;
      FNewConnects.Add(ConnectRequest);
      FThreadGuard.Release;
    end;
  finally
    FGuard.Release;
  end;
end;

procedure TDnTcpConnectWatcher.AbortRequest(Request: TDnTcpConnectRequest);
var ChannelImpl: TDnTcpChannel;
begin
  Request.FErrorCode := WSAECONNABORTED;
  if FReactor.Active then
  begin
    ChannelImpl := TDnTcpChannel.CheckImpl(Request.FChannel);
    ChannelImpl._AddRef;
    FReactor.PostChannel(ChannelImpl);
  end;
  if FExecutor.Active then
    FExecutor.PostEvent(Request);
end;

procedure TDnTcpConnectWatcher.AbortRequests;
var i: Integer;
    ChannelImpl: TDnTcpChannel;
    Request: TDnTcpConnectRequest;
begin
  FGuard.Acquire;
  FreeAndNil(FThread);
  for i:=0 to FConnects.Count-1 do
    AbortRequest(TDnTcpConnectRequest(FConnects[i]));
  for i:=0 to FNewConnects.Count-1 do
    AbortRequest(TDnTcpConnectRequest(FNewConnects[i]));
  FGuard.Release;
end;

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

constructor TDnTcpConnectWatcherThread.Create(Watcher: TDnTcpConnectWatcher);
begin
  inherited Create;
  if Watcher = Nil then
    raise EDnException.Create(ErrInvalidParameter, 0, 'TDnTcpConnectWatcherThread.Create');
  FWatcher := Watcher;
end;

destructor TDnTcpConnectWatcherThread.Destroy;
begin
  inherited Destroy;
end;

procedure TDnTcpConnectWatcherThread.CreateContext;
begin
end;

procedure TDnTcpConnectWatcherThread.DestroyContext;
begin
end;

procedure TDnTcpConnectWatcherThread.ThreadRoutine;
var resCode, i, connCount, connIndex: Integer;
    connects: array[0..63] of THandle;
    request: TDnTcpConnectRequest;
begin
  connIndex := 0;
  while not Terminated do
  begin
    if connIndex < FWatcher.FConnects.Count then
    begin//check a new connects
      //fill events array
      connCount := FWatcher.FConnects.Count - connIndex;
      if connCount > 63 then
        connCount := 63;
      i:=connIndex;
      while i <= connIndex+connCount-1 do
      begin
        Request := TDnTcpConnectRequest(FWatcher.FConnects[i]);
        if (Request.FTimeOut <> 0) and (CurrentTimeFromLaunch() - Request.FStartTick > Request.FTimeOut) then
        begin
          FWatcher.FConnects.Delete(i);
          FWatcher.ConnectFinished(request);
        end;
        connects[i-connIndex] := TDnTcpConnectRequest(FWatcher.FConnects[i]).FConnectSignal;
        Inc(i);
      end;

      //check events array for fired event
      resCode := WSAWaitForMultipleEvents(connCount, @connects, False, 0, False);
      if (resCode >= WSA_WAIT_EVENT_0) and (resCode < WSA_WAIT_EVENT_0 + connCount) then
      begin //have connection
        request := TDnTcpConnectRequest(FWatcher.FConnects[connIndex+resCode-WSA_WAIT_EVENT_0]);
        FWatcher.FConnects.Delete(connIndex+resCode-WSA_WAIT_EVENT_0);
        FWatcher.ConnectFinished(request);
      end else
        inc(connIndex, connCount);
    end else
    begin
      connIndex := 0;
      Sleep(1);
    end;
    //check for new requests
    FWatcher.FThreadGuard.Acquire;
    if FWatcher.FNewConnects.Count > 0 then
    begin
      for i:=0 to FWatcher.FNewConnects.Count-1 do
        FWatcher.FConnects.Add(FWatcher.FNewConnects[i]);
      FWatcher.FNewConnects.Clear;
    end;
    FWatcher.FThreadGuard.Release;
  end;
end;


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

constructor TDnTcpConnectRequest.Create(Channel: IDnChannel; Key: Pointer; TimeOut: Cardinal;
                                        Handler: IDnTcpConnectHandler;
                                        ConnectWatcher: TDnTcpConnectWatcher);
begin
  //check the incoming parameters
  if (Channel = Nil) or (Handler = Nil) or (ConnectWatcher = Nil) then
    raise EDnException.Create(ErrInvalidParameter, 0, 'TDnTcpConnectRequest.Create');
  FChannel := Channel;
  FKey := Key;
  FHandler := Handler;
  FConnectSignal := WSACreateEvent();
  FConnectWatcher := ConnectWatcher;
  FRefCount := 0;
  FErrorCode := 0;
  FTimeOut := TimeOut;
  FStartTick := 0;
end;


procedure TDnTcpConnectRequest.Execute;
var ChannelImpl: TDnTcpChannel;
    ResCode: Integer;
begin
  ChannelImpl := TDnTcpChannel.CheckImpl(FChannel);
  //9x/NT/2000 implementation (XP must use ConnectEx instead)
  FStartTick := CurrentTimeFromLaunch();
  Winsock2.WSAEventSelect(ChannelImpl.SocketHandle, FConnectSignal, FD_CONNECT);
  ResCode := WSAConnect(ChannelImpl.SocketHandle, ChannelImpl.RemoteAddrPtr,
          SizeOf(TSockAddrIn), Nil, Nil, Nil, Nil);
  if ResCode <> 0 then
    FErrorCode := WSAGetLastError;
end;

procedure TDnTcpConnectRequest.CallHandler(Context: TDnThreadContext);
begin
  if FErrorCode = 0 then
    FHandler.DoConnect(Context, FChannel, FKey, (FChannel as IDnChannel).RemoteAddr,
      (FChannel as IDnChannel).RemotePort)
  else
    if FErrorCode <> WSAEWOULDBLOCK then
      FHandler.DoConnectError(Context, FChannel, FKey, FErrorCode);
end;

function TDnTcpConnectRequest.IsComplete: Boolean;
begin
  Result := FErrorCode <> WSA_IO_PENDING;
end;

function  TDnTcpConnectRequest.Channel: IDnIOTrackerHolder;
begin
  Result := FChannel as IDnIOTrackerHolder;
end;

destructor  TDnTcpConnectRequest.Destroy;
begin
  CloseHandle(FConnectSignal);
  inherited Destroy;
end;

end.