Componentes.Terceros.RemObjects/internal/5.0.23.613/1/RemObjects SDK for Delphi/Source/uIPAsyncSocket.pas

unit uIPAsyncSocket;
{$R-,Q-}
interface

{$I RemObjects.inc}

uses
  SysUtils, Classes, SyncObjs, synsock, uROThread;

const
  SocketsPerWorkerThread = 64;

type
  EIPSocketError = class(Exception);
  TIPBaseAsyncSocket = class;
  TIPSocketMode = (rsmTCPIP4, rsmTCPIP6);
  TIPSocketCallback = procedure(Sender: TIPBaseAsyncSocket) of object;
  TIPBaseAsyncSocket = class(TObject)
  private
    fBuffered: Boolean;
    fMaxLineLength: Integer;

    fOnDisconnected: TIPSocketCallback;
    fBufWriteCallback, fBufReadCallback: TIPSocketCallback;

    fLineSoFar: string;
    fReadLineCallback: TIPSocketCallback;
    fReadLineBuf: array of Byte;
    fReadLineBufStart,
    fReadLineBufEnd: Integer;
    fIdleTimeout: Integer;
    fOnTimeout: TIPSocketCallback;

    procedure BufferedSendCallback(Sender: TIPBaseAsyncSocket);
    procedure BufferedReceiveCallback(sender: TIPBaseAsyncSocket);
    procedure ReadLineCallback(sender: TIPBaseAsyncSocket);

    procedure BufferedBeginReceive(aCallback: TIPSocketCallback);
    function BufferedEndReceive: Integer;
  protected
    fLastData: TDateTime;

    fWriteDestData: Pointer;
    fWriteStart,
    fWriteLength: Integer;

    fReadDestData: Pointer;
    fReadStart,
    fReadLength: Integer;
    function GetReceiverBufferSize: Integer; virtual; abstract;
    function GetSendBufferSize: Integer; virtual; abstract;
    procedure SetReceiverBufferSize(const Value: Integer); virtual; abstract;
    procedure SetSendBufferSize(const Value: Integer); virtual; abstract;

    procedure IntBeginSend(aCallback: TIPSocketCallback); virtual; abstract;
    function IntEndSend: Integer; virtual; abstract;
    procedure IntBeginReceive(aCallback: TIPSocketCallback); virtual; abstract;
    function IntEndReceive: Integer; virtual; abstract;
  public
    constructor Create;
    destructor Destroy; override;

    property IdleTimeout: Integer read fIdleTimeout write fIdleTimeout;   // default: 0 = no timeout; Is checked every 30 seconds
    property Buffered: Boolean read fBuffered write fBuffered;
    property MaxLineLength: Integer read fMaxLineLength write fMaxLineLength; // 8192
    function GetLocalPort: Integer; virtual; abstract;
    function GetLocalIp: string; virtual; abstract;
    function GetRemotePort: Integer; virtual; abstract;
    function GetRemoteIp: string; virtual; abstract;

    property ReceiverBufferSize: Integer read GetReceiverBufferSize write SetReceiverBufferSize;
    property SendBufferSize: Integer read GetSendBufferSize write SetSendBufferSize;

    procedure BeginAccept(aCallback: TIPSocketCallback); virtual; abstract;
    function EndAccept: TIPBaseAsyncSocket; virtual; abstract;

    procedure BeginConnect(const aIp: string; aPort: Integer; aCallback: TIPSocketCallback); virtual; abstract;
    function EndConnect: Boolean; virtual; abstract;

    procedure BeginDisconnect(aCallback: TIPSocketCallback); virtual; abstract;
    procedure EndDisconnect; virtual; abstract;

    // When Buffered is set; the data to be sent and the data to be received is guaranteed to be of the size requested; or the user has been disconnected
    procedure BeginSend(aData: Pointer; aStart, aLength: Integer; aCallback: TIPSocketCallback);
    function EndSend: Integer;

    procedure BeginReceive(aData: Pointer; aStart, aLength: Integer; aCallback: TIPSocketCallback);
    function EndReceive: Integer;

    // returns true if there still is a line in the buffers
    function BufferReadLine(var aLine: string): Boolean;

    procedure BeginReadLine(aCallback: TIPSocketCallback);
    function EndReadLine(var aLine: string): Boolean;

    procedure Bind(const aIp: string; aPort: Integer); virtual; abstract;
    procedure Listen(aBacklog: Integer); virtual; abstract;

    property OnDisconnected: TIPSocketCallback read fOnDisconnected write fOnDisconnected;
    property OnTimeout: TIPSocketCallback read fOnTimeout write fOnTimeout;
  end;

  TIPAsyncSocket = class;

  TIPSocketWorkerThread = class(TROThread)
  private
    fReadSet,
    fWriteSet,
    fErrorSet: TFDSet;

    fCount: Integer;
    fLastTimeoutCheck: TDateTime;

    fData: array[0.. SocketsPerWorkerThread-1] of TIPAsyncSocket;
    fMyCs: TCriticalSection;
    fSleepEvent: TEvent;
  protected
    procedure Execute; override;
  public
    constructor Create;
    destructor Destroy; override;
  end;

  TIPAsyncSocket = class(TIPBaseAsyncSocket)
  private
    fSock: TSocket;
    fLocSin,
    fRemSin: TVarSin;
    fMode: TIPSocketMode;
    fThread: TIPSocketWorkerThread;
    fReadEvent,
    fWriteEvent,
    fAcceptEvent,
    fConnectEvent,
    fDisconnectEvent: TIPSocketCallback;
    fWantRead,
    fWantWrite,
    fWantError: Boolean;
    fConnectResult: Integer;
  protected
    function GetReceiverBufferSize: Integer; override;
    function GetSendBufferSize: Integer; override;
    procedure IntBeginReceive(aCallback: TIPSocketCallback); override;
    procedure IntBeginSend(aCallback: TIPSocketCallback); override;
    function IntEndReceive: Integer; override;
    function IntEndSend: Integer; override;
    procedure SetReceiverBufferSize(const Value: Integer); override;
    procedure SetSendBufferSize(const Value: Integer); override;

    constructor Create(aMode: TIPSocketMode; aHandle: TSocket); overload;
  public
    constructor Create(aMode: TIPSocketMode); overload;
    destructor Destroy; override;

    procedure BeginAccept(aCallback: TIPSocketCallback); override;
    function EndAccept: TIPBaseAsyncSocket; override;

    procedure BeginConnect(const aIp: String; aPort: Integer; aCallback: TIPSocketCallback); override;
    function EndConnect: Boolean; override;

    procedure BeginDisconnect(aCallback: TIPSocketCallback); override;
    procedure EndDisconnect; override;

    procedure Bind(const aIp: String; aPort: Integer); override;

    function GetLocalIp: String; override;
    function GetLocalPort: Integer; override;
    function GetRemoteIp: String; override;
    function GetRemotePort: Integer; override;
    procedure Listen(aBacklog: Integer); override;
  end;
  

function InitSocket: Boolean;

implementation

var
  SockInitialized: Boolean = False;
  WsaData: TWSAData;

type
  TByteArray = array[0..4095] of byte;
  PByteArray = ^TByteArray;


var
  WorkerCS: TCriticalSection;
  Workers: TList;

function GetWorker(aSock: TIPAsyncSocket): TIPSocketWorkerThread;
var
  i, j: Integer;
  lWorker: TIPSocketWorkerThread;
begin
  WorkerCS.Enter;
  try
    for i := 0 to Workers.Count -1 do begin
      lWorker := TIPSocketWorkerThread(Workers[i]);
      if lWorker.fCount < SocketsPerWorkerThread then begin
        lWorker.Fcount := lWorker.fCount +1;
        lWorker.fMyCs.Enter;
        try
          for j :=  0 to SocketsPerWorkerThread -1 do begin
            if lworker.fData[j] = nil then begin
              lWorker.FData[j] := aSock;
              result := lWorker;
              exit;
            end;
          end;
        finally
         lWorker.fMyCs.Leave;
        end;
      end;
    end;
    lWorker := TIPSocketWorkerThread.Create;
    lWorker.fMyCs.Enter;
    try
      lWorker.fData[0] :=aSock;
      lWorker.fCount := 1;
    finally
      lWorker.fMyCs.Leave;
    end;
    Workers.Add(lWorker);
    result := lWorker;
  finally
    WorkerCS.Leave;
  end;
end;

procedure ReleaseWorker(aWorker: TIPSocketWorkerThread; fSocket: TIPAsyncSocket);
var
  j: Integer;
begin
  if aworker = nil then exit;
  if fSocket = nil then exit;
  aWorker.fMyCs.Enter;
  try
    for j :=  0 to SocketsPerWorkerThread -1 do begin
      if aWorker.fData[j] = fSocket then begin
        aWorker.FData[j] := nil;
        aWorker.fCount := aWorker.fCount -1;
        Exit;
      end;
    end;
  finally
    aWorker.fMyCs.Leave;
  end;
end;

function InitSocket: Boolean;
begin
  if SockInitialized then begin
    result := true;
    exit;
  end;

  if not InitSocketInterface(DLLStackName) then begin
    result := False;
    exit;
  end;
  synsock.WSAStartup(WinsockLevel, WsaData);
  SockInitialized := true;
  result := true;
end;


{ TIPBaseAsyncSocket }

procedure TIPBaseAsyncSocket.BeginReadLine(aCallback: TIPSocketCallback);
var
  i, lLen: Integer;
begin
  if not fBuffered then raise EIPSocketError.Create('Buffered has to be set for BeginReadLine to work');
  if (@fBufReadCallback <> nil) or (@fReadLineCallback <> nil) then raise EIPSocketError.Create('BeginReceive already running');
  fReadLineCallback := aCallback;
  if fReadLineBuf <> nil then begin
    for i := fReadLineBufStart to fReadLineBufEnd -1 do begin
      if fReadLineBuf[i] = 10 then begin
        lLen := i - fReadLineBufStart;
        if (i > fReadLineBufStart) and (fReadLineBuf[i-1] = 13) then dec(lLen);
        SetLength(fLineSoFar, lLen);
        Move(fReadLineBuf[fReadlineBufStart], fLineSoFar[1], lLen);
        fReadLineBufStart := i +1;
        fReadLineCallback := nil;
        aCallback(self);
        exit;
      end;
    end;
    SetLength(fLineSoFar, fReadLineBufEnd - fReadLineBufStart);
    Move(fReadLineBuf[fReadlineBufStart], fLineSoFar[1], fReadLineBufEnd - fReadLineBufStart);
  end;
  if fReadLineBuf = nil then SetLength(fReadLineBuf, 4096);
  fReadLineBufStart := 0;
  fReadLineBufEnd := 0;
  fReadDestData := fReadLineBuf;
  fReadStart := 0;
  fReadLength := 4096;
  IntBeginReceive(ReadLineCallback);
end;

procedure TIPBaseAsyncSocket.BeginReceive(aData: Pointer; aStart, aLength: Integer; aCallback: TIPSocketCallback);
begin
  if not fBuffered then begin
    fReadDestData := aData;
    fReadStart := aStart;
    fReadLength := aLength;
    IntBeginReceive(aCallback);
    exit;
  end;
  if @aCallback = nil then raise EIPSocketError.Create('Invalid callback');
  if aLength <= 0 then raise EIPSocketError.Create('Invalid length');
  if (@fBufReadCallback <> nil) or (@fReadLineCallback <> nil) then raise EIPSocketError.Create('BeginReceive already running');
  fBufReadCallback := aCallback;
  fReadDestData := @(PByteArray(aData)^[aStart]); // move ahead so we always start at 0;
  fReadStart := 0;
  fReadLength := aLength;
  BufferedBeginReceive(BufferedReceiveCallback);
end;


procedure TIPBaseAsyncSocket.BufferedBeginReceive(aCallback: TIPSocketCallback);
begin
  if (fReadLineBuf <> nil) and (fReadLineBufStart < fReadLineBufEnd) then begin
    aCallback(self);
    exit;
  end;
  IntBeginReceive(aCallback);
end;

function TIPBaseAsyncSocket.BufferedEndReceive: Integer;
begin
  if (fReadLineBufStart < fReadLineBufEnd) and (fReadLineBuf <> nil) then begin
    result := fReadLineBufEnd - fReadLineBufStart;
    if result > fReadLength then result := fReadLength;

    Move(fReadLineBuf[fReadLineBufStart], PByteArray(fReadDestData)^[fReadStart], Result);
    fReadLineBufStart := fReadLineBufStart + Result; 
    fReadStart := fReadStart + Result;
    exit;
  end;
  result := IntEndReceive;
  fReadStart := fReadStart + Result;
end;


procedure TIPBaseAsyncSocket.BufferedReceiveCallback(
  sender: TIPBaseAsyncSocket);
var
  lCallback: TIPSocketCallback;
  lEndReceiveResult: Integer;
begin
  lEndReceiveResult := BufferedEndReceive;
  if lEndReceiveResult <= 0 then begin // disconnected
    // fReadStart := fReadStart + lEndReceiveResult;
    lCallback := fBufReadCallback;
    fBufReadCallback := nil;
    lCallback(self);
    if assigned(fOnDisconnected) then
      fOnDisconnected(Self);
    exit;
  end;
 //ReadStart := fReadStart + lEndReceiveResult;
  if fReadStart < fReadLength then begin
      BufferedBeginReceive(BufferedReceiveCallback);
  end else begin
    lCallback := fBufReadCallback;
    fBufReadCallback := nil;
    lCallback(self);
  end;
end;

function TIPBaseAsyncSocket.EndReceive: Integer;
begin
  if not fBuffered then begin
    result := IntEndReceive;
    exit;
  end;
  result := fReadStart;
  fWriteDestData := nil;
  fBufWriteCallback := nil;
end;

procedure TIPBaseAsyncSocket.BeginSend(aData: Pointer; aStart,
  aLength: Integer; aCallback: TIPSocketCallback);
begin
  if not fBuffered then begin
    fWriteDestData:= aData;
    fWriteStart := aStart;
    fWriteLength := aLength;
    IntBeginSend(aCallback);
    exit;
  end;
  if @aCallback = nil then raise EIPSocketError.Create('Invalid callback');
  if aLength <= 0 then raise EIPSocketError.Create('Invalid length');
  if @fBufWriteCallback <> nil then raise EIPSocketError.Create('BeginSend already running');

  fBufWriteCallback := aCallback;
  fWriteDestData:= @(PByteArray(aData)^[aStart]); // move ahead so we always start at 0
  fWriteStart := 0;
  fWriteLength := aLength;
  IntBeginSend(BufferedSendCallback);
end;

procedure TIPBaseAsyncSocket.BufferedSendCallback(Sender: TIPBaseAsyncSocket);
var
  lCallback: TIPSocketCallback;
  lEndSendResult: Integer;
begin
  lEndSendResult := IntEndSend;
  if lEndSendResult <= 0 then begin // disconnected
    fWriteStart := fWriteStart + lEndSendResult;
    lCallback := fBufWriteCallback;
    fBufWriteCallback := nil;
    lCallback(self);
    if assigned(fOnDisconnected) then
      fOnDisconnected(Self);
    exit;
  end;
  fWriteStart := fWriteStart + lEndSendResult;
  if fWriteStart < fWriteLength then begin
    IntBeginSend(BufferedSendCallback);
  end else begin
    lCallback := fBufWriteCallback;
    fBufWriteCallback := nil;
    lCallback(self);
  end;
end;

function TIPBaseAsyncSocket.EndSend: Integer;
begin
  if not fBuffered then begin
    result := IntEndSend;
    if (Result <= 0) and assigned(fOnDisconnected) then fOnDisconnected(Self);
    exit;
  end;
  result := fWriteStart;
  fWriteDestData := nil;
  fBufWriteCallback := nil;
end;

function TIPBaseAsyncSocket.BufferReadLine(var aLine: string): Boolean;
var
  lLen, i: Integer;
begin
  if not fBuffered then raise EIPSocketError.Create('Buffered has to be set for EndReadLine to work');
  if fReadLineBuf <> nil then begin
    for i := fReadLineBufStart to fReadLineBufEnd -1 do begin
      if fReadLineBuf[i] = 10 then begin
        lLen := i - fReadLineBufStart;
        if (i > fReadLineBufStart) and (fReadLineBuf[i-1] = 13) then dec(lLen);
        SetLength(aLine, lLen);
        Move(fReadLineBuf[fReadlineBufStart], aLine[1], lLen);
        fReadLineBufStart := i +1;
        result := true;
        exit;
      end;
    end;
  end;
  result := false;
end;

constructor TIPBaseAsyncSocket.Create;
begin
  inherited;
  fMaxLineLength := 8192;
end;

destructor TIPBaseAsyncSocket.Destroy;
begin
  inherited;
end;

function TIPBaseAsyncSocket.EndReadLine(var aLine: string): Boolean;
begin
  aLine := fLineSoFar;
  result := fReadLineBufEnd >= 0;
end;

procedure TIPBaseAsyncSocket.ReadLineCallback(sender: TIPBaseAsyncSocket);
var
  s: string;
  lCb: TIPSocketCallback;
  i, lLen: Integer;
begin
  fReadLineBufEnd := IntEndReceive;
  if (fReadLineBufEnd <= 0) then begin
    if assigned(fOnDisconnected) then
      OnDisconnected(self);
    exit;
  end;

  for i := fReadLineBufStart to fReadLineBufEnd -1 do begin
    if fReadLineBuf[i] = 10 then begin
      lLen := i - fReadLineBufStart;
      if (i > fReadLineBufStart) and (fReadLineBuf[i-1] = 13) then dec(lLen);
      SetLength(s, lLen);
      Move(fReadLineBuf[fReadlineBufStart], s[1], lLen);
      fReadLineBufStart := i +1;
      fLineSoFar := fLineSoFar + s;
      lCb := fReadLineCallback;
      fReadLineCallback := nil;
      lCb(self);
      exit;
    end;
  end;
  SetLength(s, fReadLineBufEnd - fReadLineBufStart);
  Move(fReadLineBuf[fReadlineBufStart], s[1], fReadLineBufEnd - fReadLineBufStart);
  fLineSoFar := fLineSoFar + s;
  fReadLineBufStart := 0;
  fReadLineBufEnd := 0;
  fReadDestData := fReadLineBuf;
  fReadStart := 0;
  fReadLength := 4096;
  IntBeginReceive(ReadLineCallback);
end;

{ TIPAsyncSocket }

procedure TIPAsyncSocket.BeginAccept(aCallback: TIPSocketCallback);
begin
  if assigned(fAcceptEvent) then raise EIPSocketError.Create('BeginAccept already called');
  if fThread = nil then fThread := GetWorker(self);
  fAcceptEvent := aCallback;
  fWantRead := true;
end;

procedure TIPAsyncSocket.BeginConnect(const aIp: String; aPort: Integer;
  aCallback: TIPSocketCallback);
var
  lSin: TVarSin;
  lRes: Integer;
begin
  if assigned(fConnectEvent) then raise EIPSocketError.Create('BeginConnect already called');
  if fThread = nil then fThread := GetWorker(self);

  fConnectEvent := aCallback;
  fConnectResult := 0;
  if fMode = rsmTCPIP6 then
    lRes := SetVarSin(lSin, aIp, IntToStr(aPort), AF_INET6, PF_INET6, SOCK_STREAM, false)
  else
    lRes := SetVarSin(lSin, aIp, IntToStr(aPort), AF_INET, PF_INET, SOCK_STREAM, true);
  if lRes <> 0 then raise EIPSocketError.Create('Invalid Address/Port: '+inttostr(WSAGetLastError));
  lRes := Connect(fSock, lSin);
  if lRes <> 0 then begin
    LRes := WSAGetLastError;
    if lRes <> WSAEWOULDBLOCK then raise EIPSocketError.Create('Cannot connect: '+inttostr(Lres));
  end;
  fWantWrite := true;
  fWantError := true;
end;

procedure TIPAsyncSocket.BeginDisconnect(aCallback: TIPSocketCallback);
begin
  if assigned(fDisconnectEvent) then raise EIPSocketError.Create('BeginDisconnect already called');
  fDisconnectEvent := aCallback;
  Synsock.Shutdown(FSock, 1);
  fWantRead := true;
end;

procedure TIPAsyncSocket.Bind(const aIp: String; aPort: Integer);
var
  lRes: Integer;
  lSin: TVarSin;
begin
  if fMode = rsmTCPIP6 then
    lRes := SetVarSin(lSin, aIp, IntToStr(aPort), AF_INET6, PF_INET6, SOCK_STREAM, false)
  else
    lRes := SetVarSin(lSin, aIp, IntToStr(aPort), AF_INET, PF_INET, SOCK_STREAM, true);
  if lRes <> 0 then raise EIPSocketError.Create('Invalid Address/Port: '+inttostr(WSAGetLastError));
  lRes := SynSock.Bind(fSock, lSin);
  if lRes <> 0 then raise EIPSocketError.Create('Cannot bind: '+inttostr(WSAGetLastError));
end;

constructor TIPAsyncSocket.Create(aMode: TIPSocketMode);
var
  lRes, lVal: Integer;
begin
  inherited Create;
  if not InitSocket then raise EIPSocketError.Create('Could not load socket stack');
  fMode := aMode;
  if aMode = rsmTCPIP6 then
    fSock := Socket(AF_INET6, SOCK_STREAM, IPPROTO_TCP)
  else
    fSock := Socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
  if fSock <= 0 then raise EIPSocketError.Create('Unable to create socket');
  lVal := 1;
  lRes := IoctlSocket(fsock, FIONBIO, lVal);
  if lRes <> 0 then raise EIPSocketError.Create('Cannot set socket to asynchronous mode: '+inttostr(synsock.WSAGetLastError));
end;

constructor TIPAsyncSocket.Create(aMode: TIPSocketMode; aHandle: TSocket);
var
  lVal, lRes: Integer;

begin
  inherited Create;
  fMode := aMode;
  fSock := aHandle;
  lVal := 1;
  lRes := IoctlSocket(fsock, FIONBIO, lVal);
  if lRes <> 0 then raise EIPSocketError.Create('Cannot set socket to asynchronous mode');
  if fThread = nil then fThread := GetWorker(self);
end;

destructor TIPAsyncSocket.Destroy;
begin
  if fThread <> nil then begin
    ReleaseWorker(fThread, self);
    fThread := nil;
  end;
  if FSock <>0 then
    CloseSocket(fSock);
  inherited;
end;

function TIPAsyncSocket.EndAccept: TIPBaseAsyncSocket;
var
  lAddr: TVarSin;
  lSock: TSocket;
begin
  fAcceptEvent := nil;

  lSock := synsock.Accept(fSock, lAddr);

  if lSock <= 0 then raise EIPSocketError.Create('Cannot accept connection');
  Result := TIPAsyncSocket.Create(fMode, lSock);
  fWantRead := false;
end;

function TIPAsyncSocket.EndConnect: Boolean;
begin
  if fConnectResult = 0 then result := true
  else result := false;
  fConnectEvent := nil;
  fWantWrite := false;
  fWantError := false;
end;

procedure TIPAsyncSocket.EndDisconnect;
begin
  fDisconnectEvent := nil;
  CloseSocket(fSock);
  if fThread <> nil then begin
    ReleaseWorker(fThread, self);
    fThread := nil;
  end;
  fSock := 0;
  fWantRead := false;
end;

function TIPAsyncSocket.GetLocalIp: String;
begin
  if fLocSin.AddressFamily = 0 then
    synsock.GetSockName(FSock, fLocSin);
  result := GetSinIP(fLocSin);
end;

function TIPAsyncSocket.GetLocalPort: Integer;
begin
  if fLocSin.AddressFamily = 0 then
    synsock.GetSockName(FSock, fLocSin);
  result := GetSinPort(fLocSin);
end;

function TIPAsyncSocket.GetReceiverBufferSize: Integer;
var
  lRes: Integer;
begin
  result := 0;
  lRes := 4;
  GetSockOpt(fSock, SOL_Socket, SO_RCVBUF, PChar(@Result), lRes);
end;

function TIPAsyncSocket.GetRemoteIp: String;
begin
  if fRemSin.AddressFamily = 0 then
    synsock.GetPeerName(FSock, fRemSin);
  result := GetSinIp(fRemSin);
end;

function TIPAsyncSocket.GetRemotePort: Integer;
begin
  if fRemSin.AddressFamily = 0 then
    synsock.GetPeerName(FSock, fRemSin);
  result := GetSinPort(fRemSin);
end;

function TIPAsyncSocket.GetSendBufferSize: Integer;
var
  lRes: Integer;
begin
  result := 0;
  lRes := 4;
  GetSockOpt(fSock, SOL_Socket, SO_SNDBUF, PChar(@Result), lRes);
end;

procedure TIPAsyncSocket.IntBeginReceive(aCallback: TIPSocketCallback);
begin
  if assigned(fReadEvent) then raise EIPSocketError.Create('beginReceive already called');
  fLastData := Now;
  fReadEvent := aCallback;
  fWantRead := true;
end;

procedure TIPAsyncSocket.IntBeginSend(aCallback: TIPSocketCallback);
begin
  if assigned(fWriteEvent) then raise EIPSocketError.Create('BeginSend already called');
  fLastData := Now;
  fWriteEvent := aCallback;
  fWantWrite := true;
end;

function TIPAsyncSocket.IntEndReceive: Integer;
begin
  fwantRead := false;
  fReadEvent := nil;
  fLastData := Now;
  result := Recv(fSock, @PByteArray(fReadDestData)[fReadStart], fReadLength, 0);
end;

function TIPAsyncSocket.IntEndSend: Integer;
begin
  fWriteEvent := nil;
  fWantWrite := false;
  fLastData := Now;
  Result := Send(fSock, @PByteArray(fWriteDestData)[fWriteStart], fWriteLength, 0);
end;

procedure TIPAsyncSocket.Listen(aBacklog: Integer);
begin
  synsock.Listen(fSock, aBacklog);
end;

procedure TIPAsyncSocket.SetReceiverBufferSize(const Value: Integer);
var
  lRes: Integer;
begin
  lRes := SetSockOpt(fsock, SOL_Socket, SO_RCVBUF, pchar(@value), 4);
  if lRes <> 0 then raise EIPSocketError.Create('Unable to set option: '+Inttostr(WSAGetLastError));
end;

procedure TIPAsyncSocket.SetSendBufferSize(const Value: Integer);
var
  lRes: Integer;
begin
  lRes := SetSockOpt(fsock, SOL_Socket, SO_SNDBUF, pchar(@value), 4);
  if lRes <> 0 then raise EIPSocketError.Create('Unable to set option: '+Inttostr(WSAGetLastError));
end;

procedure CleanupWorkers;
var
  i: Integer;
  lWorker: TIPSocketWorkerThread;
begin
  WorkerCS.Enter;
  try
    for i := Workers.Count - 1 downto 0 do begin
      lWorker := TIPSocketWorkerThread(Workers[i]);
      lWorker.Free;
    end;
  finally
    WorkerCS.Leave;
  end;
end;

{ TIPSocketWorkerThread }

constructor TIPSocketWorkerThread.Create;
begin
  inherited Create(True);
  fMyCs := TCriticalSection.Create;
  fSleepEvent := TEvent.Create(nil, false, false, '');
  Resume;
end;

destructor TIPSocketWorkerThread.Destroy;
begin
  fSleepEvent.Free;
  FMyCs.Free;
  inherited Destroy;
end;

const
  TimeoutCheckSpan = 1.0 / (24 * 60 * 2); // every 30 sec
  OneSecond = 1.0 / (24 * 60 * 60);

procedure TIPSocketWorkerThread.Execute;
var
  i: Integer;
  lCount: Integer;
  lDummy: Integer;
  lDate: TDateTime;
  lSock: TIPAsyncSocket;
  lEvent: TIPSocketCallback;
  TimeVal: TTimeVal;
begin
  TimeVal.tv_sec := 0;
  TimeVal.tv_usec := 50;
  fLastTimeoutCheck := Now;
  while not Terminated do
  begin
    fMyCs.Enter;
    try
      lDate := Now;
      if (lDate - fLastTimeoutCheck) > TimeoutCheckSpan then begin
        fLastTimeoutCheck := lDate;
        for i := 0 to SocketsPerWorkerThread -1 do begin
          lSock := fData[i];
          if lSock <> nil then begin
            if (lSock.IdleTimeout > 0) and (lSock.fLastData + (OneSecond * lSock.IdleTimeout) < lDate) then begin
              fMyCs.Leave;
              try
                if assigned(lSock.OnTimeout) then lSock.OnTimeout(lSock);
              finally
                fMyCs.Enter;
              end;
            end;
          end;
        end;

      end;
      fSleepEvent.ResetEvent;
      lCount := fCount ;
      if lCount > 0 then begin
        FD_ZERO(fReadSet);
        FD_ZERO(fWriteSet);
        FD_ZERO(fErrorSet);

        for i := 0 to SocketsPerWorkerThread -1 do begin
          if fData[i] <> nil then begin
            if fData[i].fWantRead then
              FD_Set(fData[i].fSock, fReadset);
            if fData[i].fWantWrite then
              FD_Set(fData[i].fSock, fWriteset);
            if fData[i].fWantError then
              FD_Set(fData[i].fSock, fErrorSet);
          end;
        end;
      end;
    finally
      fMyCs.Leave;
    end;
    if lCount = 0 then fSleepEvent.WaitFor(1000) else begin
      lCount := Select(lCount, @fReadSet, @FwriteSet, @fErrorSet, @TimeVal);
      if lCount > 0 then begin
        fMycs.Enter;
        try
          for i := 0 to SocketsPerWorkerThread do begin
            if lCount = 0 then break;
            if Fdata[i] = nil then continue;


            if FD_ISSET(fData[i].fSock, fReadSet) then begin // Read, Accept and Disconnect
              lEvent := fData[i].fReadEvent;
              if assigned(lEvent) then begin
                fMyCS.Leave;
                try
                  dec(lCount);
                  lEvent(fData[i]);
                finally
                  fMyCs.Enter;
                end;
              end else begin
                lEvent := fData[i].fAcceptEvent;
                if assigned(lEvent) then begin
                  fMyCS.Leave;
                  try
                    dec(lCount);
                    lEvent(fData[i]);
                  finally
                    fMyCs.Enter;
                  end;
                end else begin
                  lEvent := fData[i].fDisconnectEvent;
                  if assigned(lEvent) then begin
                    fMyCS.Leave;
                    try
                      dec(lCount);
                      lEvent(fData[i]);
                    finally
                      fMyCs.Enter;
                    end;
                  end;
                end;
              end;
            end else

            if FD_ISSET(fData[i].FSock, FWriteset) then begin // Connect, Write
              lEvent := fData[i].fWriteEvent;
              if assigned(lEvent) then begin
                fMyCS.Leave;
                try
                  dec(lCount);
                  lEvent(fData[i]);
                finally
                  fMyCs.Enter;
                end;
              end else begin
                lEvent := fData[i].fConnectEvent;
                if assigned(lEvent) then begin
                  fMyCs.Leave;
                  try
                    dec(lCount);
                    lEvent(FData[i]);
                  finally
                    fMyCs.Enter;
                  end;
                end;
              end;
            end else

            if FD_ISSET(fData[i].FSock, fErrorSet) then begin // Connect
              lEvent := fData[i].fConnectEvent;
              lDummy := 4;
              GetSockOpt(fData[i].FSock, SOL_SOCKET, SO_ERROR, Pchar(@fData[i].fConnectResult), lDummy);
              if assigned(lEvent) then begin
                fMyCS.Leave;
                try
                  dec(lCount);
                  lEvent(fData[i]);
                finally
                  fMyCs.Enter;
                end;
              end;
            end;
          end;
        finally
          fMycs.Leave;
        end;
      end;
    end;
  end;
end;

initialization
  WorkerCS := TCriticalSection.Create;
  Workers := TList.Create;
finalization
  if SockInitialized then begin
    synsock.WSACleanup;
    DestroySocketInterface;
  end;
  CleanupWorkers;
  Workers.Free;
  WorkerCS.Free;
end.