unit uROAsyncSuperTcpServer;
{----------------------------------------------------------------------------}
{ RemObjects SDK Library - Core Library                                      }
{                                                                            }
{ compiler: Delphi 5 and up, Kylix 2 and up                                  }
{ platform: Win32, Linux                                                     }
{                                                                            }
{ (c)opyright RemObjects Software. all rights reserved.                      }
{                                                                            }
{ Using this code requires a valid license of the RemObjects SDK             }
{ which can be obtained at http://www.remobjects.com.                        }
{----------------------------------------------------------------------------}

{$I RemObjects.inc}

interface
uses
  SysUtils, Classes, uROClasses, uROClient, uROClientIntf, uROServerIntf,
  uROSessions, uROServer{$IFDEF MSWINDOWS}, Windows{$ENDIF}, uROAsyncSCHelpers,
  uROThreadPool, uROEventRepository;

type
  TROAsyncConnectionEvent = procedure (aChannel: IROTransport; const aGuid: TGuid) of object;
  TROBaseAsyncSuperTcpServer = class(TROServer)
  private
    fAckWaitTimeout: Integer;
    fMaxPackageSize: Integer;
    fGuidToClientMap: TStringList;
    fSkipAck: Boolean;
    fEventRepository: TROEventRepository;
    fBlockingEvents: Boolean;
    fOnClientConnected: TROAsyncConnectionEvent;
    fOnClientDisconnected: TROAsyncConnectionEvent;
    procedure SetThreadPool(const Value: TROThreadPool);
  protected
    fOwnsThreadPool: Boolean;
    fThreadPool: TROThreadPool;
    fClients: TThreadList;
    procedure HasData(Id: Integer; aClient: TROAsyncSuperChannelWorker; aData: TStream);

    procedure Notification(AComponent: TComponent; Operation: TOperation); override;
    procedure DispatchEvent(anEventDataItem : TROEventData; aSessionReference : TGUID; aSender: TObject);
  public
    constructor Create(aOwner: TComponent); override;
    destructor Destroy; override;

    property AckWaitTimeout: Integer read fAckWaitTimeout write fAckWaitTimeout default 10000;
    property MaxPackageSize: Integer read fMaxPackageSize write fMaxPackageSize default 10*1024*1024;
    property SkipAck: Boolean read fSkipAck write fSkipAck default false;

    property ThreadPool: TROThreadPool read fThreadPool write SetThreadPool;
    property EventRepository: TROEventRepository read fEventRepository write fEventRepository;
    property GuidToClientMap: TStringList read fGuidToClientMap;
    property BlockingEvents: Boolean read fBlockingEvents write fBlockingEvents default False;
    property OnClientConnected: TROAsyncConnectionEvent read fOnClientConnected write fOnClientConnected;
    property OnClientDisconnected: TROAsyncConnectionEvent read fOnClientDisconnected write fOnClientDisconnected;
  end;

  TROAsyncInvokerQueueItem = class(TInterfacedObject, IROThreadPoolCallback)
  private
    fData: TStream;
    fId: Integer;
    fCaller: TROBaseAsyncSuperTcpServer;
    FClient: TROAsyncSuperChannelWorker;
  public
    constructor Create(aCaller: TROBaseAsyncSuperTcpServer; aClient: TROAsyncSuperChannelWorker; Id: Integer; aData: TStream);
    procedure Callback(Caller: TROThreadPool; Thread: TThread);
    destructor Destroy; override;
  end;

  TSendEvent = class(TInterfacedObject, IROThreadPoolCallback)
  private
    fStream: TROEventData;
    fOwner: TROBaseAsyncSuperTcpServer;
    fClientGuid: TGUID;
    fSender: TROEventRepository;
    fWorkerOverride: IROTransport;
  protected
  public
    constructor Create(aOwner: TROBaseAsyncSuperTcpServer; aClientGuid: TGUID; aSender: TROEventRepository; aData: TROEventData);
    destructor Destroy; override;

    procedure Callback(Caller: TROThreadPool; Thread: TThread);
    property WorkerOverride: IROTransport read fWorkerOverride write fWorkerOverride;
  end;

implementation
uses
  uROTypes;

{ TROBaseAsyncSuperTcpServer }

constructor TROBaseAsyncSuperTcpServer.Create(aOwner: TComponent);
begin                                                   
  inherited Create(aOwner);
  fClients := TThreadList.Create;
  AckWaitTimeout := 10000;
  fMaxPackageSize := 10 * 1024 * 1024;
  fGuidToClientMap := TStringList.Create;
  fGuidToClientMap.Sorted := true;

end;

destructor TROBaseAsyncSuperTcpServer.Destroy;
begin
  Active := False;
  fGuidToClientMap.Free;
  fClients.Free;
  if fOwnsThreadPool then
    fThreadPool.Free;
  inherited Destroy;
end;

procedure TROBaseAsyncSuperTcpServer.HasData(Id: Integer; aClient: TROAsyncSuperChannelWorker; aData: TStream);
var
  lItem: IROThreadPoolCallback;
  lOrg: TROAsyncInvokerQueueItem;
begin
  lOrg := TROAsyncInvokerQueueItem.Create(Self, aClient, Id, aData);
  lItem := lOrg;
  try
    fThreadPool.QueueItem(lItem);
  except
    aClient.SendError(Id, ScCmdNoAck_QueueFull);
    lOrg.fData := nil; // caller will free it on exceptions
    raise;
  end;
end;

procedure TROBaseAsyncSuperTcpServer.Notification(AComponent: TComponent; Operation: TOperation);
begin
  if (Operation = opRemove) then begin
    if AComponent = fThreadPool then
      fThreadPool := nil;
    if AComponent = fEventRepository then
      fEventRepository := nil;
  end;
  inherited;
end;

procedure TROBaseAsyncSuperTcpServer.DispatchEvent(anEventDataItem : TROEventData; aSessionReference : TGUID; aSender: TObject);
var
  i: Integer;
  aref: IROSessionsChangesListener;
  se: IROThreadPoolCallback;
begin
  fClients.LockList;
  try
    i := fGuidToClientMap.IndexOf(GUIDToString(aSessionReference));
    if i = -1 then begin
      if fEventRepository <> nil then begin
        if Supports(fEventRepository, IROSessionsChangesListener, aref) then
          aref.SessionsChangedNotification(aSessionReference, saRemoveActiveListener, nil);
        fEventRepository.StoreEventData(EmptyGUID, Binary(anEventDataItem.Data), false, false, GUIDToString(aSessionReference));
        if anEventDataItem.RemoveRef = 0 then anEventDataItem.Free;
        exit;
      end;
    end;
  finally
    fClients.UnlockList;
  end;
  se := TSendEvent.Create(Self, aSessionReference, TROEventRepository(aSender), anEventDataItem);
  if BlockingEvents then
    se.Callback(fThreadPool, nil)
  else
    fThreadPool.QueueItem(se);
end;

procedure TROBaseAsyncSuperTcpServer.SetThreadPool(const Value: TROThreadPool);
begin
  if fOwnsThreadPool then
  begin
    fOwnsThreadPool := false;
    fThreadPool.Free;
  end;
  fThreadPool := Value;
end;



procedure TSendEvent.Callback(Caller: TROThreadPool; Thread: TThread);
var
  ms: TROConstantMemoryStream;
  i: Integer;
  wak: IROPackageAck;
  ass: IROActiveEventServer;
  obj: TROAsyncSuperChannelWorker;
  aRef: IROSessionsChangesListener;
begin
  obj := nil;
  ms := TROConstantMemoryStream.Create(TROConstantMemoryStream(fStream.Data));
  try
    try
      fOwner.fClients.LockList;
      try
        i := fOwner.fGuidToClientMap.IndexOf(GUIDToString(fClientGuid));
        if i = -1 then exit;
        if fWorkerOverride <> nil then
          obj := TROAsyncSuperChannelWorker(fWorkerOverride.GetTransportObject)
        else
          obj := TROAsyncSuperChannelWorker(fOwner.fGuidToClientMap.Objects[i]);
        wak := obj.SendPackage(ms, 0);
        if not Supports(obj, IROSessionsChangesListener, ass) then ass := nil;
      finally
        fOwner.fClients.UnlockList;
      end;
      TROAsyncSuperChannelWorker.WaitForAck(wak, fOwner.fAckWaitTimeout);
    except
      if fOwner.fEventRepository <> nil then begin
        if Supports(fOwner.fEventRepository, IROSessionsChangesListener, aref) then
          aref.SessionsChangedNotification(fClientGuid, saRemoveActiveListener, obj);
        fOwner.fEventRepository.StoreEventData(EmptyGUID, ms, false, false, GUIDToString(fClientGuid));
        exit;
      end;
    end;
  finally
    ms.Free;
  end;
end;

constructor TSendEvent.Create(aOwner: TROBaseAsyncSuperTcpServer; aClientGuid: TGUID; aSender: TROEventRepository; aData: TROEventData);
begin
  inherited Create;
  fStream := aData;
  fSender := aSender;
  fClientGuid := aClientGuid;
  fOwner := aOwner;
end;

destructor TSendEvent.Destroy;
begin
  if fStream.RemoveRef = 0 then
    fStream.Free;
  inherited Destroy;
end;

{ TROAsyncInvokerQueueItem }

procedure TROAsyncInvokerQueueItem.Callback(Caller: TROThreadPool; Thread: TThread);
var
  lResponse: TMemoryStream;
  l: TList;
  lClient: IROTransport;
begin
  lResponse := TMemoryStream.Create;
  try
    lClient := FClient;
    try
      fCaller.DispatchMessage(lClient, fData, lResponse);
    finally
      lClient := nil;
    end;
    lResponse.Seek(0, soFromBeginning);
    l := fCaller.fClients.LockList;
    try
      if l.IndexOf(FClient) = -1 then exit;
    finally
      fCaller.fClients.UnlockList;
    end;
    if lResponse.Size > FClient.MaxPackageSize then
      FClient.SendError(fId, ScCmdNoAck_MsgTooLarge)
    else
      FClient.SendPackage(lResponse, fId).RemoveFromList; //  we're not going to wait
  finally
    lResponse.Free;
  end;
end;

constructor TROAsyncInvokerQueueItem.Create(aCaller: TROBaseAsyncSuperTcpServer; aClient: TROAsyncSuperChannelWorker; Id: Integer; aData: TStream);
begin
  inherited Create;
  fCaller := aCaller;
  fId := Id;
  fData := aData;
  fClient := aClient;
end;

destructor TROAsyncInvokerQueueItem.Destroy;
begin
  fData.Free;
  inherited Destroy;
end;

end.