DelphiWebsockets/IdSocketIOHandling.pas

1687 lines
49 KiB
ObjectPascal
Raw Normal View History

2013-11-11 21:14:42 +01:00
unit IdSocketIOHandling;
interface
uses
Classes, Generics.Collections,
superobject,
IdServerBaseHandling, IdContext, IdException, IdIOHandlerWebsocket, IdHTTP,
2014-06-25 15:30:18 +02:00
SyncObjs, SysUtils;
2013-11-11 21:14:42 +01:00
type
TSocketIOContext = class;
TSocketIOCallbackObj = class;
TIdBaseSocketIOHandling = class;
TIdSocketIOHandling = class;
2013-11-11 21:14:42 +01:00
ISocketIOContext = interface;
ISocketIOCallback = interface;
2013-11-11 21:14:42 +01:00
TSocketIOMsg = reference to procedure(const ASocket: ISocketIOContext; const aText: string; const aCallback: ISocketIOCallback);
TSocketIOMsgJSON = reference to procedure(const ASocket: ISocketIOContext; const aJSON: ISuperObject; const aCallback: ISocketIOCallback);
2013-11-11 21:14:42 +01:00
TSocketIONotify = reference to procedure(const ASocket: ISocketIOContext);
TSocketIOEvent = reference to procedure(const ASocket: ISocketIOContext; const aArgument: TSuperArray; const aCallback: ISocketIOCallback);
2013-11-18 14:27:13 +01:00
TSocketIOError = reference to procedure(const ASocket: ISocketIOContext; const aErrorClass, aErrorMessage: string);
2014-06-25 15:30:18 +02:00
TSocketIOEventError = reference to procedure(const ASocket: ISocketIOContext; const aCallback: ISocketIOCallback; E: Exception);
2013-11-11 21:14:42 +01:00
TSocketIONotifyList = class(TList<TSocketIONotify>);
TSocketIOEventList = class(TList<TSocketIOEvent>);
EIdSocketIoUnhandledMessage = class(EIdSilentException);
ISocketIOContext = interface
['{ACCAC678-054C-4D75-8BAD-5922F55623AB}']
function GetCustomData: TObject;
function GetOwnsCustomData: Boolean;
procedure SetCustomData(const Value: TObject);
procedure SetOwnsCustomData(const Value: Boolean);
property CustomData: TObject read GetCustomData write SetCustomData;
property OwnsCustomData: Boolean read GetOwnsCustomData write SetOwnsCustomData;
2013-11-11 21:14:42 +01:00
function ResourceName: string;
function PeerIP: string;
function PeerPort: Integer;
procedure Lock;
procedure UnLock;
function IsDisconnected: Boolean;
procedure EmitEvent(const aEventName: string; const aData: ISuperObject; const aCallback: TSocketIOMsgJSON = nil; const aOnError: TSocketIOError = nil);overload;
procedure EmitEvent(const aEventName: string; const aData: string; const aCallback: TSocketIOMsgJSON = nil; const aOnError: TSocketIOError = nil);overload;
2013-11-18 14:27:13 +01:00
procedure Send(const aData: string; const aCallback: TSocketIOMsgJSON = nil; const aOnError: TSocketIOError = nil);
procedure SendJSON(const aJSON: ISuperObject; const aCallback: TSocketIOMsgJSON = nil; const aOnError: TSocketIOError = nil);
2013-11-11 21:14:42 +01:00
end;
TSocketIOContext = class(TInterfacedObject,
ISocketIOContext)
private
FLock: TCriticalSection;
2013-11-11 21:14:42 +01:00
FPingSend: Boolean;
FConnectSend: Boolean;
FGUID: string;
FPeerIP: string;
FCustomData: TObject;
FOwnsCustomData: Boolean;
2013-11-11 21:14:42 +01:00
procedure SetContext(const Value: TIdContext);
procedure SetConnectSend(const Value: Boolean);
procedure SetPingSend(const Value: Boolean);
function GetCustomData: TObject;
function GetOwnsCustomData: Boolean;
procedure SetCustomData(const Value: TObject);
procedure SetOwnsCustomData(const Value: Boolean);
2013-11-11 21:14:42 +01:00
protected
FHandling: TIdBaseSocketIOHandling;
FContext: TIdContext;
FIOHandler: TIdIOHandlerWebsocket;
FClient: TIdHTTP;
FEvent: TEvent;
FQueue: TList<string>;
FPendingMessages: TList<Int64>;
2013-11-11 21:14:42 +01:00
procedure QueueData(const aData: string);
procedure ServerContextDestroy(AContext: TIdContext);
public
constructor Create();overload;
constructor Create(aClient: TIdHTTP);overload;
procedure AfterConstruction; override;
2013-11-11 21:14:42 +01:00
destructor Destroy; override;
procedure Lock;
procedure UnLock;
function WaitForQueue(aTimeout_ms: Integer): string;
function ResourceName: string;
function PeerIP: string;
function PeerPort: Integer;
function IsDisconnected: Boolean;
2013-11-11 21:14:42 +01:00
property GUID: string read FGUID;
property Context: TIdContext read FContext write SetContext;
property PingSend: Boolean read FPingSend write SetPingSend;
property ConnectSend: Boolean read FConnectSend write SetConnectSend;
property CustomData: TObject read GetCustomData write SetCustomData;
property OwnsCustomData: Boolean read GetOwnsCustomData write SetOwnsCustomData;
2013-11-11 21:14:42 +01:00
//todo: OnEvent per socket
//todo: store session info per connection (see Socket.IO Set + Get -> Storing data associated to a client)
//todo: namespace using "Of"
procedure EmitEvent(const aEventName: string; const aData: ISuperObject; const aCallback: TSocketIOMsgJSON = nil; const aOnError: TSocketIOError = nil);overload;
procedure EmitEvent(const aEventName: string; const aData: string; const aCallback: TSocketIOMsgJSON = nil; const aOnError: TSocketIOError = nil);overload;
2013-11-11 21:14:42 +01:00
// procedure BroadcastEventToOthers(const aEventName: string; const aData: ISuperObject; const aCallback: TSocketIOMsgJSON = nil);
2013-11-18 14:27:13 +01:00
procedure Send(const aData: string; const aCallback: TSocketIOMsgJSON = nil; const aOnError: TSocketIOError = nil);
procedure SendJSON(const aJSON: ISuperObject; const aCallback: TSocketIOMsgJSON = nil; const aOnError: TSocketIOError = nil);
2013-11-11 21:14:42 +01:00
end;
ISocketIOCallback = interface
['{BCC31817-7FD8-4CF6-B68B-0F9BAA80DF90}']
procedure SendResponse(const aResponse: string);
function IsResponseSend: Boolean;
end;
TSocketIOCallbackObj = class(TInterfacedObject,
ISocketIOCallback)
2013-11-11 21:14:42 +01:00
protected
FHandling: TIdBaseSocketIOHandling;
FSocket: ISocketIOContext;
2013-11-11 21:14:42 +01:00
FMsgNr: Integer;
{ISocketIOCallback}
2013-11-11 21:14:42 +01:00
procedure SendResponse(const aResponse: string);
2013-11-18 14:27:13 +01:00
function IsResponseSend: Boolean;
public
constructor Create(aHandling: TIdBaseSocketIOHandling; aSocket: ISocketIOContext; aMsgNr: Integer);
end;
TSocketIOPromise = class
protected
Done, Success: Boolean;
Error: Exception;
Data : ISuperObject;
Event: TEvent;
public
procedure AfterConstruction; override;
destructor Destroy; override;
2013-11-11 21:14:42 +01:00
end;
ESocketIOException = class(Exception);
ESocketIOTimeout = class(ESocketIOException);
ESocketIODisconnect = class(ESocketIOException);
2013-11-11 21:14:42 +01:00
TIdBaseSocketIOHandling = class(TIdServerBaseHandling)
protected
FLock: TCriticalSection;
FConnections: TObjectDictionary<TIdContext,ISocketIOContext>;
FConnectionsGUID: TObjectDictionary<string,ISocketIOContext>;
2013-11-11 21:14:42 +01:00
FOnConnectionList,
FOnDisconnectList: TSocketIONotifyList;
FOnEventList: TObjectDictionary<string,TSocketIOEventList>;
FOnSocketIOMsg: TSocketIOMsg;
FOnSocketIOJson: TSocketIOMsgJSON;
procedure ProcessEvent(const AContext: ISocketIOContext; const aText: string; aMsgNr: Integer; aHasCallback: Boolean);
2014-06-25 15:30:18 +02:00
private
FOnEventError: TSocketIOEventError;
2013-11-11 21:14:42 +01:00
protected
type
TSocketIOCallback = procedure(const aData: string) of object;
TSocketIOCallbackRef = reference to procedure(const aData: string);
var
FSocketIOMsgNr: Integer;
FSocketIOEventCallback: TDictionary<Integer,TSocketIOCallback>;
FSocketIOEventCallbackRef: TDictionary<Integer,TSocketIOCallbackRef>;
//FSocketIOEventPromises: TDictionary<Integer,TSocketIOPromise>;
2013-11-18 14:27:13 +01:00
FSocketIOErrorRef: TDictionary<Integer,TSocketIOError>;
2013-11-11 21:14:42 +01:00
function WriteConnect(const ASocket: ISocketIOContext): string; overload;
procedure WriteDisConnect(const ASocket: ISocketIOContext);overload;
procedure WritePing(const ASocket: ISocketIOContext);overload;
2013-11-11 21:14:42 +01:00
//
function WriteConnect(const AContext: TIdContext): string; overload;
procedure WriteDisConnect(const AContext: TIdContext);overload;
procedure WritePing(const AContext: TIdContext);overload;
procedure WriteSocketIOMsg(const ASocket: ISocketIOContext; const aRoom, aData: string; aCallback: TSocketIOCallbackRef = nil; const aOnError: TSocketIOError = nil);
procedure WriteSocketIOJSON(const ASocket: ISocketIOContext; const aRoom, aJSON: string; aCallback: TSocketIOCallbackRef = nil; const aOnError: TSocketIOError = nil);
procedure WriteSocketIOEvent(const ASocket: ISocketIOContext; const aRoom, aEventName, aJSONArray: string; aCallback: TSocketIOCallback; const aOnError: TSocketIOError);
procedure WriteSocketIOEventRef(const ASocket: ISocketIOContext; const aRoom, aEventName, aJSONArray: string; aCallback: TSocketIOCallbackRef; const aOnError: TSocketIOError);
2014-08-01 12:19:25 +02:00
function WriteSocketIOEventSync(const ASocket: ISocketIOContext; const aRoom, aEventName, aJSONArray: string; aMaxwait_ms: Cardinal = INFINITE): ISuperObject;
procedure WriteSocketIOResult(const ASocket: ISocketIOContext; aRequestMsgNr: Integer; const aRoom, aData: string);
2013-11-11 21:14:42 +01:00
procedure ProcessSocketIO_XHR(const aGUID: string; const aStrmRequest, aStrmResponse: TStream);
procedure ProcessSocketIORequest(const ASocket: ISocketIOContext; const strmRequest: TMemoryStream);overload;
procedure ProcessSocketIORequest(const ASocket: ISocketIOContext; const aData: string);overload;
2013-11-11 21:14:42 +01:00
procedure ProcessSocketIORequest(const AContext: TIdContext; const strmRequest: TMemoryStream);overload;
procedure ProcessHeatbeatRequest(const ASocket: ISocketIOContext; const aText: string);virtual;
procedure ProcessCloseChannel(const ASocket: ISocketIOContext; const aChannel: string);virtual;
function WriteString(const ASocket: ISocketIOContext; const aText: string): string; virtual;
2013-11-11 21:14:42 +01:00
public
procedure AfterConstruction; override;
destructor Destroy; override;
procedure Lock;
procedure UnLock;
2014-01-31 20:22:10 +01:00
function ConnectionCount: Integer;
2013-11-11 21:14:42 +01:00
function GetSocketIOContext(const AContext: TIdContext): ISocketIOContext;
2013-11-11 21:14:42 +01:00
// procedure EmitEventToAll(const aEventName: string; const aData: ISuperObject; const aCallback: TSocketIOMsgJSON = nil);
function NewConnection(const AContext: TIdContext): ISocketIOContext;overload;
function NewConnection(const aGUID, aPeerIP: string): ISocketIOContext;overload;
2013-11-11 21:14:42 +01:00
procedure FreeConnection(const AContext: TIdContext);overload;
procedure FreeConnection(const ASocket: ISocketIOContext);overload;
2013-11-11 21:14:42 +01:00
property OnSocketIOMsg : TSocketIOMsg read FOnSocketIOMsg write FOnSocketIOMsg;
property OnSocketIOJson : TSocketIOMsgJSON read FOnSocketIOJson write FOnSocketIOJson;
procedure OnEvent (const aEventName: string; const aCallback: TSocketIOEvent);
procedure OnConnection(const aCallback: TSocketIONotify);
procedure OnDisconnect(const aCallback: TSocketIONotify);
2014-06-25 15:30:18 +02:00
property OnEventError: TSocketIOEventError read FOnEventError write FOnEventError;
procedure EnumerateSockets(const aEachSocketCallback: TSocketIONotify);
2013-11-11 21:14:42 +01:00
end;
TIdSocketIOHandling = class(TIdBaseSocketIOHandling)
public
2013-11-18 14:27:13 +01:00
procedure Send(const aMessage: string; const aCallback: TSocketIOMsgJSON = nil; const aOnError: TSocketIOError = nil);
procedure Emit(const aEventName: string; const aData: ISuperObject; const aCallback: TSocketIOMsgJSON = nil; const aOnError: TSocketIOError = nil);overload;
2014-08-01 12:19:25 +02:00
function EmitSync(const aEventName: string; const aData: ISuperObject; aMaxwait_ms: Cardinal = INFINITE): ISuperobject;
//procedure Emit(const aEventName: string; const aData: string; const aCallback: TSocketIOMsgJSON = nil; const aOnError: TSocketIOError = nil);overload;
2013-11-11 21:14:42 +01:00
end;
implementation
uses
2014-06-25 15:30:18 +02:00
StrUtils, IdServerWebsocketContext, IdHTTPWebsocketClient, Windows;
2013-11-11 21:14:42 +01:00
procedure TIdBaseSocketIOHandling.AfterConstruction;
begin
inherited;
FLock := TCriticalSection.Create;
FConnections := TObjectDictionary<TIdContext,ISocketIOContext>.Create([]);
FConnectionsGUID := TObjectDictionary<string,ISocketIOContext>.Create([]);
2013-11-11 21:14:42 +01:00
FOnConnectionList := TSocketIONotifyList.Create;
FOnDisconnectList := TSocketIONotifyList.Create;
FOnEventList := TObjectDictionary<string,TSocketIOEventList>.Create([doOwnsValues]);
FSocketIOEventCallback := TDictionary<Integer,TSocketIOCallback>.Create;
FSocketIOEventCallbackRef := TDictionary<Integer,TSocketIOCallbackRef>.Create;
2013-11-18 14:27:13 +01:00
FSocketIOErrorRef := TDictionary<Integer,TSocketIOError>.Create;
//FSocketIOEventPromises := TDictionary<Integer,TSocketIOPromise>.Create;
2013-11-11 21:14:42 +01:00
end;
2014-01-31 20:22:10 +01:00
function TIdBaseSocketIOHandling.ConnectionCount: Integer;
var
context: ISocketIOContext;
2014-01-31 20:22:10 +01:00
begin
Lock;
try
Result := 0;
//note: is single connection?
for context in FConnections.Values do
begin
if context.IsDisconnected then Continue;
Inc(Result);
end;
for context in FConnectionsGUID.Values do
begin
if context.IsDisconnected then Continue;
Inc(Result);
end;
finally
UnLock;
end;
end;
2013-11-11 21:14:42 +01:00
destructor TIdBaseSocketIOHandling.Destroy;
var squid: string;
idcontext: TIdContext;
begin
Lock;
2013-11-11 21:14:42 +01:00
FSocketIOEventCallback.Free;
FSocketIOEventCallbackRef.Free;
2013-11-18 14:27:13 +01:00
FSocketIOErrorRef.Free;
//FSocketIOEventPromises.Free;
2013-11-11 21:14:42 +01:00
FOnEventList.Free;
FOnConnectionList.Free;
FOnDisconnectList.Free;
while FConnections.Count > 0 do
for idcontext in FConnections.Keys do
begin
//FConnections.Items[idcontext]._Release;
2013-11-11 21:14:42 +01:00
FConnections.ExtractPair(idcontext);
end;
while FConnectionsGUID.Count > 0 do
for squid in FConnectionsGUID.Keys do
begin
//FConnectionsGUID.Items[squid]._Release;
2013-11-11 21:14:42 +01:00
FConnectionsGUID.ExtractPair(squid);
end;
FConnections.Free;
FConnections := nil;
2013-11-11 21:14:42 +01:00
FConnectionsGUID.Free;
UnLock;
FLock.Free;
2013-11-11 21:14:42 +01:00
inherited;
end;
procedure TIdBaseSocketIOHandling.EnumerateSockets(
const aEachSocketCallback: TSocketIONotify);
var socket: ISocketIOContext;
begin
Assert(Assigned(aEachSocketCallback));
Lock;
try
for socket in FConnections.Values do
try
aEachSocketCallback(socket);
except
//continue: e.g. connnection closed etc
end;
for socket in FConnectionsGUID.Values do
try
aEachSocketCallback(socket);
except
//continue: e.g. connnection closed etc
end;
finally
Unlock;
end;
end;
2013-11-11 21:14:42 +01:00
procedure TIdBaseSocketIOHandling.FreeConnection(
const ASocket: ISocketIOContext);
2013-11-11 21:14:42 +01:00
var squid: string;
idcontext: TIdContext;
errorref: TSocketIOError;
i: Integer;
2013-11-11 21:14:42 +01:00
begin
if ASocket = nil then Exit;
Lock;
try
with TSocketIOContext(ASocket) do
begin
Context := nil;
FIOHandler := nil;
FClient := nil;
FHandling := nil;
FGUID := '';
FPeerIP := '';
end;
2013-11-11 21:14:42 +01:00
for idcontext in FConnections.Keys do
2013-11-11 21:14:42 +01:00
begin
if FConnections.Items[idcontext] = ASocket then
begin
FConnections.ExtractPair(idcontext);
//ASocket._Release;
end;
2013-11-11 21:14:42 +01:00
end;
for squid in FConnectionsGUID.Keys do
2013-11-11 21:14:42 +01:00
begin
if FConnectionsGUID.Items[squid] = ASocket then
begin
FConnectionsGUID.ExtractPair(squid);
//ASocket._Release; //use reference count? otherwise AV when used in TThread.Queue
end;
end;
//pending callbacks? then exceute error messages
for i in TSocketIOContext(ASocket).FPendingMessages do
begin
FSocketIOEventCallback.Remove(i);
FSocketIOEventCallbackRef.Remove(i);
if FSocketIOErrorRef.TryGetValue(i, errorref) then
begin
FSocketIOErrorRef.Remove(i);
try
errorref(ASocket, ESocketIODisconnect.ClassName, 'Connection disconnected');
except
end;
end;
2013-11-11 21:14:42 +01:00
end;
finally
Unlock;
2013-11-11 21:14:42 +01:00
end;
end;
function TIdBaseSocketIOHandling.GetSocketIOContext(const AContext: TIdContext): ISocketIOContext;
var
socket: ISocketIOContext;
begin
Result := nil;
Lock;
try
if FConnections.TryGetValue(AContext, socket) then
Exit(socket);
finally
UnLock;
end;
end;
2013-11-11 21:14:42 +01:00
procedure TIdBaseSocketIOHandling.FreeConnection(const AContext: TIdContext);
var
socket: ISocketIOContext;
2013-11-11 21:14:42 +01:00
begin
Lock;
try
if FConnections.TryGetValue(AContext, socket) then
FreeConnection(socket);
finally
UnLock;
end;
end;
procedure TIdBaseSocketIOHandling.Lock;
begin
// Assert(FConnections <> nil);
// System.TMonitor.Enter(Self);
FLock.Enter;
2013-11-11 21:14:42 +01:00
end;
function TIdBaseSocketIOHandling.NewConnection(
const AGUID, aPeerIP: string): ISocketIOContext;
2013-11-11 21:14:42 +01:00
var
socket: TSocketIOContext;
begin
Lock;
try
if not FConnectionsGUID.TryGetValue(AGUID, Result) then
2013-11-11 21:14:42 +01:00
begin
socket := TSocketIOContext.Create;
FConnectionsGUID.Add(AGUID, socket);
end
else
socket := TSocketIOContext(Result);
2013-11-11 21:14:42 +01:00
//socket.Context := AContext;
socket.FGUID := AGUID;
if aPeerIP <> '' then
socket.FPeerIP := aPeerIP;
socket.FHandling := Self;
socket.FConnectSend := False;
socket.FPingSend := False;
Result := socket;
finally
UnLock;
end;
end;
function TIdBaseSocketIOHandling.NewConnection(const AContext: TIdContext): ISocketIOContext;
2013-11-11 21:14:42 +01:00
var
socket: TSocketIOContext;
begin
Lock;
try
if not FConnections.TryGetValue(AContext, Result) then
2013-11-11 21:14:42 +01:00
begin
socket := TSocketIOContext.Create;
FConnections.Add(AContext, socket);
end
else
socket := TSocketIOContext(Result);
2013-11-11 21:14:42 +01:00
socket.Context := AContext;
socket.FHandling := Self;
socket.FConnectSend := False;
socket.FPingSend := False;
Result := socket;
finally
UnLock;
end;
end;
procedure TIdBaseSocketIOHandling.OnConnection(const aCallback: TSocketIONotify);
var context: ISocketIOContext;
2013-11-11 21:14:42 +01:00
begin
FOnConnectionList.Add(aCallback);
Lock;
try
for context in FConnections.Values do
aCallback(context);
for context in FConnectionsGUID.Values do
aCallback(context);
finally
UnLock;
end;
end;
procedure TIdBaseSocketIOHandling.OnDisconnect(const aCallback: TSocketIONotify);
begin
FOnDisconnectList.Add(aCallback);
end;
procedure TIdBaseSocketIOHandling.OnEvent(const aEventName: string;
const aCallback: TSocketIOEvent);
var list: TSocketIOEventList;
begin
if not FOnEventList.TryGetValue(aEventName, list) then
begin
list := TSocketIOEventList.Create;
FOnEventList.Add(aEventName, list);
end;
list.Add(aCallback);
end;
procedure TIdBaseSocketIOHandling.ProcessCloseChannel(
const ASocket: ISocketIOContext; const aChannel: string);
2013-11-11 21:14:42 +01:00
begin
if aChannel <> '' then
//todo: close channel
else if (TSocketIOContext(ASocket).FContext <> nil) then
TSocketIOContext(ASocket).FContext.Connection.Disconnect;
2013-11-11 21:14:42 +01:00
end;
procedure TIdBaseSocketIOHandling.ProcessEvent(
const AContext: ISocketIOContext; const aText: string; aMsgNr: Integer;
2013-11-11 21:14:42 +01:00
aHasCallback: Boolean);
var
json: ISuperObject;
name: string;
args: TSuperArray;
list: TSocketIOEventList;
event: TSocketIOEvent;
callback: ISocketIOCallback;
2013-11-11 21:14:42 +01:00
// socket: TSocketIOContext;
begin
//'5:' [message id ('+')] ':' [message endpoint] ':' [json encoded event]
//5::/chat:{"name":"my other event","args":[{"my":"data"}]}
//5:1+:/chat:{"name":"GetLocations","args":[""]}
json := SO(aText);
// args := nil;
try
name := json.S['name']; //"my other event
args := json.A['args']; //[{"my":"data"}]
if FOnEventList.TryGetValue(name, list) then
begin
if list.Count = 0 then
raise EIdSocketIoUnhandledMessage.Create(aText);
// socket := FConnections.Items[AContext];
if aHasCallback then
callback := TSocketIOCallbackObj.Create(Self, AContext, aMsgNr)
2013-11-11 21:14:42 +01:00
else
callback := nil;
try
2014-06-25 15:30:18 +02:00
for event in list do
2013-11-11 21:14:42 +01:00
try
2014-06-25 15:30:18 +02:00
event(AContext, args, callback);
except on E:Exception do
if Assigned(OnEventError) then
OnEventError(AContext, callback, e)
else
2013-11-11 21:14:42 +01:00
if callback <> nil then
2014-06-25 15:30:18 +02:00
callback.SendResponse( SO(['Error', SO(['msg', e.message])]).AsJSon );
2013-11-11 21:14:42 +01:00
end;
finally
callback := nil;
2013-11-11 21:14:42 +01:00
end;
end
else
raise EIdSocketIoUnhandledMessage.Create(aText);
finally
// args.Free;
json := nil;
end;
end;
procedure TIdBaseSocketIOHandling.ProcessHeatbeatRequest(const ASocket: ISocketIOContext; const aText: string);
2013-11-11 21:14:42 +01:00
begin
with TSocketIOContext(ASocket) do
2013-11-11 21:14:42 +01:00
begin
if PingSend then
PingSend := False //reset, client responded with 2:: heartbeat too
else
begin
PingSend := True; //stop infinite ping response loops
WriteString(ASocket, aText); //write same connect back, e.g. 2::
end;
2013-11-11 21:14:42 +01:00
end;
end;
procedure TIdBaseSocketIOHandling.ProcessSocketIORequest(
const ASocket: ISocketIOContext; const strmRequest: TMemoryStream);
2013-11-11 21:14:42 +01:00
function __ReadToEnd: string;
var
utf8: TBytes;
ilength: Integer;
begin
Result := '';
ilength := strmRequest.Size - strmRequest.Position;
if ilength <= 0 then
Exit;
2013-11-11 21:14:42 +01:00
SetLength(utf8, ilength);
strmRequest.Read(utf8[0], ilength);
Result := TEncoding.UTF8.GetString(utf8);
end;
var str: string;
begin
str := __ReadToEnd;
ProcessSocketIORequest(ASocket, str);
end;
procedure TIdBaseSocketIOHandling.ProcessSocketIORequest(const AContext: TIdContext;
const strmRequest: TMemoryStream);
var
socket: ISocketIOContext;
2013-11-11 21:14:42 +01:00
begin
if not FConnections.TryGetValue(AContext, socket) then
begin
socket := NewConnection(AContext);
end;
ProcessSocketIORequest(socket, strmRequest);
end;
procedure TIdBaseSocketIOHandling.ProcessSocketIO_XHR(const aGUID: string; // const AContext: TIdContext;
const aStrmRequest, aStrmResponse: TStream);
var
socket: ISocketIOContext;
2013-11-11 21:14:42 +01:00
sdata: string;
i, ilength: Integer;
bytes, singlemsg: TBytes;
begin
if not FConnectionsGUID.TryGetValue(aGUID, socket) or
socket.IsDisconnected
then
socket := NewConnection(aGUID, '');
if not TSocketIOContext(socket).FConnectSend then
2013-11-11 21:14:42 +01:00
WriteConnect(socket);
if (aStrmRequest <> nil) and
(aStrmRequest.Size > 0) then
begin
aStrmRequest.Position := 0;
SetLength(bytes, aStrmRequest.Size);
aStrmRequest.Read(bytes[0], aStrmRequest.Size);
if (Length(bytes) > 3) and
(bytes[0] = 239) and (bytes[1] = 191) and (bytes[2] = 189) then
begin
//io.parser.encodePayload(msgs)
//'\ufffd' + packet.length + '\ufffd'
//'<27>17<31>3:::singlemessage<67>52<35>5:4+::{"name":"registerScanner","args":["scanner1"]}'
while bytes <> nil do
begin
i := 3;
//search second '\ufffd'
while not ( (bytes[i+0] = 239) and (bytes[i+1] = 191) and (bytes[i+2] = 189) ) do
begin
Inc(i);
if i+2 > High(bytes) then Exit; //wrong data
end;
//get data between
ilength := StrToInt( TEncoding.UTF8.GetString(bytes, 3, i-3) ); //17
singlemsg := Copy(bytes, i+3, ilength);
bytes := Copy(bytes, i+3 + ilength, Length(bytes));
sdata := TEncoding.UTF8.GetString(singlemsg); //3:::singlemessage
try
ProcessSocketIORequest(socket, sdata);
except
//next
end;
end;
end
else
begin
sdata := TEncoding.UTF8.GetString(bytes);
ProcessSocketIORequest(socket, sdata);
end;
end;
//e.g. POST, no GET?
if aStrmResponse = nil then Exit;
//wait till some response data to be send (long polling)
sdata := TSocketIOContext(socket).WaitForQueue(5 * 1000);
2013-11-11 21:14:42 +01:00
if sdata = '' then
begin
//no data? then send ping
WritePing(socket);
sdata := TSocketIOContext(socket).WaitForQueue(0);
2013-11-11 21:14:42 +01:00
end;
//send response back
if sdata <> '' then
begin
{$WARN SYMBOL_PLATFORM OFF}
// if DebugHook <> 0 then
// Windows.OutputDebugString(PChar('Send: ' + sdata));
2013-11-11 21:14:42 +01:00
bytes := TEncoding.UTF8.GetBytes(sdata);
aStrmResponse.Write(bytes[0], Length(bytes));
end;
end;
procedure TIdBaseSocketIOHandling.UnLock;
begin
//System.TMonitor.Exit(Self);
FLock.Leave;
2013-11-11 21:14:42 +01:00
end;
procedure TIdBaseSocketIOHandling.ProcessSocketIORequest(
const ASocket: ISocketIOContext; const aData: string);
2013-11-11 21:14:42 +01:00
function __GetSocketIOPart(const aData: string; aIndex: Integer): string;
var ipos: Integer;
i: Integer;
begin
//'5::/chat:{"name":"hi!"}'
//0 = 5
//1 =
//2 = /chat
//3 = {"name":"hi!"}
ipos := 0;
for i := 0 to aIndex-1 do
ipos := PosEx(':', aData, ipos+1);
if ipos >= 0 then
begin
Result := Copy(aData, ipos+1, Length(aData));
if aIndex < 3 then // /chat:{"name":"hi!"}'
begin
ipos := PosEx(':', Result, 1); // :{"name":"hi!"}'
if ipos > 0 then
Result := Copy(Result, 1, ipos-1); // /chat
end;
end;
end;
var
str, smsg, schannel, sdata: string;
imsg: Integer;
bCallback: Boolean;
// socket: TSocketIOContext;
callback: TSocketIOCallback;
callbackref: TSocketIOCallbackRef;
callbackobj: ISocketIOCallback;
2013-11-18 14:27:13 +01:00
errorref: TSocketIOError;
error: ISuperObject;
socket: TSocketIOContext;
2013-11-11 21:14:42 +01:00
begin
2014-01-02 14:59:48 +01:00
if ASocket = nil then Exit;
socket := ASocket as TSocketIOContext;
2014-01-02 14:59:48 +01:00
if not FConnections.ContainsValue(socket) and
not FConnectionsGUID.ContainsValue(socket) then
2013-11-11 21:14:42 +01:00
begin
Lock;
try
2014-08-01 12:19:25 +02:00
//ASocket._AddRef;
FConnections.Add(nil, socket); //clients do not have a TIdContext?
2013-11-11 21:14:42 +01:00
finally
UnLock;
end;
end;
str := aData;
if str = '' then Exit;
// if DebugHook <> 0 then
// Windows.OutputDebugString(PChar('Received: ' + str));
2014-01-03 18:48:05 +01:00
while str[1] = #0 do
Delete(str, 1, 1);
2013-11-11 21:14:42 +01:00
//5:1+:/chat:test
smsg := __GetSocketIOPart(str, 1);
imsg := 0;
bCallback := False;
if smsg <> '' then // 1+
begin
imsg := StrToIntDef(ReplaceStr(smsg,'+',''), 0); // 1
bCallback := (Pos('+', smsg) > 1); //trailing +, e.g. 1+
end;
schannel := __GetSocketIOPart(str, 2); // /chat
sdata := __GetSocketIOPart(str, 3); // test
//(0) Disconnect
if StartsStr('0:', str) then
begin
schannel := __GetSocketIOPart(str, 2);
ProcessCloseChannel(socket, schannel);
2013-11-11 21:14:42 +01:00
end
//(1) Connect
//'1::' [path] [query]
else if StartsStr('1:', str) then
begin
//todo: add channel/room to authorized channel/room list
if not socket.ConnectSend then
WriteString(socket, str); //write same connect back, e.g. 1::/chat
2013-11-11 21:14:42 +01:00
end
//(2) Heartbeat
else if StartsStr('2:', str) then
begin
//todo: timer to disconnect client if no ping within time
ProcessHeatbeatRequest(socket, str);
2013-11-11 21:14:42 +01:00
end
//(3) Message (https://github.com/LearnBoost/socket.io-spec#3-message)
//'3:' [message id ('+')] ':' [message endpoint] ':' [data]
//3::/chat:hi
else if StartsStr('3:', str) then
begin
if Assigned(OnSocketIOMsg) then
begin
if bCallback then
begin
callbackobj := TSocketIOCallbackObj.Create(Self, socket, imsg);
2013-11-11 21:14:42 +01:00
try
2013-11-18 14:27:13 +01:00
try
OnSocketIOMsg(socket, sdata, callbackobj); //, imsg, bCallback);
2013-11-18 14:27:13 +01:00
except
on E:Exception do
begin
if not callbackobj.IsResponseSend then
callbackobj.SendResponse( SO(['Error', SO(['Type', e.ClassName, 'Message', e.Message])]).AsJSon );
end;
end;
2013-11-11 21:14:42 +01:00
finally
callbackobj := nil;
2013-11-11 21:14:42 +01:00
end
end
else
OnSocketIOMsg(ASocket, sdata, nil);
end
else
raise EIdSocketIoUnhandledMessage.Create(str);
end
//(4) JSON Message
//'4:' [message id ('+')] ':' [message endpoint] ':' [json]
//4:1::{"a":"b"}
else if StartsStr('4:', str) then
begin
if Assigned(OnSocketIOJson) then
begin
if bCallback then
begin
callbackobj := TSocketIOCallbackObj.Create(Self, socket, imsg);
2013-11-11 21:14:42 +01:00
try
2013-11-18 14:27:13 +01:00
try
OnSocketIOJson(socket, SO(sdata), callbackobj); //, imsg, bCallback);
2013-11-18 14:27:13 +01:00
except
on E:Exception do
begin
if not callbackobj.IsResponseSend then
callbackobj.SendResponse( SO(['Error', SO(['Type', e.ClassName, 'Message', e.Message])]).AsJSon );
end;
end;
2013-11-11 21:14:42 +01:00
finally
callbackobj := nil;
2013-11-11 21:14:42 +01:00
end
end
else
OnSocketIOJson(ASocket, SO(sdata), nil); //, imsg, bCallback);
end
else
raise EIdSocketIoUnhandledMessage.Create(str);
end
//(5) Event
//'5:' [message id ('+')] ':' [message endpoint] ':' [json encoded event]
//5::/chat:{"name":"my other event","args":[{"my":"data"}]}
//5:1+:/chat:{"name":"GetLocations","args":[""]}
else if StartsStr('5:', str) then
begin
//if Assigned(OnSocketIOEvent) then
// OnSocketIOEvent(AContext, sdata, imsg, bCallback);
try
ProcessEvent(socket, sdata, imsg, bCallback);
2013-11-11 21:14:42 +01:00
except
on e:exception do
//
end
end
//(6) ACK
//6::/news:1+["callback"]
//6:::1+["Response"]
else if StartsStr('6:', str) then
begin
smsg := Copy(sdata, 1, Pos('+', sData)-1);
imsg := StrToIntDef(smsg, 0);
sData := Copy(sdata, Pos('+', sData)+1, Length(sData));
TSocketIOContext(ASocket).FPendingMessages.Remove(imsg);
2013-11-18 14:27:13 +01:00
if FSocketIOErrorRef.TryGetValue(imsg, errorref) then
begin
FSocketIOErrorRef.Remove(imsg);
//'[{"Error":{"Message":"Operation aborted","Type":"EAbort"}}]'
if ContainsText(sdata, '{"Error":') then
begin
error := SO(sdata);
if error.IsType(stArray) then
error := error.O['0'];
error := error.O['Error'];
if error.S['Message'] <> '' then
errorref(ASocket, error.S['Type'], error.S['Message'])
else
errorref(ASocket, 'Unknown', sdata);
FSocketIOEventCallback.Remove(imsg);
FSocketIOEventCallbackRef.Remove(imsg);
2013-11-18 14:27:13 +01:00
Exit;
end;
end;
2013-11-11 21:14:42 +01:00
if FSocketIOEventCallback.TryGetValue(imsg, callback) then
begin
FSocketIOEventCallback.Remove(imsg);
if Assigned(callback) then
callback(sdata);
2013-11-11 21:14:42 +01:00
end
else if FSocketIOEventCallbackRef.TryGetValue(imsg, callbackref) then
begin
FSocketIOEventCallbackRef.Remove(imsg);
if Assigned(callbackref) then
callbackref(sdata);
2013-11-11 21:14:42 +01:00
end
else ;
//raise EIdSocketIoUnhandledMessage.Create(str);
end
//(7) Error
else if StartsStr('7:', str) then
raise EIdSocketIoUnhandledMessage.Create(str)
//(8) Noop
else if StartsStr('8:', str) then
begin
//nothing
end
else
raise Exception.CreateFmt('Unsupported data: "%s"', [str]);
end;
function TIdBaseSocketIOHandling.WriteConnect(
const ASocket: ISocketIOContext): string;
2013-11-11 21:14:42 +01:00
var
notify: TSocketIONotify;
begin
Lock;
try
if not FConnections.ContainsValue(ASocket) and
not FConnectionsGUID.ContainsValue(ASocket) then
begin
2014-08-01 12:19:25 +02:00
//ASocket._AddRef;
2013-11-11 21:14:42 +01:00
FConnections.Add(nil, ASocket); //clients do not have a TIdContext?
end;
if not TSocketIOContext(ASocket).ConnectSend then
2013-11-11 21:14:42 +01:00
begin
TSocketIOContext(ASocket).ConnectSend := True;
2013-11-11 21:14:42 +01:00
Result := WriteString(ASocket, '1::');
end;
finally
UnLock;
end;
for notify in FOnConnectionList do
notify(ASocket);
end;
procedure TIdBaseSocketIOHandling.WriteDisConnect(
const ASocket: ISocketIOContext);
2013-11-11 21:14:42 +01:00
var
notify: TSocketIONotify;
begin
if ASocket = nil then Exit;
for notify in FOnDisconnectList do
notify(ASocket);
Lock;
try
if not ASocket.IsDisconnected then
try
WriteString(ASocket, '0::');
except
end;
FreeConnection(ASocket);
finally
UnLock;
end;
end;
procedure TIdBaseSocketIOHandling.WritePing(
const ASocket: ISocketIOContext);
2013-11-11 21:14:42 +01:00
begin
TSocketIOContext(ASocket).PingSend := True;
2013-11-11 21:14:42 +01:00
WriteString(ASocket, '2::') //heartbeat: https://github.com/LearnBoost/socket.io-spec
end;
procedure TIdBaseSocketIOHandling.WriteSocketIOEvent(const ASocket: ISocketIOContext; const aRoom, aEventName,
2013-11-18 14:27:13 +01:00
aJSONArray: string; aCallback: TSocketIOCallback; const aOnError: TSocketIOError);
2013-11-11 21:14:42 +01:00
var
sresult: string;
2014-01-03 18:48:05 +01:00
inr: Integer;
2013-11-11 21:14:42 +01:00
begin
//see TROIndyHTTPWebsocketServer.ProcessSocketIORequest too
//5:1+:/chat:{"name":"GetLocations","args":[""]}
2014-01-03 18:48:05 +01:00
inr := InterlockedIncrement(FSocketIOMsgNr);
2013-11-11 21:14:42 +01:00
if not Assigned(aCallback) then
sresult := Format('5:%d:%s:{"name":"%s", "args":%s}',
2014-01-03 18:48:05 +01:00
[inr, aRoom, aEventName, aJSONArray])
2013-11-11 21:14:42 +01:00
else
begin
//if FSocketIOEventCallback = nil then
// FSocketIOEventCallback := TDictionary<Integer,TSocketIOCallback>.Create;
2013-11-11 21:14:42 +01:00
sresult := Format('5:%d+:%s:{"name":"%s", "args":%s}',
2014-01-03 18:48:05 +01:00
[inr, aRoom, aEventName, aJSONArray]);
FSocketIOEventCallback.Add(inr, aCallback);
TSocketIOContext(ASocket).FPendingMessages.Add(inr);
2013-11-18 14:27:13 +01:00
if Assigned(aOnError) then
2014-01-03 18:48:05 +01:00
FSocketIOErrorRef.Add(inr, aOnError);
2013-11-11 21:14:42 +01:00
end;
WriteString(ASocket, sresult);
end;
procedure TIdBaseSocketIOHandling.WriteSocketIOEventRef(const ASocket: ISocketIOContext;
2013-11-18 14:27:13 +01:00
const aRoom, aEventName, aJSONArray: string; aCallback: TSocketIOCallbackRef; const aOnError: TSocketIOError);
2013-11-11 21:14:42 +01:00
var
sresult: string;
2014-01-03 18:48:05 +01:00
inr: Integer;
2013-11-11 21:14:42 +01:00
begin
//see TROIndyHTTPWebsocketServer.ProcessSocketIORequest too
//5:1+:/chat:{"name":"GetLocations","args":[""]}
2014-01-03 18:48:05 +01:00
inr := InterlockedIncrement(FSocketIOMsgNr);
2013-11-11 21:14:42 +01:00
if not Assigned(aCallback) then
sresult := Format('5:%d:%s:{"name":"%s", "args":%s}',
2014-01-03 18:48:05 +01:00
[inr, aRoom, aEventName, aJSONArray])
2013-11-11 21:14:42 +01:00
else
begin
//if FSocketIOEventCallbackRef = nil then
// FSocketIOEventCallbackRef := TDictionary<Integer,TSocketIOCallbackRef>.Create;
2013-11-11 21:14:42 +01:00
sresult := Format('5:%d+:%s:{"name":"%s", "args":%s}',
2014-01-03 18:48:05 +01:00
[inr, aRoom, aEventName, aJSONArray]);
FSocketIOEventCallbackRef.Add(inr, aCallback);
TSocketIOContext(ASocket).FPendingMessages.Add(inr);
2013-11-18 14:27:13 +01:00
if Assigned(aOnError) then
2014-01-03 18:48:05 +01:00
FSocketIOErrorRef.Add(inr, aOnError);
2013-11-11 21:14:42 +01:00
end;
WriteString(ASocket, sresult);
end;
function TIdBaseSocketIOHandling.WriteSocketIOEventSync(const ASocket: ISocketIOContext; const aRoom, aEventName,
2014-08-01 12:19:25 +02:00
aJSONArray: string; aMaxwait_ms: Cardinal = INFINITE): ISuperObject;
var
sresult: string;
inr: Integer;
promise: TSocketIOPromise;
begin
Result := nil;
if (ASocket = nil) or (ASocket.IsDisconnected) then
raise ESocketIOException.CreateFmt('Socket is not connected, cannot send "%s" request', [aEventName]);
//see TROIndyHTTPWebsocketServer.ProcessSocketIORequest too
//5:1+:/chat:{"name":"GetLocations","args":[""]}
inr := InterlockedIncrement(FSocketIOMsgNr);
// if FSocketIOEventPromises = nil then
// FSocketIOEventPromises := TDictionary<Integer,TSocketIOPromise>.Create;
promise := TSocketIOPromise.Create;
// FSocketIOEventPromises.Add(inr, promise);
sresult := Format('5:%d+:%s:{"name":"%s", "args":%s}',
[inr, aRoom, aEventName, aJSONArray]);
Lock;
try
FSocketIOEventCallbackRef.Add(inr,
procedure(const aData: string)
begin
promise.Success := True;
promise.Error := nil;
promise.Data := SO(aData);
promise.Done := True;
promise.Event.SetEvent;
end);
FSocketIOErrorRef.Add(inr,
procedure(const ASocket: ISocketIOContext; const aErrorClass, aErrorMessage: string)
begin
promise.Success := False;
promise.Error := ESocketIOException.Create(aErrorClass + ': ' + aErrorMessage);
promise.Data := nil;
promise.Done := True;
promise.Event.SetEvent;
end);
TSocketIOContext(ASocket).FPendingMessages.Add(inr);
finally
Unlock;
end;
try
try
if ASocket.IsDisconnected then
raise ESocketIOException.CreateFmt('Socket is disconnected, cannot send "%s" request', [aEventName])
else
WriteString(ASocket, sresult);
//wait for callback
if promise.Event.WaitFor(aMaxwait_ms) = wrSignaled then
Assert(promise.Done)
else
//timeout
raise ESocketIOTimeout.CreateFmt('No response received for "%s" request', [aEventName]);
except
Lock;
try
FSocketIOEventCallbackRef.Remove(inr);
FSocketIOErrorRef.Remove(inr);
TSocketIOContext(ASocket).FPendingMessages.Remove(inr);
finally
UnLock;
end;
raise;
end;
Result := promise.Data;
if promise.Error <> nil then
begin
Assert(not promise.Success);
raise promise.Error;
end
else
Assert(promise.Success);
finally
promise.Free;
end;
end;
procedure TIdBaseSocketIOHandling.WriteSocketIOJSON(const ASocket: ISocketIOContext;
2013-11-18 14:27:13 +01:00
const aRoom, aJSON: string; aCallback: TSocketIOCallbackRef = nil; const aOnError: TSocketIOError = nil);
2013-11-11 21:14:42 +01:00
var
sresult: string;
2014-01-03 18:48:05 +01:00
inr: Integer;
2013-11-11 21:14:42 +01:00
begin
//see TROIndyHTTPWebsocketServer.ProcessSocketIORequest too
//4:1::{"a":"b"}
2014-01-03 18:48:05 +01:00
inr := InterlockedIncrement(FSocketIOMsgNr);
2013-11-11 21:14:42 +01:00
if not Assigned(aCallback) then
2014-01-03 18:48:05 +01:00
sresult := Format('4:%d:%s:%s', [inr, aRoom, aJSON])
2013-11-11 21:14:42 +01:00
else
begin
//if FSocketIOEventCallbackRef = nil then
// FSocketIOEventCallbackRef := TDictionary<Integer,TSocketIOCallbackRef>.Create;
2013-11-11 21:14:42 +01:00
sresult := Format('4:%d+:%s:%s',
2014-01-03 18:48:05 +01:00
[inr, aRoom, aJSON]);
FSocketIOEventCallbackRef.Add(inr, aCallback);
TSocketIOContext(ASocket).FPendingMessages.Add(inr);
2013-11-18 14:27:13 +01:00
if Assigned(aOnError) then
2014-01-03 18:48:05 +01:00
FSocketIOErrorRef.Add(inr, aOnError);
2013-11-11 21:14:42 +01:00
end;
WriteString(ASocket, sresult);
end;
procedure TIdBaseSocketIOHandling.WriteSocketIOMsg(const ASocket: ISocketIOContext;
2013-11-18 14:27:13 +01:00
const aRoom, aData: string; aCallback: TSocketIOCallbackRef = nil; const aOnError: TSocketIOError = nil);
2013-11-11 21:14:42 +01:00
var
sresult: string;
2014-01-03 18:48:05 +01:00
inr: Integer;
2013-11-11 21:14:42 +01:00
begin
//see TROIndyHTTPWebsocketServer.ProcessSocketIORequest too
//3::/chat:hi
2014-01-03 18:48:05 +01:00
inr := InterlockedIncrement(FSocketIOMsgNr);
2013-11-11 21:14:42 +01:00
if not Assigned(aCallback) then
2014-01-03 18:48:05 +01:00
sresult := Format('3:%d:%s:%s', [inr, aRoom, aData])
2013-11-11 21:14:42 +01:00
else
begin
//if FSocketIOEventCallbackRef = nil then
// FSocketIOEventCallbackRef := TDictionary<Integer,TSocketIOCallbackRef>.Create;
2013-11-11 21:14:42 +01:00
sresult := Format('3:%d+:%s:%s',
2014-01-03 18:48:05 +01:00
[inr, aRoom, aData]);
FSocketIOEventCallbackRef.Add(inr, aCallback);
TSocketIOContext(ASocket).FPendingMessages.Add(inr);
2013-11-18 14:27:13 +01:00
if Assigned(aOnError) then
2014-01-03 18:48:05 +01:00
FSocketIOErrorRef.Add(inr, aOnError);
2013-11-11 21:14:42 +01:00
end;
WriteString(ASocket, sresult);
end;
procedure TIdBaseSocketIOHandling.WriteSocketIOResult(const ASocket: ISocketIOContext;
2013-11-11 21:14:42 +01:00
aRequestMsgNr: Integer; const aRoom, aData: string);
var
sresult: string;
begin
//6::/news:2+["callback"]
sresult := Format('6::%s:%d+[%s]', [aRoom, aRequestMsgNr, aData]);
WriteString(ASocket, sresult);
end;
function TIdBaseSocketIOHandling.WriteString(const ASocket: ISocketIOContext;
2013-11-11 21:14:42 +01:00
const aText: string): string;
begin
if ASocket = nil then Exit;
ASocket.Lock;
try
with TSocketIOContext(ASocket) do
2013-11-11 21:14:42 +01:00
begin
if FIOHandler = nil then
begin
if FContext <> nil then
FIOHandler := (FContext as TIdServerWSContext).IOHandler;
end;
2013-11-11 21:14:42 +01:00
if (FIOHandler <> nil) then
begin
//Assert(ASocket.FIOHandler.IsWebsocket);
// if DebugHook <> 0 then
// Windows.OutputDebugString(PChar('Send: ' + aText));
FIOHandler.Write(aText);
end
else if GUID <> '' then
begin
QueueData(aText);
Result := aText; //for xhr-polling the data must be send using responseinfo instead of direct write!
end;
//else //disconnected
// Assert(False, 'disconnected');
end;
2013-11-11 21:14:42 +01:00
finally
ASocket.UnLock;
end;
end;
{ TSocketIOCallbackObj }
constructor TSocketIOCallbackObj.Create(aHandling: TIdBaseSocketIOHandling;
aSocket: ISocketIOContext; aMsgNr: Integer);
begin
FHandling := aHandling;
FSocket := aSocket;
FMsgNr := aMsgNr;
inherited Create();
end;
2013-11-18 14:27:13 +01:00
function TSocketIOCallbackObj.IsResponseSend: Boolean;
begin
Result := (FMsgNr < 0);
end;
2013-11-11 21:14:42 +01:00
procedure TSocketIOCallbackObj.SendResponse(const aResponse: string);
begin
FHandling.WriteSocketIOResult(FSocket, FMsgNr, '', aResponse);
2013-11-18 14:27:13 +01:00
FMsgNr := -1;
2013-11-11 21:14:42 +01:00
end;
{ TSocketIOContext }
procedure TSocketIOContext.AfterConstruction;
begin
inherited;
FLock := TCriticalSection.Create;
FQueue := TList<string>.Create;
FPendingMessages := TList<Int64>.Create;
end;
2013-11-11 21:14:42 +01:00
constructor TSocketIOContext.Create(aClient: TIdHTTP);
begin
FClient := aClient;
if aClient is TIdHTTPWebsocketClient then
begin
FHandling := (aClient as TIdHTTPWebsocketClient).SocketIO;
end;
FIOHandler := (aClient as TIdHTTPWebsocketClient).IOHandler;
end;
destructor TSocketIOContext.Destroy;
begin
2014-01-03 18:48:05 +01:00
Lock;
2013-11-11 21:14:42 +01:00
FEvent.Free;
FreeAndNil(FQueue);
UnLock;
FLock.Free;
if OwnsCustomData then
FCustomData.Free;
FPendingMessages.Free;
2013-11-11 21:14:42 +01:00
inherited;
end;
procedure TSocketIOContext.EmitEvent(const aEventName, aData: string;
2013-11-18 14:27:13 +01:00
const aCallback: TSocketIOMsgJSON; const aOnError: TSocketIOError);
2013-11-11 21:14:42 +01:00
begin
Assert(FHandling <> nil);
2013-11-11 21:14:42 +01:00
if not Assigned(aCallback) then
FHandling.WriteSocketIOEvent(Self, '', aEventName, '[' + aData + ']', nil, nil)
2013-11-11 21:14:42 +01:00
else
begin
FHandling.WriteSocketIOEventRef(Self, '', aEventName, '[' + aData + ']',
2013-11-11 21:14:42 +01:00
procedure(const aData: string)
begin
aCallback(Self, SO(aData), nil);
2013-11-18 14:27:13 +01:00
end, aOnError);
2013-11-11 21:14:42 +01:00
end;
end;
procedure TSocketIOContext.EmitEvent(const aEventName: string; const aData: ISuperObject;
const aCallback: TSocketIOMsgJSON; const aOnError: TSocketIOError);
begin
if aData <> nil then
EmitEvent(aEventName, aData.AsJSon, aCallback, aOnError)
else
EmitEvent(aEventName, '', aCallback, aOnError);
end;
function TSocketIOContext.GetCustomData: TObject;
begin
Result := FCustomData;
end;
function TSocketIOContext.GetOwnsCustomData: Boolean;
begin
Result := FOwnsCustomData;
end;
2013-11-11 21:14:42 +01:00
function TSocketIOContext.IsDisconnected: Boolean;
begin
Result := (FClient = nil) and (FContext = nil) and (FGUID = '');
end;
procedure TSocketIOContext.Lock;
begin
// Assert(FQueue <> nil);
// System.TMonitor.Enter(Self);
FLock.Enter;
2013-11-11 21:14:42 +01:00
end;
constructor TSocketIOContext.Create;
begin
//
end;
function TSocketIOContext.PeerIP: string;
begin
Result := FPeerIP;
if FContext is TIdServerWSContext then
Result := (FContext as TIdServerWSContext).Binding.PeerIP
else if FIOHandler <> nil then
Result := FIOHandler.Binding.PeerIP;
end;
function TSocketIOContext.PeerPort: Integer;
begin
Result := 0;
if FContext is TIdServerWSContext then
Result := (FContext as TIdServerWSContext).Binding.PeerPort
else if FIOHandler <> nil then
Result := FIOHandler.Binding.PeerPort
end;
procedure TSocketIOContext.QueueData(const aData: string);
begin
if FEvent = nil then
FEvent := TEvent.Create;
FQueue.Add(aData);
//max 1000 items in queue (otherwise infinite mem leak possible?)
while FQueue.Count > 1000 do
FQueue.Delete(0);
2013-11-11 21:14:42 +01:00
FEvent.SetEvent;
end;
function TSocketIOContext.ResourceName: string;
begin
if FContext is TIdServerWSContext then
Result := (FContext as TIdServerWSContext).ResourceName
else if FClient <> nil then
Result := (FClient as TIdHTTPWebsocketClient).WSResourceName
end;
procedure TSocketIOContext.Send(const aData: string;
2013-11-18 14:27:13 +01:00
const aCallback: TSocketIOMsgJSON; const aOnError: TSocketIOError);
2013-11-11 21:14:42 +01:00
begin
if not Assigned(aCallback) then
FHandling.WriteSocketIOMsg(Self, '', aData)
else
begin
FHandling.WriteSocketIOMsg(Self, '', aData,
procedure(const aData: string)
begin
aCallback(Self, SO(aData), nil);
2013-11-18 14:27:13 +01:00
end, aOnError);
2013-11-11 21:14:42 +01:00
end;
end;
procedure TSocketIOContext.SendJSON(const aJSON: ISuperObject;
2013-11-18 14:27:13 +01:00
const aCallback: TSocketIOMsgJSON; const aOnError: TSocketIOError);
2013-11-11 21:14:42 +01:00
begin
if not Assigned(aCallback) then
FHandling.WriteSocketIOJSON(Self, '', aJSON.AsJSon())
else
begin
FHandling.WriteSocketIOMsg(Self, '', aJSON.AsJSon(),
procedure(const aData: string)
begin
aCallback(Self, SO(aData), nil);
2013-11-18 14:27:13 +01:00
end, aOnError);
2013-11-11 21:14:42 +01:00
end;
end;
procedure TSocketIOContext.ServerContextDestroy(AContext: TIdContext);
begin
Self.Context := nil;
Self.FIOHandler := nil;
2014-01-23 10:17:13 +01:00
if FHandling <> nil then
Self.FHandling.FreeConnection(AContext);
2013-11-11 21:14:42 +01:00
end;
procedure TSocketIOContext.SetConnectSend(const Value: Boolean);
begin
FConnectSend := Value;
end;
procedure TSocketIOContext.SetContext(const Value: TIdContext);
begin
if (Value <> FContext) and (Value = nil) and
(FContext is TIdServerWSContext) then
(FContext as TIdServerWSContext).OnDestroy := nil;
FContext := Value;
if FContext is TIdServerWSContext then
(FContext as TIdServerWSContext).OnDestroy := Self.ServerContextDestroy;
end;
procedure TSocketIOContext.SetCustomData(const Value: TObject);
begin
FCustomData := Value;
end;
procedure TSocketIOContext.SetOwnsCustomData(const Value: Boolean);
begin
FOwnsCustomData := Value;
end;
2013-11-11 21:14:42 +01:00
procedure TSocketIOContext.SetPingSend(const Value: Boolean);
begin
FPingSend := Value;
end;
procedure TSocketIOContext.UnLock;
begin
//System.TMonitor.Exit(Self);
FLock.Leave;
2013-11-11 21:14:42 +01:00
end;
function TSocketIOContext.WaitForQueue(aTimeout_ms: Integer): string;
begin
if (FEvent = nil) or (FQueue = nil) then
begin
Lock;
try
if FEvent = nil then
FEvent := TEvent.Create;
if FQueue = nil then
FQueue := TList<string>.Create;
finally
UnLock;
end;
end;
if (FQueue.Count > 0) or
(FEvent.WaitFor(aTimeout_ms) = wrSignaled) then
begin
Lock;
try
FEvent.ResetEvent;
if (FQueue.Count > 0) then
begin
Result := FQueue.First;
FQueue.Delete(0);
end;
finally
UnLock;
end;
end;
end;
function TIdBaseSocketIOHandling.WriteConnect(const AContext: TIdContext): string;
var
socket: ISocketIOContext;
2013-11-11 21:14:42 +01:00
begin
//if not FConnections.TryGetValue(AContext, socket) then
socket := NewConnection(AContext);
Result := WriteConnect(socket);
end;
procedure TIdBaseSocketIOHandling.WriteDisConnect(const AContext: TIdContext);
var
socket: ISocketIOContext;
2013-11-11 21:14:42 +01:00
begin
if not FConnections.TryGetValue(AContext, socket) then
socket := NewConnection(AContext);
WriteDisConnect(socket);
end;
procedure TIdBaseSocketIOHandling.WritePing(const AContext: TIdContext);
var
socket: ISocketIOContext;
2013-11-11 21:14:42 +01:00
begin
if not FConnections.TryGetValue(AContext, socket) then
socket := NewConnection(AContext);
WritePing(socket);
end;
{ TIdSocketIOHandling }
procedure TIdSocketIOHandling.Emit(const aEventName: string;
2013-11-18 14:27:13 +01:00
const aData: ISuperObject; const aCallback: TSocketIOMsgJSON; const aOnError: TSocketIOError);
2013-11-11 21:14:42 +01:00
var
context: ISocketIOContext;
2013-11-11 21:14:42 +01:00
jsonarray: string;
isendcount: Integer;
begin
2014-01-31 20:22:10 +01:00
if aData <> nil then
begin
if aData.IsType(stArray) then
jsonarray := aData.AsString
else if aData.IsType(stString) then
jsonarray := '["' + aData.AsString + '"]'
else
jsonarray := '[' + aData.AsString + ']';
end;
2013-11-11 21:14:42 +01:00
Lock;
try
isendcount := 0;
//note: client has single connection?
2013-11-11 21:14:42 +01:00
for context in FConnections.Values do
begin
if context.IsDisconnected then Continue;
if not Assigned(aCallback) then
2013-11-18 14:27:13 +01:00
WriteSocketIOEvent(context, ''{no room}, aEventName, jsonarray, nil, nil)
2013-11-11 21:14:42 +01:00
else
WriteSocketIOEventRef(context, ''{no room}, aEventName, jsonarray,
procedure(const aData: string)
begin
aCallback(context, SO(aData), nil);
2013-11-18 14:27:13 +01:00
end, aOnError);
2013-11-11 21:14:42 +01:00
Inc(isendcount);
end;
for context in FConnectionsGUID.Values do
begin
if context.IsDisconnected then Continue;
if not Assigned(aCallback) then
2013-11-18 14:27:13 +01:00
WriteSocketIOEvent(context, ''{no room}, aEventName, jsonarray, nil, nil)
2013-11-11 21:14:42 +01:00
else
WriteSocketIOEventRef(context, ''{no room}, aEventName, jsonarray,
procedure(const aData: string)
begin
aCallback(context, SO(aData), nil);
2013-11-18 14:27:13 +01:00
end, aOnError);
2013-11-11 21:14:42 +01:00
Inc(isendcount);
end;
if isendcount = 0 then
2014-01-31 20:22:10 +01:00
raise EIdSocketIoUnhandledMessage.Create('Cannot emit: no socket.io connections!');
2013-11-11 21:14:42 +01:00
finally
UnLock;
end;
end;
2014-08-01 12:19:25 +02:00
function TIdSocketIOHandling.EmitSync(const aEventName: string; const aData: ISuperObject; aMaxwait_ms: Cardinal = INFINITE): ISuperobject;
var
firstcontext, context: ISocketIOContext;
jsonarray: string;
isendcount: Integer;
begin
if aData <> nil then
begin
if aData.IsType(stArray) then
jsonarray := aData.AsString
else if aData.IsType(stString) then
jsonarray := '["' + aData.AsString + '"]'
else
jsonarray := '[' + aData.AsString + ']';
end;
Lock;
try
isendcount := 0;
//note: client has single connection?
for context in FConnections.Values do
begin
if context.IsDisconnected then Continue;
firstcontext := context;
Inc(isendcount);
end;
for context in FConnectionsGUID.Values do
begin
if context.IsDisconnected then Continue;
firstcontext := context;
Inc(isendcount);
end;
//todo: use multiple promises?
if isendcount > 1 then
raise EIdSocketIoUnhandledMessage.Create('Cannot emit synchronized to more than one connection!');
finally
UnLock;
end;
Result := WriteSocketIOEventSync(firstcontext, ''{no room}, aEventName, jsonarray, aMaxwait_ms);
end;
2013-11-11 21:14:42 +01:00
procedure TIdSocketIOHandling.Send(const aMessage: string;
2013-11-18 14:27:13 +01:00
const aCallback: TSocketIOMsgJSON; const aOnError: TSocketIOError);
2013-11-11 21:14:42 +01:00
var
context: ISocketIOContext;
2013-11-11 21:14:42 +01:00
isendcount: Integer;
begin
Lock;
try
isendcount := 0;
//note: is single connection?
for context in FConnections.Values do
begin
if context.IsDisconnected then Continue;
if not Assigned(aCallback) then
WriteSocketIOMsg(context, ''{no room}, aMessage, nil)
else
WriteSocketIOMsg(context, ''{no room}, aMessage,
procedure(const aData: string)
begin
aCallback(context, SO(aData), nil);
2013-11-18 14:27:13 +01:00
end, aOnError);
2013-11-11 21:14:42 +01:00
Inc(isendcount);
end;
for context in FConnectionsGUID.Values do
begin
if context.IsDisconnected then Continue;
if not Assigned(aCallback) then
WriteSocketIOMsg(context, ''{no room}, aMessage, nil)
else
WriteSocketIOMsg(context, ''{no room}, aMessage,
procedure(const aData: string)
begin
aCallback(context, SO(aData), nil);
2013-11-18 14:27:13 +01:00
end, aOnError);
2013-11-11 21:14:42 +01:00
Inc(isendcount);
end;
if isendcount = 0 then
2014-01-31 20:22:10 +01:00
raise EIdSocketIoUnhandledMessage.Create('Cannot send: no socket.io connections!');
2013-11-11 21:14:42 +01:00
finally
UnLock;
end;
end;
{ TSocketIOPromise }
procedure TSocketIOPromise.AfterConstruction;
begin
inherited;
Event := TEvent.Create();
end;
destructor TSocketIOPromise.Destroy;
begin
Event.Free;
inherited;
end;
2013-11-11 21:14:42 +01:00
end.