DelphiWebsockets/uROHTTPWebsocketServer.pas
André Mussche 7a59f2b272 local sync
2014-01-02 14:59:48 +01:00

333 lines
11 KiB
ObjectPascal

unit uROHTTPWebsocketServer;
interface
uses
Classes, IdServerIOHandlerWebsocket, IdIOHandlerWebsocket,
uROIndyHTTPServer, uROClientIntf, uROServer, uROHTTPDispatch,
IdContext, IdCustomHTTPServer, IdCustomTCPServer, uROHash, uROServerIntf,
IdServerWebsocketContext, IdServerSocketIOHandling,
IdServerWebsocketHandling;
type
TROTransportContext = class;
TROIndyHTTPWebsocketServer = class(TROIndyHTTPServer)
private
FOnCustomChannelExecute: TWebsocketChannelRequest;
FSocketIO: TIdServerSocketIOHandling_Ext;
function GetSocketIO: TIdServerSocketIOHandling;
protected
FROTransportContexts: TInterfaceList;
procedure InternalServerConnect(AThread: TIdContext); override;
procedure InternalServerDisConnect(AThread: TIdContext); virtual;
procedure InternalServerCommandGet(AThread: TIdThreadClass;
ARequestInfo: TIdHTTPRequestInfo; AResponseInfo: TIdHTTPResponseInfo); override;
procedure ProcessRemObjectsRequest(const AThread: TIdContext; const strmRequest: TMemoryStream; const strmResponse: TMemoryStream);
function GetDispatchersClass: TROMessageDispatchersClass; override;
public
procedure AfterConstruction; override;
destructor Destroy; override;
procedure Loaded; override;
property SocketIO: TIdServerSocketIOHandling read GetSocketIO;
property OnCustomChannelExecute: TWebsocketChannelRequest read FOnCustomChannelExecute write FOnCustomChannelExecute;
end;
TROHTTPDispatcher_Websocket = class(TROHTTPDispatcher)
public
function CanHandleMessage(const aTransport: IROTransport; aRequeststream : TStream): boolean; override;
end;
TROHTTPMessageDispatchers_WebSocket = class(TROHTTPMessageDispatchers)
protected
function GetDispatcherClass : TROMessageDispatcherClass; override;
end;
TROTransportContext = class(TInterfacedObject,
IROTransport, IROTCPTransport,
IROActiveEventServer)
private
FROServer: TROIndyHTTPServer;
FIdContext: TIdServerWSContext;
FEventCount: Integer;
FClientId: TGUID;
private
class var FGlobalEventCount: Integer;
protected
{IROTransport}
function GetTransportObject: TObject;
{IROTCPTransport}
function GetClientAddress : string;
{IROActiveEventServer}
procedure EventsRegistered(aSender : TObject; aClient: TGUID);
procedure DispatchEvent(anEventDataItem : TROEventData; aSessionReference : TGUID; aSender: TObject); // asender is TROEventRepository
public
//constructor Create(aROServer: TROIndyHTTPServer; aIOHandler: TIdIOHandlerWebsocket);
constructor Create(aROServer: TROIndyHTTPServer; aIdContext: TIdServerWSContext);
property Context: TIdServerWSContext read FIdContext;
property ClientId: TGUID read FClientId write FClientId;
end;
procedure Register;
implementation
uses
SysUtils, IdCoderMIME, Windows, uROEventRepository, uROSessions, uROClient,
uROClasses, StrUtils, uROIdServerWebsocketHandling;
procedure Register;
begin
RegisterComponents('RBK', [TROIndyHTTPWebsocketServer]);
end;
procedure TROIndyHTTPWebsocketServer.AfterConstruction;
begin
inherited;
FSocketIO := TIdServerSocketIOHandling_Ext.Create;
FROTransportContexts := TInterfaceList.Create;
IndyServer.ContextClass := TROIdServerWSContext;
if Self.IndyServer.IOHandler = nil then
IndyServer.IOHandler := TIdServerIOHandlerWebsocket.Create(Self);
IndyServer.OnDisconnect := InternalServerDisConnect;
end;
destructor TROIndyHTTPWebsocketServer.Destroy;
begin
inherited;
FSocketIO.Free;
FROTransportContexts.Free;
end;
function TROIndyHTTPWebsocketServer.GetDispatchersClass: TROMessageDispatchersClass;
begin
Result := TROHTTPMessageDispatchers_Websocket;
end;
function TROIndyHTTPWebsocketServer.GetSocketIO: TIdServerSocketIOHandling;
begin
Result := FSocketIO;
end;
procedure TROIndyHTTPWebsocketServer.InternalServerCommandGet(AThread: TIdThreadClass;
ARequestInfo: TIdHTTPRequestInfo; AResponseInfo: TIdHTTPResponseInfo);
begin
(AThread as TIdServerWSContext).OnCustomChannelExecute := Self.OnCustomChannelExecute;
(AThread as TIdServerWSContext).SocketIO := Self.FSocketIO;
(AThread as TROIdServerWSContext).OnRemObjectsRequest := Self.ProcessRemObjectsRequest;
if not TROIdServerWebsocketHandling.ProcessServerCommandGet(AThread as TIdServerWSContext, ARequestInfo, AResponseInfo) then
inherited InternalServerCommandGet(AThread, ARequestInfo, AResponseInfo)
end;
procedure TROIndyHTTPWebsocketServer.InternalServerConnect(AThread: TIdContext);
begin
inherited;
(AThread as TIdServerWSContext).OnCustomChannelExecute := Self.OnCustomChannelExecute;
(AThread as TROIdServerWSContext).OnRemObjectsRequest := Self.ProcessRemObjectsRequest;
end;
procedure TROIndyHTTPWebsocketServer.InternalServerDisConnect(
AThread: TIdContext);
var
transport: TROTransportContext;
begin
transport := AThread.Data as TROTransportContext;
if transport <> nil then
FROTransportContexts.Remove(transport);
//transport._Release;
AThread.Data := nil;
end;
procedure TROIndyHTTPWebsocketServer.Loaded;
begin
//do before inherited in case of designtime connection
if Self.IndyServer.IOHandler = nil then
IndyServer.IOHandler := TIdServerIOHandlerWebsocket.Create(Self);
inherited;
end;
procedure TROIndyHTTPWebsocketServer.ProcessRemObjectsRequest(
const AThread: TIdContext; const strmRequest: TMemoryStream; const strmResponse: TMemoryStream);
var
cWSNR: array[0..High(C_ROWSNR)] of AnsiChar;
msg: TROMessageDispatcher;
iMsgNr: Integer;
imsg: IROMessage;
transport: TROTransportContext;
begin
if strmRequest.Size < Length(C_ROWSNR) + SizeOf(iMsgNr) then Exit;
//read messagenr from the end
strmRequest.Position := strmRequest.Size - Length(C_ROWSNR) - SizeOf(iMsgNr);
strmRequest.Read(cWSNR[0], Length(C_ROWSNR));
if (cWSNR <> C_ROWSNR) then Exit;
strmRequest.Read(iMsgNr, SizeOf(iMsgNr));
strmRequest.Position := 0;
//trunc extra data
strmRequest.Size := strmRequest.Size - Length(C_ROWSNR) - SizeOf(iMsgNr);
transport := AThread.Data as TROTransportContext;
//no RO transport object already made?
if transport = nil then
begin
//create IROTransport object
transport := TROTransportContext.Create(Self, AThread as TIdServerWSContext);
//(transport as IROTransport)._AddRef;
FROTransportContexts.Add(transport);
//attach RO transport to indy context
AThread.Data := transport;
//todo: enveloppes
//read client GUID the first time (needed to be able to send RO events)
msg := Self.Dispatchers.FindDispatcher(transport, strmRequest);
if msg = nil then
raise EROException.Create('No suiteable message dispatcher found!');
imsg := (msg.MessageIntf as IROMessageCloneable).Clone;
imsg.InitializeRead(transport);
imsg.ReadFromStream(strmRequest);
transport.ClientId := imsg.ClientID;
imsg := nil;
Assert(not IsEqualGUID(transport.ClientID, EmptyGUID));
end;
//EXECUTE FUNCTION
Self.DispatchMessage(transport, strmRequest, strmResponse);
//write number at end
strmResponse.Position := strmResponse.Size;
strmResponse.Write(C_ROWSNR, Length(C_ROWSNR));
strmResponse.Write(iMsgNr, SizeOf(iMsgNr));
strmResponse.Position := 0;
end;
{ TROTransport }
constructor TROTransportContext.Create(aROServer: TROIndyHTTPServer;
aIdContext: TIdServerWSContext);
begin
FROServer := aROServer;
FIdContext := aIdContext;
end;
procedure TROTransportContext.EventsRegistered(aSender: TObject; aClient: TGUID);
begin
//
end;
procedure TROTransportContext.DispatchEvent(anEventDataItem: TROEventData;
aSessionReference: TGUID; aSender: TObject);
var
i: Integer;
LContext: TIdContext;
transport: TROTransportContext;
l: TList;
ws: TIdIOHandlerWebsocket;
cWSNR: array[0..High(C_ROWSNR)] of AnsiChar;
begin
l := FROServer.IndyServer.Contexts.LockList;
try
if l.Count <= 0 then Exit;
anEventDataItem.Data.Position := anEventDataItem.Data.Size - Length(C_ROWSNR) - SizeOf(FEventCount);
anEventDataItem.Data.Read(cWSNR[0], Length(cWSNR));
//event number not written already?
if cWSNR <> C_ROWSNR then
begin
//new event nr
FEventCount := -1 * InterlockedIncrement(FGlobalEventCount); //negative = event, positive is normal RO message
//overflow? then start again from 0
if FEventCount > 0 then
begin
InterlockedExchange(FGlobalEventCount, 0);
FEventCount := -1 * InterlockedIncrement(FGlobalEventCount); //negative = event, positive is normal RO message
end;
Assert(FEventCount < 0);
//write nr at end of message
anEventDataItem.Data.Position := anEventDataItem.Data.Size;
anEventDataItem.Data.Write(C_ROWSNR, Length(C_ROWSNR));
anEventDataItem.Data.Write(FEventCount, SizeOf(FEventCount));
anEventDataItem.Data.Position := 0;
end;
//search specific client
for i := 0 to l.Count - 1 do
begin
LContext := TIdContext(l.Items[i]);
transport := LContext.Data as TROTransportContext;
if transport = nil then Continue;
if not IsEqualGUID(transport.ClientId, aSessionReference) then Continue;
//direct write event data
ws := (LContext.Connection.IOHandler as TIdIOHandlerWebsocket);
if not ws.IsWebsocket then Exit;
ws.Lock;
try
try ws.Write(anEventDataItem.Data, wdtBinary) except {continue with other connections} end;
finally
ws.Unlock;
end;
end;
finally
anEventDataItem.RemoveRef;
FROServer.IndyServer.Contexts.UnlockList;
end;
end;
function TROTransportContext.GetClientAddress: string;
begin
Result := FIdContext.Binding.PeerIP;
end;
function TROTransportContext.GetTransportObject: TObject;
begin
Result := FROServer;
end;
{ TROHTTPMessageDispatchers_WebSocket }
function TROHTTPMessageDispatchers_WebSocket.GetDispatcherClass: TROMessageDispatcherClass;
begin
result := TROHTTPDispatcher_Websocket;
end;
{ TROHTTPDispatcher_Websocket }
function TROHTTPDispatcher_Websocket.CanHandleMessage(
const aTransport: IROTransport; aRequeststream: TStream): boolean;
var
tcp: IROTCPTransport;
buf: array [0..5] of AnsiChar;
begin
if aRequeststream = nil then result := FALSE else // for preventing warning in FPC
result := FALSE;
if not Enabled or
not Supports(aTransport, IROTCPTransport, tcp)
then
Exit;
if (tcp as TROTransportContext).FIdContext.IOHandler.IsWebsocket then
begin
//we can handle all kind of messages, independent on the path, so check which kind of message we have
Result := Self.Message.IsValidMessage((aRequeststream as TMemoryStream).Memory, aRequeststream.Size);
//goes wrong with enveloppes!
//TROMessage.Envelopes_ProcessIncoming
if not Result and
(aRequeststream.Size > 6) then
begin
aRequeststream.Read(buf,6);
Result := (buf[0] = EnvelopeSignature[0]) and
(buf[1] = EnvelopeSignature[1]) and
(buf[2] = EnvelopeSignature[2]) and
(buf[3] = EnvelopeSignature[3]) and
(buf[4] = EnvelopeSignature[4]);
aRequeststream.Position := 0;
end;
end
else
Result := inherited CanHandleMessage(aTransport, aRequeststream);
end;
end.