dnudpreactor.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 DnUDPReactor;
interface
uses
  SysUtils, contnrs, Windows, Classes, WinSock2,
  DnRtl, DnConst, DnInterfaces, DnAbstractExecutor, DnAbstractLogger;

type

  TDnUdpReactor = class;
  
  TDnUdpReactorThread = class;
  TUdpQueryContext = record
    Overlapped: TOverlapped;
    Request: Pointer;
  end;


  PUdpQueryContext = ^TUdpQueryContext;

  IDnUdpSocket = interface
  end;

  TDnUdpSocket = class (TDnObject, IDnUdpSocket, IDnImplementation)
  protected
    FSocket:   TSocket;
    FReactor:  TDnUDPReactor;
    //IUnknown
    function  GetImplementation: Pointer;
  public
    constructor Create; overload;
    constructor Create(Port: Word; const IP: String); overload;
    destructor  Destroy; override;
    class function CheckImpl(obj: IUnknown): TDnUdpSocket;
    property SocketHandle: TSocket read FSocket;
    property Reactor: TDnUDPReactor read FReactor write FReactor;
  end;

  TDnUdpRequest = class (TDnObject, IDnIORequest, IDnIOResponse)
  protected
    FContext:   TUdpQueryContext;
    FErrorCode: Integer;
    FWSABuf:    WSABuf;
    FKey:       Pointer;
    FSocket:    IDnUDPSocket;
    FBuffer:    PChar;
    FBufSize:   Cardinal;
    
    //IDnIORequest
    procedure Execute; virtual;
    function  IsComplete: Boolean; virtual;
    procedure ReExecute; virtual; abstract;
    function  RequestType: TDnIORequestType; virtual; abstract;
    function  IsCPUNeeded: Boolean; virtual; abstract;

    //IDnIOResponse
    function  Channel: IDnIOTrackerHolder;
    procedure CallHandler(Context: TDnThreadContext); virtual; abstract;


    procedure SetTransferred(Transferred: Cardinal); virtual; abstract;
    procedure PostError;
  public
    constructor Create(Sock: IDnUdpSocket; Key: Pointer);
    destructor Destroy; override;
  end;

{$IFDEF ROOTISCOMPONENT}
  TDnUdpReactor = class (TComponent)
{$ELSE}
  TDnUdpReactor = class (TObject)
{$ENDIF}
  protected
    FExecutor:    TDnAbstractExecutor;
    FLogger:      TDnAbstractLogger;
    FLogLevel:    TDnLogLevel;
    FActive:      Boolean;
    FGuard:       TDnMutex;
    FRequestList: TObjectList;
    FPort:        THandle;
    FThread:      TDnUdpReactorThread;
    FRequestID:   Cardinal;

    function  TurnOn: Boolean;
    function  TurnOff: Boolean;
    procedure SetActive(Value: Boolean);
    {$IFDEF ROOTISCOMPONENT}
    procedure Notification(AComponent: TComponent; Operation: TOperation); override;
    {$ENDIF}

  public
    constructor Create {$IFDEF ROOTISCOMPONENT}(AOwner: TComponent); override {$ENDIF};
    destructor  Destroy; override;

    function   CreateUdpSocket: IDnUdpSocket;
    function   CreateBoundUdpSocket(Port: Word; const IP: String): IDnUdpSocket;
  published
    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;
    property  PortHandle: THandle read FPort write FPort;
  end;

  TDnUdpReactorThread = class(TDnThread)
  protected
    FReactor: TDnUdpReactor;

    procedure ThreadRoutine; override;
    procedure ParseIONotification(Transferred: Cardinal; Overlapped: POverlapped);
    procedure ParseIOError(Overlapped: POverlapped);
    procedure CreateContext; override;
    procedure DestroyContext; override;

  public
    constructor Create(Reactor: TDnUdpReactor);
    destructor Destroy; override;
  end;

procedure Register;

implementation

constructor TDnUdpSocket.Create;
begin
  inherited Create;
  FSocket := Winsock2.WSASocket(AF_INET, SOCK_DGRAM, IPPROTO_UDP, Nil, 0, WSA_FLAG_OVERLAPPED);
end;

constructor TDnUdpSocket.Create(Port: Word; const IP: String);
var SockAddr: TSockAddr;
    ResCode:  Integer;
begin
  inherited Create;
  FSocket := Winsock2.WSASocket(AF_INET, SOCK_DGRAM, IPPROTO_UDP, Nil, 0, WSA_FLAG_OVERLAPPED);
  FillChar(SockAddr, sizeof(SockAddr), 0);
  SockAddr.sin_family := AF_INET;
  SockAddr.sin_port := Winsock2.htons(Port);
  SockAddr.sin_addr.S_addr := Winsock2.inet_addr(PChar(IP));
  ResCode := Winsock2.Bind(FSocket, @SockAddr, sizeof(SockAddr));
  if ResCode <> 0 then
    raise EDnException.Create(ErrWin32Error, WSAGetLastError(), 'Bind');
end;

destructor TDnUdpSocket.Destroy;
begin
  Winsock2.closesocket(FSocket);
  inherited Destroy;
end;

class function TDnUdpSocket.CheckImpl(obj: IUnknown): TDnUdpSocket;
var impl: IDnImplementation;
begin
  Result := Nil;
  if IDnUdpSocket(obj) <> Nil then
  begin
    impl := obj as IDnImplementation;
    if impl <> Nil then
      Result := TDnUdpSocket(impl.GetImplementation())
  end;
end;

function  TDnUdpSocket.GetImplementation: Pointer;
begin
  Result := Pointer(Self);
end;


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

constructor TDnUdpRequest.Create(Sock: IDnUdpSocket; Key: Pointer);
begin
  inherited Create;
  FContext.Request := Self;
  FErrorCode := 0;
  FillChar(FContext.Overlapped, SizeOf(FContext.Overlapped), 0);
  FRefCount := 0;
  FSocket := Sock;
end;


destructor TDnUdpRequest.Destroy;
begin
  inherited Destroy;
end;

procedure TDnUdpRequest.Execute;
begin
  ;
end;

procedure TDnUdpRequest.PostError;
var SockImpl: TDnUdpSocket;
begin
  SockImpl := TDnUdpSocket.CheckImpl(FSocket);
  FErrorCode := WSAGetLastError();
  if LongBool(PostQueuedCompletionStatus(SockImpl.Reactor.PortHandle, 0,
                                  0, @FContext )) = False
  then
    raise EDnException.Create(ErrWin32Error, GetLastError(), 'PostQueuedCompletionStatus');
end;

function  TDnUdpRequest.IsComplete: Boolean;
begin
  Result := True;
end;

function  TDnUdpRequest.Channel: IDnIOTrackerHolder;
begin
  Result := Nil;
end;
//--------------------------------------------------------------------
//--------------------------------------------------------------------

constructor TDnUdpReactorThread.Create(Reactor: TDnUdpReactor);
begin
  inherited Create;
  if Reactor = Nil then
    raise EDnException.Create(ErrInvalidParameter, 0, 'TDnUdpReactorThread.Create');
  FReactor := Reactor;
  Resume;
end;

destructor TDnUdpReactorThread.Destroy;
begin
  inherited Destroy;
end;

procedure TDnUdpReactorThread.ThreadRoutine;
var Transferred, Key: Cardinal;
    Overlapped: POverlapped;
    ResCode: LongBool;
begin
  while not Terminated do
  begin
    ResCode := GetQueuedCompletionStatus(FReactor.FPort, Transferred, Key, overlapped, INFINITE);
    if (ResCode = False) and (Overlapped <> Nil) then
      ParseIOError(Overlapped)
    else if (ResCode = True) and (Overlapped <> Nil) then
      ParseIONotification(Transferred, Overlapped)
    else if (Transferred = 0) and (Overlapped = Nil) and (Key = 0) then
      break;
  end;
end;

procedure TDnUdpReactorThread.ParseIONotification(Transferred: Cardinal; Overlapped: POverlapped);
var Context: PUdpQueryContext;
    Request: TDnUdpRequest;
begin
  Context := PUdpQueryContext(Overlapped);
  Request := TDnUdpRequest(Context^.Request);
  Request.SetTransferred(Transferred);
  if Request.IsComplete then
    FReactor.FExecutor.PostEvent(Request)
  else
    Request.ReExecute;
end;

procedure TDnUdpReactorThread.ParseIOError(Overlapped: POverlapped);
var Context: PUdpQueryContext;
    Request: TDnUdpRequest;
begin
  Context := PUdpQueryContext(Overlapped);
  Request := TDnUdpRequest(Context^.Request);
  Request.FErrorCode := Winsock2.WSAGetLastError();
  FReactor.FExecutor.PostEvent(Request);
end;

procedure TDnUdpReactorThread.CreateContext;
begin
  //CurrentContext := Nil;
end;

procedure TDnUdpReactorThread.DestroyContext;
begin
  //CurrentContext := Nil;
end;
//-------------------------------------------------------------------
//-------------------------------------------------------------------

constructor TDnUdpReactor.Create{$IFDEF ROOTISCOMPONENT}(AOwner: TComponent){$ENDIF};
begin
  inherited Create{$IFDEF ROOTISCOMPONENT}(AOwner){$ENDIF};
  FExecutor := Nil;
  FLogger := Nil;
  FLogLevel := llMandatory;
  FActive := False;
  FGuard := TDnMutex.Create;
  FThread := Nil;
  FRequestList := Nil;
end;

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

procedure TDnUdpReactor.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 TDnUdpReactor.CreateUdpSocket: IDnUdpSocket;
var UdpSocket: TDnUdpSocket;
begin
  UdpSocket := TDnUdpSocket.Create;
  UdpSocket.Reactor := Self;
  CreateIOCompletionPort(UdpSocket.SocketHandle, FPort, 0, 1);
end;

function TDnUdpReactor.CreateBoundUdpSocket(Port: Word; const IP: String): IDnUdpSocket;
var UdpSocket: TDnUdpSocket;
begin
  UdpSocket := TDnUdpSocket.Create(Port, IP);
  CreateIOCompletionPort(UdpSocket.SocketHandle, FPort, 0, 1);
  Result := UdpSocket;
end;

function TDnUdpReactor.TurnOn: Boolean;
begin
  if (FLogger = Nil) or (FExecutor = Nil) then
    raise EDnException.Create(ErrInvalidConfig, 0, 'TDnUdpReactor.TurnOn');

  FPort := CreateIOCompletionPort(INVALID_HANDLE_VALUE, 0, 0, 1);
  if FPort = INVALID_HANDLE_VALUE then
    raise EDnException.Create(ErrWin32Error, GetLastError(), 'CreateIOCompletionPort');
  FRequestList := TObjectList.Create;
  FThread := TDnUdpReactorThread.Create(Self);
  Result := True;
end;

function TDnUdpReactor.TurnOff: Boolean;
begin
  PostQueuedCompletionStatus(FPort, 0, 0, Nil);
  FreeAndNil(FThread);
  Result := False;
end;

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

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

end.