cmueller/DelphiWebsockets
1
0
Fork 0
Code Issues Pull requests Projects Releases Wiki Activity

Compare commits

merge into: cmueller:master
Branches Tags
cmueller:master
cmueller:master-Yvi71-pull
cmueller:main-SSL-support
...
pull from: cmueller:main-SSL-support
Branches Tags
cmueller:master
cmueller:master-Yvi71-pull
cmueller:main-SSL-support

2 commits
master ... main-SSL-s

Author SHA1 Message Date
Administrator ffc090df44 better error handling, added DefaultErrorCallback 2015-11-16 10:45:56 +01:00
Administrator 69d3f52ab4 SSL support (seperated websocket implementation from IOHandler to an proxy object) 2015-11-16 10:44:39 +01:00
7 changed files with 765 additions and 601 deletions
Show stats Expand all files Collapse all files

555
IdHTTPWebsocketClient.pas
View file

@ -13,13 +13,13 @@ uses
{$IFEND}
IdIOHandler,
IdIOHandlerWebsocket,
{$ifdef FMX}
FMX.Types,
{$ELSE}
ExtCtrls,
{$ENDIF}
// {$ifdef FMX}
// FMX.Types,
// {$ELSE}
// ExtCtrls,
// {$ENDIF}
IdWinsock2, Generics.Collections, SyncObjs,
IdSocketIOHandling;
IdSocketIOHandling, IdIOHandlerStack;
type
TWebsocketMsgBin = procedure(const aData: TStream) of object;
@ -39,11 +39,14 @@ type
FOnTextData: TWebsocketMsgText;
FNoAsyncRead: Boolean;
FWriteTimeout: Integer;
function GetIOHandlerWS: TIdIOHandlerWebsocket;
procedure SetIOHandlerWS(const Value: TIdIOHandlerWebsocket);
FUseSSL: boolean;
FWebsocketImpl: TWebsocketImplementationProxy;
function GetIOHandlerWS: TWebsocketImplementationProxy;
procedure SetOnData(const Value: TWebsocketMsgBin);
procedure SetOnTextData(const Value: TWebsocketMsgText);
procedure SetWriteTimeout(const Value: Integer);
function GetIOHandler: TIdIOHandlerStack;
procedure SetIOHandlerStack(const Value: TIdIOHandlerStack);
protected
FSocketIOCompatible: Boolean;
FSocketIOHandshakeResponse: string;
@ -81,7 +84,8 @@ type
procedure Ping;
procedure ReadAndProcessData;
property IOHandler: TIdIOHandlerWebsocket read GetIOHandlerWS write SetIOHandlerWS;
property IOHandler: TIdIOHandlerStack read GetIOHandler write SetIOHandlerStack;
property IOHandlerWS: TWebsocketImplementationProxy read GetIOHandlerWS; // write SetIOHandlerWS;
//websockets
property OnBinData : TWebsocketMsgBin read FOnData write SetOnData;
@ -96,43 +100,11 @@ type
property Host;
property Port;
property WSResourceName: string read FWSResourceName write FWSResourceName;
property UseSSL: boolean read FUseSSL write FUseSSL;
property WriteTimeout: Integer read FWriteTimeout write SetWriteTimeout default 2000;
end;
// on error
(*
TIdHTTPSocketIOClient_old = class(TIdHTTPWebsocketClient)
private
FOnConnected: TNotifyEvent;
FOnDisConnected: TNotifyEvent;
FOnSocketIOMsg: TSocketIOMsg;
FOnSocketIOEvent: TSocketIOMsg;
FOnSocketIOJson: TSocketIOMsg;
protected
FHeartBeat: TTimer;
procedure HeartBeatTimer(Sender: TObject);
procedure InternalUpgradeToWebsocket(aRaiseException: Boolean; out aFailedReason: string);override;
public
procedure AsyncDispatchEvent(const aEvent: string); override;
public
procedure AfterConstruction; override;
destructor Destroy; override;
procedure AutoConnect;
property SocketIOHandshakeResponse: string read FSocketIOHandshakeResponse;
property OnConnected: TNotifyEvent read FOnConnected write FOnConnected;
property OnDisConnected: TNotifyEvent read FOnDisConnected write FOnDisConnected;
// procedure ProcessSocketIORequest(const strmRequest: TStream);
property OnSocketIOMsg : TSocketIOMsg read FOnSocketIOMsg write FOnSocketIOMsg;
property OnSocketIOJson : TSocketIOMsg read FOnSocketIOJson write FOnSocketIOJson;
property OnSocketIOEvent: TSocketIOMsg read FOnSocketIOEvent write FOnSocketIOEvent;
end;
*)
TWSThreadList = class(TThreadList)
public
function Count: Integer;
@ -191,35 +163,6 @@ uses
var
GUnitFinalized: Boolean = false;
//type
// TAnonymousThread = class(TThread)
// protected
// FThreadProc: TThreadProcedure;
// procedure Execute; override;
// public
// constructor Create(AThreadProc: TThreadProcedure);
// end;
//procedure CreateAnonymousThread(AThreadProc: TThreadProcedure);
//begin
// TAnonymousThread.Create(AThreadProc);
//end;
{ TAnonymousThread }
//constructor TAnonymousThread.Create(AThreadProc: TThreadProcedure);
//begin
// FThreadProc := AThreadProc;
// FreeOnTerminate := True;
// inherited Create(False {direct start});
//end;
//
//procedure TAnonymousThread.Execute;
//begin
// if Assigned(FThreadProc) then
// FThreadProc();
//end;
{ TIdHTTPWebsocketClient }
procedure TIdHTTPWebsocketClient.AfterConstruction;
@ -227,9 +170,9 @@ begin
inherited;
FHash := TIdHashSHA1.Create;
IOHandler := TIdIOHandlerWebsocket.Create(nil);
IOHandler.UseNagle := False;
ManagedIOHandler := True;
//IOHandler := TIdIOHandlerWebsocket.Create(nil);
//IOHandler.RealIOHandler.UseNagle := False;
//ManagedIOHandler := True;
FSocketIO := TIdSocketIOHandling_Ext.Create;
// FHeartBeat := TTimer.Create(nil);
@ -326,7 +269,7 @@ begin
else
begin
//clear inputbuffer, otherwise it can't connect :(
if (IOHandler <> nil) then IOHandler.Clear;
if (IOHandlerWS <> nil) then IOHandlerWS.Clear;
inherited Connect;
end;
finally
@ -351,7 +294,8 @@ begin
// tmr.Free;
// end);
TIdWebsocketMultiReadThread.Instance.RemoveClient(Self);
//TIdWebsocketMultiReadThread.Instance.RemoveClient(Self);
DisConnect(True);
FSocketIO.Free;
FHash.Free;
inherited;
@ -360,7 +304,7 @@ end;
procedure TIdHTTPWebsocketClient.DisConnect(ANotifyPeer: Boolean);
begin
if not SocketIOCompatible and
( (IOHandler <> nil) and not IOHandler.IsWebsocket)
( (IOHandlerWS <> nil) and not IOHandlerWS.IsWebsocket)
then
TIdWebsocketMultiReadThread.Instance.RemoveClient(Self);
@ -376,18 +320,19 @@ begin
try
if IOHandler <> nil then
begin
IOHandler.Lock;
IOHandlerWS.Lock;
try
IOHandler.IsWebsocket := False;
IOHandlerWS.IsWebsocket := False;
Self.ManagedIOHandler := False; //otherwise it gets freed while we have a lock on it...
inherited DisConnect(ANotifyPeer);
//clear buffer, other still "connected"
IOHandler.Clear;
IOHandlerWS.Clear;
//IOHandler.Free;
//IOHandler := TIdIOHandlerWebsocket.Create(nil);
finally
IOHandler.Unlock;
IOHandlerWS.Unlock;
end;
end;
finally
@ -395,72 +340,32 @@ begin
end;
end;
function TIdHTTPWebsocketClient.GetIOHandlerWS: TIdIOHandlerWebsocket;
function TIdHTTPWebsocketClient.GetIOHandler: TIdIOHandlerStack;
begin
// if inherited IOHandler is TIdIOHandlerWebsocket then
Result := inherited IOHandler as TIdIOHandlerWebsocket
// else
// Assert(False);
Result := inherited IOHandler as TIdIOHandlerStack;
if Result = nil then
begin
inherited IOHandler := MakeImplicitClientHandler;
Result := inherited IOHandler as TIdIOHandlerStack;
end;
end;
function TIdHTTPWebsocketClient.GetIOHandlerWS: TWebsocketImplementationProxy;
begin
if FWebsocketImpl = nil then
begin
inherited IOHandler := Self.MakeImplicitClientHandler;
Assert(FWebsocketImpl <> nil);
end;
Result := FWebsocketImpl;
end;
function TIdHTTPWebsocketClient.GetSocketIO: TIdSocketIOHandling;
begin
Result := FSocketIO;
end;
(*
procedure TIdHTTPWebsocketClient.HeartBeatTimer(Sender: TObject);
begin
FHeartBeat.Enabled := False;
FSocketIO.Lock;
try
try
if (IOHandler <> nil) and
not IOHandler.ClosedGracefully and
IOHandler.Connected and
(FSocketIOContext <> nil) then
begin
FSocketIO.WritePing(FSocketIOContext as TSocketIOContext); //heartbeat socket.io message
end
//retry re-connect
else
try
//clear inputbuffer, otherwise it can't connect :(
if (IOHandler <> nil) then
IOHandler.Clear;
Self.ConnectTimeout := 100; //100ms otherwise GUI hangs too much -> todo: do it in background thread!
if not Connected then
Self.Connect;
TryUpgradeToWebsocket;
except
//skip, just retried
end;
except on E:Exception do
begin
//clear inputbuffer, otherwise it stays connected :(
if (IOHandler <> nil) then
IOHandler.Clear;
Disconnect(False);
if Assigned(OnDisConnected) then
OnDisConnected(Self);
try
raise EIdException.Create('Connection lost from ' +
Format('ws://%s:%d/%s', [Host, Port, WSResourceName]) +
' - Error: ' + e.Message);
except
//eat, no error popup!
end;
end;
end;
finally
FSocketIO.UnLock;
FHeartBeat.Enabled := True; //always enable: in case of disconnect it will re-connect
end;
end;
*)
function TIdHTTPWebsocketClient.TryConnect: Boolean;
begin
Lock;
@ -493,7 +398,7 @@ begin
FSocketIOConnectBusy := True;
Lock;
try
if (IOHandler <> nil) and IOHandler.IsWebsocket then Exit(True);
if (IOHandler <> nil) and IOHandlerWS.IsWebsocket then Exit(True);
InternalUpgradeToWebsocket(False{no raise}, sError);
Result := (sError = '');
@ -519,7 +424,7 @@ begin
try
if IOHandler = nil then
Connect
else if not IOHandler.IsWebsocket then
else if not IOHandlerWS.IsWebsocket then
InternalUpgradeToWebsocket(True{raise}, sError);
finally
UnLock;
@ -535,7 +440,7 @@ var
sSocketioextended: string;
bLocked: boolean;
begin
Assert((IOHandler = nil) or not IOHandler.IsWebsocket);
Assert((IOHandler = nil) or not IOHandlerWS.IsWebsocket);
//remove from thread during connection handling
TIdWebsocketMultiReadThread.Instance.RemoveClient(Self);
@ -546,10 +451,10 @@ begin
//reset pending data
if IOHandler <> nil then
begin
IOHandler.Lock;
IOHandlerWS.Lock;
bLocked := True;
if IOHandler.IsWebsocket then Exit;
IOHandler.Clear;
if IOHandlerWS.IsWebsocket then Exit;
IOHandlerWS.Clear;
end;
//special socket.io handling, see https://github.com/LearnBoost/socket.io-spec
@ -557,10 +462,17 @@ begin
begin
Request.Clear;
Request.Connection := 'keep-alive';
if UseSSL then
sURL := Format('https://%s:%d/socket.io/1/', [Host, Port])
else
sURL := Format('http://%s:%d/socket.io/1/', [Host, Port]);
strmResponse.Clear;
ReadTimeout := 5 * 1000;
if DebugHook > 0 then
ReadTimeout := ReadTimeout * 10;
//get initial handshake
Post(sURL, strmResponse, strmResponse);
if ResponseCode = 200 {OK} then
@ -625,10 +537,15 @@ begin
//ws://host:port/<resourcename>
//about resourcename, see: http://dev.w3.org/html5/websockets/ "Parsing WebSocket URLs"
//sURL := Format('ws://%s:%d/%s', [Host, Port, WSResourceName]);
if UseSSL then
sURL := Format('https://%s:%d/%s', [Host, Port, WSResourceName])
else
sURL := Format('http://%s:%d/%s', [Host, Port, WSResourceName]);
ReadTimeout := Max(5 * 1000, ReadTimeout);
{ voorbeeld:
{ example:
GET http://localhost:9222/devtools/page/642D7227-148E-47C2-B97A-E00850E3AFA3 HTTP/1.1
Upgrade: websocket
Connection: Upgrade
@ -642,49 +559,8 @@ begin
User-Agent: Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/27.0.1453.116 Safari/537.36
Cookie: __utma=1.2040118404.1366961318.1366961318.1366961318.1; __utmc=1; __utmz=1.1366961318.1.1.utmcsr=(direct)|utmccn=(direct)|utmcmd=(none); deviceorder=0123456789101112; MultiTouchEnabled=false; device=3; network_type=0
}
if SocketIOCompatible then
begin
//1st, try to do socketio specific connection
Response.Clear;
Response.ResponseCode := 0;
Request.URL := sURL;
Request.Method := Id_HTTPMethodGet;
Request.Source := nil;
Response.ContentStream := strmResponse;
PrepareRequest(Request);
//connect and upgrade
ConnectToHost(Request, Response);
//check upgrade succesfull
CheckForGracefulDisconnect(True);
CheckConnected;
Assert(Self.Connected);
if Response.ResponseCode = 0 then
Response.ResponseText := Response.ResponseText
else if Response.ResponseCode <> 200{ok} then
begin
aFailedReason := Format('Error while upgrading: "%d: %s"',[ResponseCode, ResponseText]);
if aRaiseException then
raise EIdWebSocketHandleError.Create(aFailedReason)
else
Exit;
end;
//2nd, get websocket response
Response.Clear;
if IOHandler.CheckForDataOnSource(ReadTimeout) then
begin
Self.FHTTPProto.RetrieveHeaders(MaxHeaderLines);
//Response.RawHeaders.Text := IOHandler.InputBufferAsString();
Response.ResponseText := Response.RawHeaders.Text;
end;
end
else
begin
Get(sURL, strmResponse, [101]);
end;
//http://www.websocket.org/aboutwebsocket.html
(* HTTP/1.1 101 WebSocket Protocol Handshake
@ -698,9 +574,9 @@ begin
Access-Control-Allow-Headers: content-type *)
//'HTTP/1.1 101 Switching Protocols'
if Response.ResponseCode <> 101 then
if ResponseCode <> 101 then
begin
aFailedReason := Format('Error while upgrading: "%d: %s"',[Response.ResponseCode, Response.ResponseText]);
aFailedReason := Format('Error while upgrading: "%d: %s"',[ResponseCode, ResponseText]);
if aRaiseException then
raise EIdWebSocketHandleError.Create(aFailedReason)
else
@ -737,9 +613,10 @@ begin
else
Exit;
end;
end;
//upgrade succesful
IOHandler.IsWebsocket := True;
IOHandlerWS.IsWebsocket := True;
aFailedReason := '';
Assert(Connected);
@ -758,7 +635,7 @@ begin
strmResponse.Free;
if bLocked and (IOHandler <> nil) then
IOHandler.Unlock;
IOHandlerWS.Unlock;
Unlock;
//add to thread for auto retry/reconnect
@ -779,16 +656,27 @@ end;
function TIdHTTPWebsocketClient.MakeImplicitClientHandler: TIdIOHandler;
begin
Result := TIdIOHandlerWebsocket.Create(nil);
if UseSSL then
begin
Result := TIdIOHandlerWebsocketSSL.Create(nil);
FWebsocketImpl := (Result as TIdIOHandlerWebsocketSSL).WebsocketImpl;
end
else
begin
Result := TIdIOHandlerWebsocketPlain.Create(nil);
FWebsocketImpl := (Result as TIdIOHandlerWebsocketPlain).WebsocketImpl;
end;
(Result as TIdIOHandlerStack).UseNagle := False;
end;
procedure TIdHTTPWebsocketClient.Ping;
var
ws: TIdIOHandlerWebsocket;
ws: TWebsocketImplementationProxy;
begin
if TryLock then
try
ws := IOHandler as TIdIOHandlerWebsocket;
ws := IOHandlerWS;
ws.LastPingTime := Now;
//socket.io?
@ -824,10 +712,10 @@ var
wscode: TWSDataCode;
begin
strmEvent := nil;
IOHandler.Lock;
IOHandlerWS.Lock;
try
//try to process all events
while IOHandler.HasData or
while IOHandlerWS.HasData or
(IOHandler.Connected and
IOHandler.Readable(0)) do //has some data
begin
@ -867,7 +755,7 @@ begin
end;
end;
finally
IOHandler.Unlock;
IOHandlerWS.Unlock;
strmEvent.Free;
end;
end;
@ -881,8 +769,8 @@ begin
if IOHandler <> nil then
begin
IOHandler.InputBuffer.Clear;
IOHandler.BusyUpgrading := False;
IOHandler.IsWebsocket := False;
IOHandlerWS.BusyUpgrading := False;
IOHandlerWS.IsWebsocket := False;
//close/disconnect internal socket
//ws := IndyClient.IOHandler as TIdIOHandlerWebsocket;
//ws.Close; done in disconnect below
@ -890,10 +778,9 @@ begin
Disconnect(False);
end;
procedure TIdHTTPWebsocketClient.SetIOHandlerWS(
const Value: TIdIOHandlerWebsocket);
procedure TIdHTTPWebsocketClient.SetIOHandlerStack(const Value: TIdIOHandlerStack);
begin
SetIOHandler(Value);
inherited IOHandler := Value;
end;
procedure TIdHTTPWebsocketClient.SetOnData(const Value: TWebsocketMsgBin);
@ -929,253 +816,6 @@ begin
Self.IOHandler.Binding.SetSockOpt(SOL_SOCKET, SO_SNDTIMEO, Self.WriteTimeout);
end;
{ TIdHTTPSocketIOClient }
(*
procedure TIdHTTPSocketIOClient_old.AfterConstruction;
begin
inherited;
SocketIOCompatible := True;
FHeartBeat := TTimer.Create(nil);
FHeartBeat.Enabled := False;
FHeartBeat.OnTimer := HeartBeatTimer;
end;
procedure TIdHTTPSocketIOClient_old.AsyncDispatchEvent(const aEvent: string);
begin
//https://github.com/LearnBoost/socket.io-spec
if StartsStr('1:', aEvent) then //connect
Exit;
if aEvent = '2::' then //ping, heartbeat
Exit;
inherited AsyncDispatchEvent(aEvent);
end;
procedure TIdHTTPSocketIOClient_old.AutoConnect;
begin
//for now: timer in mainthread?
TThread.Queue(nil,
procedure
begin
FHeartBeat.Interval := 5 * 1000;
FHeartBeat.Enabled := True;
end);
end;
destructor TIdHTTPSocketIOClient_old.Destroy;
var tmr: TObject;
begin
tmr := FHeartBeat;
TThread.Queue(nil, //otherwise free in other thread than created
procedure
begin
//FHeartBeat.Free;
tmr.Free;
end);
inherited;
end;
procedure TIdHTTPSocketIOClient_old.HeartBeatTimer(Sender: TObject);
begin
FHeartBeat.Enabled := False;
try
try
if (IOHandler <> nil) and
not IOHandler.ClosedGracefully and
IOHandler.Connected then
begin
IOHandler.Write('2:::'); //heartbeat socket.io message
end
//retry connect
else
try
//clear inputbuffer, otherwise it can't connect :(
if (IOHandler <> nil) and
not IOHandler.InputBufferIsEmpty
then
IOHandler.DiscardAll;
Self.Connect;
TryUpgradeToWebsocket;
except
//skip, just retried
end;
except
//clear inputbuffer, otherwise it stays connected :(
if (IOHandler <> nil) and
not IOHandler.InputBufferIsEmpty
then
IOHandler.DiscardAll;
if Assigned(OnDisConnected) then
OnDisConnected(Self);
try
raise EIdException.Create('Connection lost from ' + Format('ws://%s:%d/%s', [Host, Port, WSResourceName]));
except
//eat, no error popup!
end;
end;
finally
FHeartBeat.Enabled := True; //always enable: in case of disconnect it will re-connect
end;
end;
procedure TIdHTTPSocketIOClient_old.InternalUpgradeToWebsocket(
aRaiseException: Boolean; out aFailedReason: string);
var
stimeout: string;
begin
inherited InternalUpgradeToWebsocket(aRaiseException, aFailedReason);
if (aFailedReason = '') and
(IOHandler as TIdIOHandlerWebsocket).IsWebsocket then
begin
stimeout := Copy(SocketIOHandshakeResponse, Pos(':', SocketIOHandshakeResponse)+1, Length(SocketIOHandshakeResponse));
stimeout := Copy(stimeout, 1, Pos(':', stimeout)-1);
if stimeout <> '' then
begin
//if (FHeartBeat.Interval > 0) then
//for now: timer in mainthread?
TThread.Queue(nil,
procedure
begin
FHeartBeat.Interval := StrToIntDef(stimeout, 15) * 1000;
if FHeartBeat.Interval >= 15000 then
//FHeartBeat.Interval := FHeartBeat.Interval - 5000
FHeartBeat.Interval := 5000
else if FHeartBeat.Interval >= 5000 then
FHeartBeat.Interval := FHeartBeat.Interval - 2000;
FHeartBeat.Enabled := (FHeartBeat.Interval > 0);
end);
end;
if Assigned(OnConnected) then
OnConnected(Self);
end;
end;
)
procedure TIdHTTPSocketIOClient_old.ProcessSocketIORequest(
const strmRequest: TStream);
function __ReadToEnd: string;
var
utf8: TBytes;
ilength: Integer;
begin
Result := '';
ilength := strmRequest.Size - strmRequest.Position;
SetLength(utf8, ilength);
strmRequest.Read(utf8[0], ilength);
Result := TEncoding.UTF8.GetString(utf8);
end;
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;
begin
str := __ReadToEnd;
if str = '' then Exit;
//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);
if schannel <> '' then
//todo: close channel
else
Self.Disconnect;
end
//(1) Connect
//'1::' [path] [query]
else if StartsStr('1:', str) then
begin
//todo: add channel/room to authorized channel/room list
Self.IOHandler.Write(str); //write same connect back, e.g. 1::/chat
end
//(2) Heartbeat
else if StartsStr('2:', str) then
begin
Self.IOHandler.Write(str); //write same connect back, e.g. 2::
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
OnSocketIOMsg(Self, sdata, imsg);
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
OnSocketIOJson(Self, sdata, imsg);
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(Self, sdata, imsg);
end
//(6) ACK
//6::/news:1+["callback"]
//6:::1+["Response"]
//(7) Error
//(8) Noop
else if StartsStr('8:', str) then
begin
//nothing
end
else
raise Exception.CreateFmt('Unsupported data: "%s"', [str]);
end;
*)
{ TIdWebsocketMultiReadThread }
procedure TIdWebsocketMultiReadThread.AddClient(
@ -1315,7 +955,7 @@ procedure TIdWebsocketMultiReadThread.PingAllChannels;
var
l: TList;
chn: TIdHTTPWebsocketClient;
ws: TIdIOHandlerWebsocket;
ws: TWebsocketImplementationProxy;
i: Integer;
begin
if Terminated then Exit;
@ -1327,10 +967,10 @@ begin
chn := TIdHTTPWebsocketClient(l.Items[i]);
if chn.NoAsyncRead then Continue;
ws := chn.IOHandler as TIdIOHandlerWebsocket;
ws := chn.IOHandlerWS;
//valid?
if (chn.IOHandler <> nil) and
(chn.IOHandler.IsWebsocket) and
(chn.IOHandlerWS.IsWebsocket) and
(chn.Socket <> nil) and
(chn.Socket.Binding <> nil) and
(chn.Socket.Binding.Handle > 0) and
@ -1397,7 +1037,7 @@ begin
end;
//try reconnect
ws := chn.IOHandler as TIdIOHandlerWebsocket;
ws := chn.IOHandlerWS;
if ( (ws = nil) or
(SecondsBetween(Now, ws.LastActivityTime) >= 5) ) then
begin
@ -1439,7 +1079,7 @@ var
iCount,
i: Integer;
iResult: NativeInt;
ws: TIdIOHandlerWebsocket;
ws: TWebsocketImplementationProxy;
begin
l := FChannels.LockList;
try
@ -1455,13 +1095,13 @@ begin
//valid?
if //not chn.Busy and also take busy channels (will be ignored later), otherwise we have to break/reset for each RO function execution
(chn.IOHandler <> nil) and
(chn.IOHandler.IsWebsocket) and
(chn.IOHandlerWS.IsWebsocket) and
(chn.Socket <> nil) and
(chn.Socket.Binding <> nil) and
(chn.Socket.Binding.Handle > 0) and
(chn.Socket.Binding.Handle <> INVALID_SOCKET) then
begin
if chn.IOHandler.HasData then
if chn.IOHandlerWS.HasData then
begin
Inc(iResult);
Break;
@ -1519,13 +1159,12 @@ begin
//check for data for all channels
for i := 0 to l.Count - 1 do
begin
if l = nil then Exit;
chn := TIdHTTPWebsocketClient(l.Items[i]);
if chn.NoAsyncRead then Continue;
if chn.TryLock then
try
ws := chn.IOHandler as TIdIOHandlerWebsocket;
ws := chn.IOHandlerWS;
if (ws = nil) then Continue;
if ws.TryLock then //IOHandler.Readable cannot be done during pending action!

524
IdIOHandlerWebsocket.pas
View file

@ -11,7 +11,7 @@ uses
Classes, SysUtils,
IdIOHandlerStack, IdGlobal, IdException, IdBuffer,
SyncObjs,
Generics.Collections;
Generics.Collections, IdSSLOpenSSL, IdIOHandler, IdSocketHandle;
type
TWSDataType = (wdtText, wdtBinary);
@ -19,14 +19,111 @@ type
TWSExtensionBit = (webBit1, webBit2, webBit3);
TWSExtensionBits = set of TWSExtensionBit;
TIdIOHandlerWebsocket = class;
TWebsocketImplementationProxy = class;
EIdWebSocketHandleError = class(EIdSocketHandleError);
{$if CompilerVersion >= 26} //XE5
TIdTextEncoding = IIdTextEncoding;
{$ifend}
TIdIOHandlerWebsocket = class(TIdIOHandlerStack)
TInheritedFunctions = record
public type
TInherited_Connected = reference to function: Boolean;
TInherited_ReadDataFromSource = reference to function (var VBuffer: TIdBytes): Integer;
TInherited_WriteDataToTarget = reference to function (const ABuffer: TIdBytes; const AOffset, ALength: Integer): Integer;
TInherited_Readable = reference to function (AMSec: Integer = IdTimeoutDefault): Boolean;
TInherited_Write = reference to procedure(const AOut: string; AEncoding: TIdTextEncoding = nil);
TInherited_WriteLn = reference to procedure(const AOut: string; AEncoding: TIdTextEncoding = nil);
TInherited_WriteLnRFC = reference to procedure(const AOut: string = ''; AEncoding: TIdTextEncoding = nil);
TInherited_Write2 = reference to procedure(AValue: TStrings; AWriteLinesCount: Boolean = False; AEncoding: TIdTextEncoding = nil);
TInherited_WriteBufferFlush = reference to procedure(AByteCount: Integer);
TInherited_Write3 = reference to procedure(AStream: TStream; ASize: TIdStreamSize = 0; AWriteByteCount: Boolean = False);
TInherited_ReadBytes = reference to procedure(var VBuffer: TIdBytes; AByteCount: Integer; AAppend: Boolean = True);
public
Inherited_Connected : TInherited_Connected;
Inherited_ReadDataFromSource : TInherited_ReadDataFromSource;
Inherited_WriteDataToTarget : TInherited_WriteDataToTarget;
Inherited_Readable : TInherited_Readable;
Inherited_Write : TInherited_Write;
Inherited_WriteLn : TInherited_WriteLn;
Inherited_WriteLnRFC : TInherited_WriteLnRFC;
Inherited_Write2 : TInherited_Write2;
Inherited_WriteBufferFlush : TInherited_WriteBufferFlush;
Inherited_Write3 : TInherited_Write3;
Inherited_ReadBytes : TInherited_ReadBytes;
end;
IWebsocketFunctions = interface
['{81E49AA4-2B64-41BD-9B3D-24FF757EED80}']
function WebsocketImpl: TWebsocketImplementationProxy;
end;
TIdIOHandlerStack_Ext = class(TIdIOHandlerStack);
TIdIOHandlerWebsocketPlain = class(TIdIOHandlerStack, IWebsocketFunctions) //, IInheritedFunctions)
private
FWebsocketImpl: TWebsocketImplementationProxy;
public
{IWebsocketFunctions}
function WebsocketImpl: TWebsocketImplementationProxy;
protected
function ReadDataFromSource(var VBuffer: TIdBytes): Integer; override;
function WriteDataToTarget (const ABuffer: TIdBytes; const AOffset, ALength: Integer): Integer; override;
public
procedure AfterConstruction;override;
destructor Destroy; override;
function Readable(AMSec: Integer = IdTimeoutDefault): Boolean; override;
function Connected: Boolean; override;
procedure Close; override;
//text/string writes
procedure Write(const AOut: string; AEncoding: TIdTextEncoding = nil); overload; override;
procedure WriteLn(const AOut: string; AEncoding: TIdTextEncoding = nil); overload; override;
procedure WriteLnRFC(const AOut: string = ''; AEncoding: TIdTextEncoding = nil); override;
procedure Write(AValue: TStrings; AWriteLinesCount: Boolean = False; AEncoding: TIdTextEncoding = nil); overload; override;
procedure Write(AStream: TStream; aType: TWSDataType); overload;
procedure WriteBufferFlush(AByteCount: Integer); override;
procedure ReadBytes(var VBuffer: TIdBytes; AByteCount: Integer; AAppend: Boolean = True); override;
end;
TIdSSLIOHandlerSocketOpenSSL_Ext = class(TIdSSLIOHandlerSocketOpenSSL);
TIdIOHandlerWebsocketSSL = class(TIdSSLIOHandlerSocketOpenSSL, IWebsocketFunctions)
private
FWebsocketImpl: TWebsocketImplementationProxy;
public
{IWebsocketFunctions}
function WebsocketImpl: TWebsocketImplementationProxy;
public
function ReadDataFromSource(var VBuffer: TIdBytes): Integer; override;
function WriteDataToTarget (const ABuffer: TIdBytes; const AOffset, ALength: Integer): Integer; override;
public
procedure AfterConstruction;override;
destructor Destroy; override;
procedure ClearSSLOptions;
function Readable(AMSec: Integer = IdTimeoutDefault): Boolean; override;
function Connected: Boolean; override;
procedure Close; override;
//text/string writes
procedure Write(const AOut: string; AEncoding: TIdTextEncoding = nil); overload; override;
procedure WriteLn(const AOut: string; AEncoding: TIdTextEncoding = nil); overload; override;
procedure WriteLnRFC(const AOut: string = ''; AEncoding: TIdTextEncoding = nil); override;
procedure Write(AValue: TStrings; AWriteLinesCount: Boolean = False; AEncoding: TIdTextEncoding = nil); overload; override;
procedure Write(AStream: TStream; aType: TWSDataType); overload;
procedure WriteBufferFlush(AByteCount: Integer); override;
procedure ReadBytes(var VBuffer: TIdBytes; AByteCount: Integer; AAppend: Boolean = True); override;
end;
TWebsocketImplementationProxy = class
private
FRealIOHandler: TIdIOHandlerStack_Ext;
FRealIOHandlerI: TInheritedFunctions;
private
FIsServerSide: Boolean;
FBusyUpgrading: Boolean;
@ -49,8 +146,8 @@ type
FPendingWriteCount: Integer;
function InternalReadDataFromSource(var VBuffer: TIdBytes; ARaiseExceptionOnTimeout: Boolean): Integer;
function ReadDataFromSource(var VBuffer: TIdBytes): Integer; override;
function WriteDataToTarget (const ABuffer: TIdBytes; const AOffset, ALength: Integer): Integer; override;
function ReadDataFromSource(var VBuffer: TIdBytes): Integer;
function WriteDataToTarget (const ABuffer: TIdBytes; const AOffset, ALength: Integer): Integer;
function ReadFrame(out aFIN, aRSV1, aRSV2, aRSV3: boolean; out aDataCode: TWSDataCode; out aData: TIdBytes): Integer;
function ReadMessage(var aBuffer: TIdBytes; out aDataCode: TWSDataCode): Integer;
@ -78,23 +175,23 @@ type
function HasData: Boolean;
procedure Clear;
function Readable(AMSec: Integer = IdTimeoutDefault): Boolean; override;
function Connected: Boolean; override;
function Readable(AMSec: Integer = IdTimeoutDefault): Boolean;
function Connected(): Boolean;
procedure Close; override;
procedure Close;
property Closing : Boolean read FClosing;
property CloseCode : Integer read FCloseCode write FCloseCode;
property CloseReason: string read FCloseReason write FCloseReason;
//text/string writes
procedure Write(const AOut: string; AEncoding: TIdTextEncoding = nil); overload; override;
procedure WriteLn(const AOut: string; AEncoding: TIdTextEncoding = nil); overload; override;
procedure WriteLnRFC(const AOut: string = ''; AEncoding: TIdTextEncoding = nil); override;
procedure Write(AValue: TStrings; AWriteLinesCount: Boolean = False; AEncoding: TIdTextEncoding = nil); overload; override;
procedure Write(const AOut: string; AEncoding: TIdTextEncoding = nil); overload;
procedure WriteLn(const AOut: string; AEncoding: TIdTextEncoding = nil); overload;
procedure WriteLnRFC(const AOut: string = ''; AEncoding: TIdTextEncoding = nil);
procedure Write(AValue: TStrings; AWriteLinesCount: Boolean = False; AEncoding: TIdTextEncoding = nil); overload;
procedure Write(AStream: TStream; aType: TWSDataType); overload;
procedure WriteBufferFlush(AByteCount: Integer); override;
procedure WriteBufferFlush(AByteCount: Integer);
procedure ReadBytes(var VBuffer: TIdBytes; AByteCount: Integer; AAppend: Boolean = True); override;
procedure ReadBytes(var VBuffer: TIdBytes; AByteCount: Integer; AAppend: Boolean = True);
property LastActivityTime: TDateTime read FLastActivityTime write FLastActivityTime;
property LastPingTime: TDateTime read FLastPingTime write FLastPingTime;
@ -231,19 +328,19 @@ end;
{ TIdIOHandlerStack_Websocket }
procedure TIdIOHandlerWebsocket.AfterConstruction;
procedure TWebsocketImplementationProxy.AfterConstruction;
begin
inherited;
inherited AfterConstruction;
FMessageStream := TMemoryStream.Create;
FWSInputBuffer := TIdBuffer.Create;
FLock := TCriticalSection.Create;
FSelectLock := TCriticalSection.Create;
end;
procedure TIdIOHandlerWebsocket.Clear;
procedure TWebsocketImplementationProxy.Clear;
begin
FWSInputBuffer.Clear;
InputBuffer.Clear;
FRealIOHandler.InputBuffer.Clear;
FBusyUpgrading := False;
FIsWebsocket := False;
FClosing := False;
@ -258,7 +355,7 @@ begin
FPendingWriteCount := 0;
end;
procedure TIdIOHandlerWebsocket.Close;
procedure TWebsocketImplementationProxy.Close;
var
iaWriteBuffer: TIdBytes;
sReason: UTF8String;
@ -267,14 +364,14 @@ var
begin
try
//valid connection?
bConnected := Opened and
SourceIsAvailable and
not ClosedGracefully;
bConnected := FRealIOHandler.Opened and
FRealIOHandler.SourceIsAvailable and
not FRealIOHandler.ClosedGracefully;
//no socket error? connection closed by software abort, connection reset by peer, etc
iOptLen := SIZE_INTEGER;
bConnected := bConnected and
(IdWinsock2.getsockopt(Self.Binding.Handle, SOL_SOCKET, SO_ERROR, PAnsiChar(@iOptVal), iOptLen) = 0) and
(IdWinsock2.getsockopt(FRealIOHandler.Binding.Handle, SOL_SOCKET, SO_ERROR, PAnsiChar(@iOptVal), iOptLen) = 0) and
(iOptVal = 0);
if bConnected and IsWebsocket then
@ -315,10 +412,10 @@ begin
if not Closing then
begin
FClosing := True;
CheckForDisconnect();
FRealIOHandler.CheckForDisconnect();
//wait till client respond with close message back
//but a pending message can be in the buffer, so process this too
while ReadFromSource(False{no disconnect error}, 1 * 1000, False) > 0 do ; //response within 1s?
while FRealIOHandler.ReadFromSource(False{no disconnect error}, 1 * 1000, False) > 0 do ; //response within 1s?
end;
end;
except
@ -327,25 +424,25 @@ begin
IsWebsocket := False;
BusyUpgrading := False;
inherited Close;
//inherited Close;
end;
function TIdIOHandlerWebsocket.Connected: Boolean;
function TWebsocketImplementationProxy.Connected(): Boolean;
begin
Lock;
try
Result := inherited Connected;
Result := FRealIOHandlerI.Inherited_Connected();
finally
Unlock;
end;
end;
class constructor TIdIOHandlerWebsocket.Create;
class constructor TWebsocketImplementationProxy.Create;
begin
//UseSingleWriteThread := True;
end;
destructor TIdIOHandlerWebsocket.Destroy;
destructor TWebsocketImplementationProxy.Destroy;
begin
while FPendingWriteCount > 0 do
Sleep(1);
@ -357,29 +454,29 @@ begin
FWSInputBuffer.Free;
FMessageStream.Free;
inherited;
inherited Destroy;
end;
function TIdIOHandlerWebsocket.HasData: Boolean;
function TWebsocketImplementationProxy.HasData: Boolean;
begin
//buffered data available? (more data from previous read)
Result := (FWSInputBuffer.Size > 0) or not InputBufferIsEmpty;
Result := (FWSInputBuffer.Size > 0) or not FRealIOHandler.InputBufferIsEmpty;
end;
function TIdIOHandlerWebsocket.InternalReadDataFromSource(
function TWebsocketImplementationProxy.InternalReadDataFromSource(
var VBuffer: TIdBytes; ARaiseExceptionOnTimeout: Boolean): Integer;
begin
SetLength(VBuffer, 0);
CheckForDisconnect;
if not Readable(ReadTimeout) or
not Opened or
not SourceIsAvailable then
FRealIOHandler.CheckForDisconnect;
if not Readable(FRealIOHandler.ReadTimeout) or
not FRealIOHandler.Opened or
not FRealIOHandler.SourceIsAvailable then
begin
CheckForDisconnect; //disconnected during wait in "Readable()"?
if not Opened then
FRealIOHandler.CheckForDisconnect; //disconnected during wait in "Readable()"?
if not FRealIOHandler.Opened then
EIdNotConnected.Toss(RSNotConnected)
else if not SourceIsAvailable then
else if not FRealIOHandler.SourceIsAvailable then
EIdClosedSocket.Toss(RSStatusDisconnected);
GStack.CheckForSocketError(GStack.WSGetLastError); //check for socket error
if ARaiseExceptionOnTimeout then
@ -388,19 +485,20 @@ begin
Exit(0);
end;
SetLength(VBuffer, RecvBufferSize);
Result := inherited ReadDataFromSource(VBuffer);
if Result = 0 then
SetLength(VBuffer, FRealIOHandler.RecvBufferSize);
Result := FRealIOHandlerI.Inherited_ReadDataFromSource(VBuffer);
if Result <= 0 then
begin
CheckForDisconnect; //disconnected in the mean time?
FRealIOHandler.CheckForDisconnect; //disconnected in the mean time?
GStack.CheckForSocketError(GStack.WSGetLastError); //check for socket error
if ARaiseExceptionOnTimeout then
EIdNoDataToRead.Toss(RSIdNoDataToRead); //nothing read? then connection is probably closed -> exit
end;
end
else
SetLength(VBuffer, Result);
end;
procedure TIdIOHandlerWebsocket.WriteLn(const AOut: string;
procedure TWebsocketImplementationProxy.WriteLn(const AOut: string;
AEncoding: TIdTextEncoding);
begin
if UseSingleWriteThread and IsWebsocket and
@ -419,7 +517,7 @@ begin
Lock;
try
FWriteTextToTarget := True;
inherited WriteLn(AOut, UTF8Encoding); //must be UTF8!
FRealIOHandlerI.Inherited_WriteLn(AOut, UTF8Encoding); //must be UTF8!
finally
FWriteTextToTarget := False;
Unlock;
@ -427,7 +525,7 @@ begin
end;
end;
procedure TIdIOHandlerWebsocket.WriteLnRFC(const AOut: string;
procedure TWebsocketImplementationProxy.WriteLnRFC(const AOut: string;
AEncoding: TIdTextEncoding);
begin
if UseSingleWriteThread and IsWebsocket and
@ -446,7 +544,7 @@ begin
Lock;
try
FWriteTextToTarget := True;
inherited WriteLnRFC(AOut, UTF8Encoding); //must be UTF8!
FRealIOHandlerI.Inherited_WriteLnRFC(AOut, UTF8Encoding); //must be UTF8!
finally
FWriteTextToTarget := False;
Unlock;
@ -454,7 +552,7 @@ begin
end;
end;
procedure TIdIOHandlerWebsocket.Write(const AOut: string;
procedure TWebsocketImplementationProxy.Write(const AOut: string;
AEncoding: TIdTextEncoding);
begin
if UseSingleWriteThread and IsWebsocket and
@ -473,7 +571,7 @@ begin
Lock;
try
FWriteTextToTarget := True;
inherited Write(AOut, UTF8Encoding); //must be UTF8!
FRealIOHandlerI.Inherited_Write(AOut, UTF8Encoding); //must be UTF8!
finally
FWriteTextToTarget := False;
Unlock;
@ -481,7 +579,7 @@ begin
end;
end;
procedure TIdIOHandlerWebsocket.Write(AValue: TStrings;
procedure TWebsocketImplementationProxy.Write(AValue: TStrings;
AWriteLinesCount: Boolean; AEncoding: TIdTextEncoding);
begin
if UseSingleWriteThread and IsWebsocket and
@ -500,7 +598,7 @@ begin
Lock;
try
FWriteTextToTarget := True;
inherited Write(AValue, AWriteLinesCount, UTF8Encoding); //must be UTF8!
FRealIOHandlerI.Inherited_Write2(AValue, AWriteLinesCount, UTF8Encoding); //must be UTF8!
finally
FWriteTextToTarget := False;
Unlock;
@ -508,7 +606,7 @@ begin
end;
end;
procedure TIdIOHandlerWebsocket.Write(AStream: TStream;
procedure TWebsocketImplementationProxy.Write(AStream: TStream;
aType: TWSDataType);
begin
if UseSingleWriteThread and IsWebsocket and
@ -527,7 +625,7 @@ begin
Lock;
try
FWriteTextToTarget := (aType = wdtText);
inherited Write(AStream);
FRealIOHandlerI.Inherited_Write3(AStream);
finally
FWriteTextToTarget := False;
Unlock;
@ -535,9 +633,9 @@ begin
end;
end;
procedure TIdIOHandlerWebsocket.WriteBufferFlush(AByteCount: Integer);
procedure TWebsocketImplementationProxy.WriteBufferFlush(AByteCount: Integer);
begin
if (FWriteBuffer = nil) or (FWriteBuffer.Size <= 0) then Exit;
if (FRealIOHandler.FWriteBuffer = nil) or (FRealIOHandler.FWriteBuffer.Size <= 0) then Exit;
if UseSingleWriteThread and IsWebsocket and
(GetCurrentThreadId <> TIdWebsocketWriteThread.Instance.ThreadID) then
@ -551,10 +649,10 @@ begin
end)
end
else
inherited WriteBufferFlush(AByteCount);
FRealIOHandlerI.Inherited_WriteBufferFlush(AByteCount);
end;
function TIdIOHandlerWebsocket.WriteDataToTarget(const ABuffer: TIdBytes;
function TWebsocketImplementationProxy.WriteDataToTarget(const ABuffer: TIdBytes;
const AOffset, ALength: Integer): Integer;
var
data: TIdBytes;
@ -570,17 +668,17 @@ begin
{$IFDEF DEBUG_WS}
if Debughook > 0 then
OutputDebugString(PChar(Format('Send (non ws, TID:%d, P:%d): %s',
[getcurrentthreadid, Self.Binding.PeerPort, BytesToStringRaw(ABuffer)])));
[getcurrentthreadid, FRealIOHandler.Binding.PeerPort, BytesToStringRaw(ABuffer)])));
{$ENDIF}
Result := inherited WriteDataToTarget(ABuffer, AOffset, ALength)
Result := FRealIOHandlerI.inherited_WriteDataToTarget(ABuffer, AOffset, ALength)
end
else
begin
data := ToBytes(ABuffer, ALength, AOffset);
{$IFDEF DEBUG_WS}
if Debughook > 0 then
OutputDebugString(PChar(Format('Send (ws, TID:%d, P:%d): %s',
[getcurrentthreadid, Self.Binding.PeerPort, BytesToStringRaw(data)])));
OutputDebugString(PChar(Format('Send (ws, IsServerSide: %s, TID:%d, P:%d): %s',
[BoolToStr(IsServerSide), getcurrentthreadid, FRealIOHandler.Binding.PeerPort, BytesToStringRaw(data)])));
{$ENDIF}
try
@ -591,7 +689,7 @@ begin
Result := WriteData(data, wdcBinary, True{send all at once},
webBit1 in ClientExtensionBits, webBit2 in ClientExtensionBits, webBit3 in ClientExtensionBits);
except
FClosedGracefully := True;
FRealIOHandler.FClosedGracefully := True;
Result := -1;
Raise;
end;
@ -601,37 +699,37 @@ begin
end;
end;
function TIdIOHandlerWebsocket.Readable(AMSec: Integer): Boolean;
function TWebsocketImplementationProxy.Readable(AMSec: Integer): Boolean;
begin
if FWSInputBuffer.Size > 0 then Exit(True);
if not FSelectLock.TryEnter then Exit(False);
try
Result := inherited Readable(AMSec);
Result := FRealIOHandlerI.Inherited_Readable(AMSec);
finally
FSelectLock.Leave;
end;
end;
procedure TIdIOHandlerWebsocket.ReadBytes(var VBuffer: TIdBytes;
procedure TWebsocketImplementationProxy.ReadBytes(var VBuffer: TIdBytes;
AByteCount: Integer; AAppend: Boolean);
begin
inherited;
FRealIOHandlerI.inherited_ReadBytes(VBuffer, AByteCount, AAppend);
{$IFDEF DEBUG_WS}
if IsWebsocket then
//if IsWebsocket then
if Debughook > 0 then
begin
OutputDebugString(PChar(Format('%d Bytes read(TID:%d): %s',
[AByteCount, getcurrentthreadid, BytesToStringRaw(VBuffer, AByteCount)])));
OutputDebugString(PChar(Format('Buffer (HeadIndex:%d): %s',
[TIdBuffer_Ext(InputBuffer).FHeadIndex,
BytesToStringRaw(TIdBuffer_Ext(InputBuffer).FBytes,
InputBuffer.Size + TIdBuffer_Ext(InputBuffer).FHeadIndex)])));
[TIdBuffer_Ext(FRealIOHandler.InputBuffer).FHeadIndex,
BytesToStringRaw(TIdBuffer_Ext(FRealIOHandler.InputBuffer).FBytes,
FRealIOHandler.InputBuffer.Size + TIdBuffer_Ext(FRealIOHandler.InputBuffer).FHeadIndex)])));
end;
{$ENDIF}
end;
function TIdIOHandlerWebsocket.ReadDataFromSource(
function TWebsocketImplementationProxy.ReadDataFromSource(
var VBuffer: TIdBytes): Integer;
var
wscode: TWSDataCode;
@ -649,11 +747,11 @@ begin
try
if not IsWebsocket then
begin
Result := inherited ReadDataFromSource(VBuffer);
Result := FRealIOHandlerI.Inherited_ReadDataFromSource(VBuffer);
{$IFDEF DEBUG_WS}
if Debughook > 0 then
OutputDebugString(PChar(Format('Received (non ws, TID:%d, P:%d): %s',
[getcurrentthreadid, Self.Binding.PeerPort, BytesToStringRaw(VBuffer, Result)])));
[getcurrentthreadid, FRealIOHandler.Binding.PeerPort, BytesToStringRaw(VBuffer, Result)])));
{$ENDIF}
end
else
@ -664,28 +762,32 @@ begin
{$IFDEF DEBUG_WS}
if Debughook > 0 then
OutputDebugString(PChar(Format('Received (ws, TID:%d, P:%d): %s',
[getcurrentthreadid, Self.Binding.PeerPort, BytesToStringRaw(VBuffer)])));
OutputDebugString(PChar(Format('Received (ws, IsServerSide: %s, TID:%d, P:%d): %s',
[BoolToStr(IsServerSide), getcurrentthreadid, FRealIOHandler.Binding.PeerPort, BytesToStringRaw(VBuffer)])));
{$ENDIF}
//first write the data code (text or binary, ping, pong)
FInputBuffer.Write(LongWord(Ord(wscode)));
FRealIOHandler.FInputBuffer.Write(LongWord(Ord(wscode)));
//we write message size here, vbuffer is written after this. This way we can use ReadStream to get 1 single message (in case multiple messages in FInputBuffer)
if LargeStream then
FInputBuffer.Write(Int64(Result))
if FRealIOHandler.LargeStream then
FRealIOHandler.FInputBuffer.Write(Int64(Result))
else
FInputBuffer.Write(LongWord(Result))
FRealIOHandler.FInputBuffer.Write(LongWord(Result))
except
FClosedGracefully := True; //closed (but not gracefully?)
on e:Exception do
begin
Unlock; //always unlock when socket exception
FRealIOHandler.FClosedGracefully := True; //closed (but not gracefully?)
Raise;
end;
end;
end;
finally
Unlock; //normal unlock (no double try finally)
end;
end;
function TIdIOHandlerWebsocket.ReadMessage(var aBuffer: TIdBytes; out aDataCode: TWSDataCode): Integer;
function TWebsocketImplementationProxy.ReadMessage(var aBuffer: TIdBytes; out aDataCode: TWSDataCode): Integer;
var
iReadCount: Integer;
iaReadBuffer: TIdBytes;
@ -802,48 +904,48 @@ begin
end;
end;
procedure TIdIOHandlerWebsocket.SetIsWebsocket(const Value: Boolean);
procedure TWebsocketImplementationProxy.SetIsWebsocket(const Value: Boolean);
var data: TIdBytes;
begin
//copy websocket data which was send/received during http upgrade
if not FIsWebsocket and Value and
(FInputBuffer.Size > 0) then
(FRealIOHandler.FInputBuffer.Size > 0) then
begin
FInputBuffer.ExtractToBytes(data);
FRealIOHandler.FInputBuffer.ExtractToBytes(data);
FWSInputBuffer.Write(data);
end;
FIsWebsocket := Value;
end;
procedure TIdIOHandlerWebsocket.Lock;
procedure TWebsocketImplementationProxy.Lock;
begin
FLock.Enter;
end;
function TIdIOHandlerWebsocket.TryLock: Boolean;
function TWebsocketImplementationProxy.TryLock: Boolean;
begin
Result := FLock.TryEnter;
end;
procedure TIdIOHandlerWebsocket.Unlock;
procedure TWebsocketImplementationProxy.Unlock;
begin
FLock.Leave;
end;
{$if CompilerVersion >= 26} //XE5
function TIdIOHandlerWebsocket.UTF8Encoding: IIdTextEncoding;
function TWebsocketImplementationProxy.UTF8Encoding: IIdTextEncoding;
begin
Result := IndyTextEncoding_UTF8;
end;
{$else}
function TIdIOHandlerWebsocket.UTF8Encoding: TEncoding;
function TWebsocketImplementationProxy.UTF8Encoding: TEncoding;
begin
Result := TIdTextEncoding.UTF8;
end;
{$ifend}
function TIdIOHandlerWebsocket.ReadFrame(out aFIN, aRSV1, aRSV2, aRSV3: boolean;
function TWebsocketImplementationProxy.ReadFrame(out aFIN, aRSV1, aRSV2, aRSV3: boolean;
out aDataCode: TWSDataCode; out aData: TIdBytes): Integer;
var
iInputPos: NativeInt;
@ -1015,11 +1117,11 @@ begin
{$IFDEF DEBUG_WS}
if debughook > 0 then
OutputDebugString(PChar(Format('Received (TID:%d, P:%d, Count=%d): %s',
[getcurrentthreadid, Self.Binding.PeerPort, Result, BytesToStringRaw(aData, Result)])));
[getcurrentthreadid, FRealIOHandler.Binding.PeerPort, Result, BytesToStringRaw(aData, Result)])));
{$ENDIF}
end;
function TIdIOHandlerWebsocket.WriteData(aData: TIdBytes;
function TWebsocketImplementationProxy.WriteData(aData: TIdBytes;
aType: TWSDataCode; aFIN, aRSV1, aRSV2, aRSV3: boolean): integer;
var
iByte: Byte;
@ -1035,7 +1137,7 @@ var
bData: TIdBytes;
begin
Result := 0;
Assert(Binding <> nil);
Assert(FRealIOHandler.Binding <> nil);
strmData := TMemoryStream.Create;
Lock;
@ -1137,7 +1239,9 @@ begin
ioffset := 0;
repeat
//Result := Binding.Send(bData, ioffset);
Result := inherited WriteDataToTarget(bdata, iOffset, (Length(bData) - ioffset)); //ssl compatible?
Result := FRealIOHandlerI.inherited_WriteDataToTarget(bdata, iOffset, (Length(bData) - ioffset)); //ssl compatible?
if Result <= 0 then
Break;
Inc(ioffset, Result);
until ioffset >= Length(bData);
@ -1154,7 +1258,7 @@ end;
procedure TIdWebsocketQueueThread.AfterConstruction;
begin
inherited;
inherited AfterConstruction;
FLock := TCriticalSection.Create;
FEvents := TList<TThreadProcedure>.Create;
FProcessing := TList<TThreadProcedure>.Create;
@ -1171,7 +1275,7 @@ begin
UnLock;
FEvents.Free;
FLock.Free;
inherited;
inherited Destroy;
end;
procedure TIdWebsocketQueueThread.Execute;
@ -1291,4 +1395,226 @@ begin
end;
end;
{ TIdIOHandlerWebsocketPlain }
procedure TIdIOHandlerWebsocketPlain.AfterConstruction;
begin
inherited AfterConstruction;
FWebsocketImpl := TWebsocketImplementationProxy.Create;
FWebsocketImpl.FRealIOHandler := TIdIOHandlerStack_Ext(Self as TIdIOHandlerStack);
FWebsocketImpl.FRealIOHandlerI.Inherited_Connected := function: Boolean begin
Result := inherited Connected() end;
FWebsocketImpl.FRealIOHandlerI.Inherited_ReadDataFromSource := function (var VBuffer: TIdBytes): Integer begin
Result := inherited ReadDataFromSource(VBuffer) end;
FWebsocketImpl.FRealIOHandlerI.Inherited_Write := procedure(const AOut: string; AEncoding: TIdTextEncoding = nil) begin
inherited Write(AOut, AEncoding) end;
FWebsocketImpl.FRealIOHandlerI.Inherited_WriteLn := procedure(const AOut: string; AEncoding: TIdTextEncoding = nil) begin
inherited WriteLn(AOut, AEncoding) end;
FWebsocketImpl.FRealIOHandlerI.Inherited_WriteLnRFC := procedure(const AOut: string = ''; AEncoding: TIdTextEncoding = nil) begin
inherited WriteLnRFC(AOut, AEncoding) end;
FWebsocketImpl.FRealIOHandlerI.Inherited_Write2 := procedure(AValue: TStrings; AWriteLinesCount: Boolean = False; AEncoding: TIdTextEncoding = nil) begin
inherited Write(AValue, AWriteLinesCount, AEncoding) end;
FWebsocketImpl.FRealIOHandlerI.Inherited_WriteBufferFlush := procedure(AByteCount: Integer) begin
inherited WriteBufferFlush(AByteCount) end;
FWebsocketImpl.FRealIOHandlerI.Inherited_Write3 := procedure(AStream: TStream; ASize: TIdStreamSize = 0; AWriteByteCount: Boolean = False) begin
inherited Write(AStream, ASize, AWriteByteCount) end;
FWebsocketImpl.FRealIOHandlerI.Inherited_WriteDataToTarget := function (const ABuffer: TIdBytes; const AOffset, ALength: Integer): Integer begin
Result := inherited WriteDataToTarget(ABuffer, AOffset, ALength) end;
FWebsocketImpl.FRealIOHandlerI.Inherited_Readable := function (AMSec: Integer = IdTimeoutDefault): Boolean begin
Result := inherited Readable(AMSec) end;
FWebsocketImpl.FRealIOHandlerI.Inherited_ReadBytes := procedure(var VBuffer: TIdBytes; AByteCount: Integer; AAppend: Boolean = True) begin
inherited ReadBytes(VBuffer, AByteCount, AAppend);
end;
end;
destructor TIdIOHandlerWebsocketPlain.Destroy;
begin
FWebsocketImpl.Free;
inherited Destroy;
end;
procedure TIdIOHandlerWebsocketPlain.Close;
begin
FWebsocketImpl.Close;
inherited Close;
end;
function TIdIOHandlerWebsocketPlain.Connected: Boolean;
begin
Result := FWebsocketImpl.Connected()
end;
function TIdIOHandlerWebsocketPlain.Readable(AMSec: Integer): Boolean;
begin
Result := WebsocketImpl.Readable(AMSec);
end;
procedure TIdIOHandlerWebsocketPlain.ReadBytes(var VBuffer: TIdBytes; AByteCount: Integer; AAppend: Boolean);
begin
WebsocketImpl.ReadBytes(VBuffer, AByteCount, AAppend);
end;
function TIdIOHandlerWebsocketPlain.ReadDataFromSource(var VBuffer: TIdBytes): Integer;
begin
Result := WebsocketImpl.ReadDataFromSource(VBuffer);
end;
function TIdIOHandlerWebsocketPlain.WebsocketImpl: TWebsocketImplementationProxy;
begin
Result := FWebsocketImpl;
end;
procedure TIdIOHandlerWebsocketPlain.Write(AStream: TStream; aType: TWSDataType);
begin
WebsocketImpl.Write(AStream, aType);
end;
procedure TIdIOHandlerWebsocketPlain.Write(const AOut: string; AEncoding: TIdTextEncoding);
begin
WebsocketImpl.Write(AOut, AEncoding);
end;
procedure TIdIOHandlerWebsocketPlain.Write(AValue: TStrings; AWriteLinesCount: Boolean; AEncoding: TIdTextEncoding);
begin
WebsocketImpl.Write(AValue, AWriteLinesCount, AEncoding);
end;
procedure TIdIOHandlerWebsocketPlain.WriteBufferFlush(AByteCount: Integer);
begin
WebsocketImpl.WriteBufferFlush(AByteCount);
end;
function TIdIOHandlerWebsocketPlain.WriteDataToTarget(const ABuffer: TIdBytes; const AOffset, ALength: Integer): Integer;
begin
Result := WebsocketImpl.WriteDataToTarget(ABuffer, AOffset, ALength);
end;
procedure TIdIOHandlerWebsocketPlain.WriteLn(const AOut: string; AEncoding: TIdTextEncoding);
begin
WebsocketImpl.WriteLn(AOut, AEncoding);
end;
procedure TIdIOHandlerWebsocketPlain.WriteLnRFC(const AOut: string; AEncoding: TIdTextEncoding);
begin
WebsocketImpl.WriteLnRFC(AOut, AEncoding);
end;
{ TIdIOHandlerWebsocketSSL }
procedure TIdIOHandlerWebsocketSSL.AfterConstruction;
begin
inherited AfterConstruction;
FWebsocketImpl := TWebsocketImplementationProxy.Create;
FWebsocketImpl.FRealIOHandlerI.Inherited_Connected := function: Boolean begin
Result := inherited Connected() end;
FWebsocketImpl.FRealIOHandlerI.Inherited_ReadDataFromSource := function (var VBuffer: TIdBytes): Integer begin
Result := inherited ReadDataFromSource(VBuffer) end;
FWebsocketImpl.FRealIOHandlerI.Inherited_Write := procedure(const AOut: string; AEncoding: TIdTextEncoding = nil) begin
inherited Write(AOut, AEncoding) end;
FWebsocketImpl.FRealIOHandlerI.Inherited_WriteLn := procedure(const AOut: string; AEncoding: TIdTextEncoding = nil) begin
inherited WriteLn(AOut, AEncoding) end;
FWebsocketImpl.FRealIOHandlerI.Inherited_WriteLnRFC := procedure(const AOut: string = ''; AEncoding: TIdTextEncoding = nil) begin
inherited WriteLnRFC(AOut, AEncoding) end;
FWebsocketImpl.FRealIOHandlerI.Inherited_Write2 := procedure(AValue: TStrings; AWriteLinesCount: Boolean = False; AEncoding: TIdTextEncoding = nil) begin
inherited Write(AValue, AWriteLinesCount, AEncoding) end;
FWebsocketImpl.FRealIOHandlerI.Inherited_WriteBufferFlush := procedure(AByteCount: Integer) begin
inherited WriteBufferFlush(AByteCount) end;
FWebsocketImpl.FRealIOHandlerI.Inherited_Write3 := procedure(AStream: TStream; ASize: TIdStreamSize = 0; AWriteByteCount: Boolean = False) begin
inherited Write(AStream, ASize, AWriteByteCount) end;
FWebsocketImpl.FRealIOHandlerI.Inherited_WriteDataToTarget := function (const ABuffer: TIdBytes; const AOffset, ALength: Integer): Integer begin
if fSSLSocket <> nil then
Result := inherited WriteDataToTarget(ABuffer, AOffset, ALength) end;
FWebsocketImpl.FRealIOHandlerI.Inherited_Readable := function (AMSec: Integer = IdTimeoutDefault): Boolean begin
Result := inherited Readable(AMSec) end;
FWebsocketImpl.FRealIOHandlerI.Inherited_ReadBytes := procedure(var VBuffer: TIdBytes; AByteCount: Integer; AAppend: Boolean = True) begin
inherited ReadBytes(VBuffer, AByteCount, AAppend) end;
FWebsocketImpl.FRealIOHandler := TIdIOHandlerStack_Ext(Self as TIdSSLIOHandlerSocketOpenSSL);
end;
procedure TIdIOHandlerWebsocketSSL.ClearSSLOptions;
begin
if not Self.IsPeer then
with Self do
begin
SSLOptions.Free;
SSLOptions := nil;
end;
end;
procedure TIdIOHandlerWebsocketSSL.Close;
begin
FWebsocketImpl.Close;
inherited Close;
end;
function TIdIOHandlerWebsocketSSL.Connected: Boolean;
begin
Result := WebsocketImpl.Connected;
end;
destructor TIdIOHandlerWebsocketSSL.Destroy;
begin
FWebsocketImpl.Free;
inherited Destroy;
end;
function TIdIOHandlerWebsocketSSL.Readable(AMSec: Integer): Boolean;
begin
Result := WebsocketImpl.Readable(AMSec);
end;
procedure TIdIOHandlerWebsocketSSL.ReadBytes(var VBuffer: TIdBytes; AByteCount: Integer; AAppend: Boolean);
begin
WebsocketImpl.ReadBytes(VBuffer, AByteCount, AAppend);
end;
function TIdIOHandlerWebsocketSSL.ReadDataFromSource(var VBuffer: TIdBytes): Integer;
begin
Result := WebsocketImpl.ReadDataFromSource(VBuffer);
end;
function TIdIOHandlerWebsocketSSL.WebsocketImpl: TWebsocketImplementationProxy;
begin
Result := FWebsocketImpl;
end;
procedure TIdIOHandlerWebsocketSSL.Write(AStream: TStream; aType: TWSDataType);
begin
WebsocketImpl.Write(AStream, aType);
end;
procedure TIdIOHandlerWebsocketSSL.Write(const AOut: string; AEncoding: TIdTextEncoding);
begin
WebsocketImpl.Write(AOut, AEncoding);
end;
procedure TIdIOHandlerWebsocketSSL.Write(AValue: TStrings; AWriteLinesCount: Boolean; AEncoding: TIdTextEncoding);
begin
WebsocketImpl.Write(AValue, AWriteLinesCount, AEncoding);
end;
procedure TIdIOHandlerWebsocketSSL.WriteBufferFlush(AByteCount: Integer);
begin
WebsocketImpl.WriteBufferFlush(AByteCount);
end;
function TIdIOHandlerWebsocketSSL.WriteDataToTarget(const ABuffer: TIdBytes; const AOffset, ALength: Integer): Integer;
begin
Result := WebsocketImpl.WriteDataToTarget(ABuffer, AOffset, ALength);
end;
procedure TIdIOHandlerWebsocketSSL.WriteLn(const AOut: string; AEncoding: TIdTextEncoding);
begin
WebsocketImpl.WriteLn(AOut, AEncoding);
end;
procedure TIdIOHandlerWebsocketSSL.WriteLnRFC(const AOut: string; AEncoding: TIdTextEncoding);
begin
WebsocketImpl.WriteLnRFC(AOut, AEncoding);
end;
end.

108
IdServerIOHandlerWebsocket.pas
View file

@ -5,15 +5,22 @@ interface
uses
Classes,
IdServerIOHandlerStack, IdIOHandlerStack, IdGlobal, IdIOHandler, IdYarn, IdThread, IdSocketHandle,
IdIOHandlerWebsocket;
IdIOHandlerWebsocket, IdSSLOpenSSL, sysutils;
type
TIdServerIOHandlerWebsocket = class(TIdServerIOHandlerStack)
protected
procedure InitComponent; override;
public
function Accept(ASocket: TIdSocketHandle; AListenerThread: TIdThread;
AYarn: TIdYarn): TIdIOHandler; override;
function Accept(ASocket: TIdSocketHandle; AListenerThread: TIdThread; AYarn: TIdYarn): TIdIOHandler; override;
function MakeClientIOHandler(ATheThread:TIdYarn): TIdIOHandler; override;
end;
TIdServerIOHandlerWebsocketSSL = class(TIdServerIOHandlersslOpenSSL)
protected
procedure InitComponent; override;
public
function Accept(ASocket: TIdSocketHandle; AListenerThread: TIdThread; AYarn: TIdYarn): TIdIOHandler; override;
function MakeClientIOHandler(ATheThread:TIdYarn): TIdIOHandler; override;
end;
@ -23,19 +30,47 @@ implementation
function TIdServerIOHandlerWebsocket.Accept(ASocket: TIdSocketHandle;
AListenerThread: TIdThread; AYarn: TIdYarn): TIdIOHandler;
var
LIOHandler: TIdIOHandlerWebsocketPlain;
begin
Result := inherited Accept(ASocket, AListenerThread, AYarn);
//Result := inherited Accept(ASocket, AListenerThread, AYarn);
//using a custom scheduler, AYarn may be nil, so don't assert
Assert(ASocket<>nil);
Assert(AListenerThread<>nil);
Result := nil;
LIOHandler := TIdIOHandlerWebsocketPlain.Create(nil);
try
LIOHandler.Open;
while not AListenerThread.Stopped do
begin
if ASocket.Select(250) then
begin
if LIOHandler.Binding.Accept(ASocket.Handle) then
begin
LIOHandler.AfterAccept;
Result := LIOHandler;
LIOHandler := nil;
Break;
end;
end;
end;
finally
FreeAndNil(LIOHandler);
end;
if Result <> nil then
begin
(Result as TIdIOHandlerWebsocket).IsServerSide := True; //server must not mask, only client
(Result as TIdIOHandlerWebsocket).UseNagle := False;
(Result as TIdIOHandlerWebsocketPlain).WebsocketImpl.IsServerSide := True; //server must not mask, only client
(Result as TIdIOHandlerWebsocketPlain).UseNagle := False;
end;
end;
procedure TIdServerIOHandlerWebsocket.InitComponent;
begin
inherited InitComponent;
IOHandlerSocketClass := TIdIOHandlerWebsocket;
//IOHandlerSocketClass := TIdIOHandlerWebsocket;
end;
function TIdServerIOHandlerWebsocket.MakeClientIOHandler(
@ -44,8 +79,63 @@ begin
Result := inherited MakeClientIOHandler(ATheThread);
if Result <> nil then
begin
(Result as TIdIOHandlerWebsocket).IsServerSide := True; //server must not mask, only client
(Result as TIdIOHandlerWebsocket).UseNagle := False;
(Result as TIdIOHandlerWebsocketPlain).WebsocketImpl.IsServerSide := True; //server must not mask, only client
(Result as TIdIOHandlerWebsocketPlain).UseNagle := False;
end;
end;
{ TIdServerIOHandlerWebsocketSSL }
function TIdServerIOHandlerWebsocketSSL.Accept(ASocket: TIdSocketHandle; AListenerThread: TIdThread;
AYarn: TIdYarn): TIdIOHandler;
var
LIO: TIdIOHandlerWebsocketSSL;
begin
Assert(ASocket<>nil);
Assert(fSSLContext<>nil);
LIO := TIdIOHandlerWebsocketSSL.Create(nil);
try
LIO.PassThrough := True; //initial no ssl?
//we need to pass the SSLOptions for the socket from the server
LIO.SSLOptions := SSLOptions;
LIO.IsPeer := True; //shared SSLOptions + fSSLContext
LIO.Open;
if LIO.Binding.Accept(ASocket.Handle) then
begin
//LIO.ClearSSLOptions;
LIO.SSLSocket := TIdSSLSocket.Create(Self);
LIO.SSLContext := fSSLContext;
end
else
begin
FreeAndNil(LIO);
end;
except
LIO.Free;
raise;
end;
Result := LIO;
if Result <> nil then
begin
(Result as TIdIOHandlerWebsocketSSL).WebsocketImpl.IsServerSide := True; //server must not mask, only client
(Result as TIdIOHandlerWebsocketSSL).UseNagle := False;
end;
end;
procedure TIdServerIOHandlerWebsocketSSL.InitComponent;
begin
inherited InitComponent;
//IOHandlerSocketClass := TIdIOHandlerWebsocket;
end;
function TIdServerIOHandlerWebsocketSSL.MakeClientIOHandler(ATheThread: TIdYarn): TIdIOHandler;
begin
Result := inherited MakeClientIOHandler(ATheThread);
if Result <> nil then
begin
(Result as TIdIOHandlerWebsocketSSL).WebsocketImpl.IsServerSide := True; //server must not mask, only client
(Result as TIdIOHandlerWebsocketSSL).UseNagle := False;
end;
end;

14
IdServerWebsocketContext.pas
View file

@ -5,7 +5,7 @@ interface
uses
Classes,
IdCustomTCPServer, IdIOHandlerWebsocket,
IdServerBaseHandling, IdServerSocketIOHandling, IdContext;
IdServerBaseHandling, IdServerSocketIOHandling, IdContext, IdIOHandlerStack;
type
TIdServerWSContext = class;
@ -29,7 +29,8 @@ type
FSocketIO: TIdServerSocketIOHandling;
FOnDestroy: TIdContextEvent;
public
function IOHandler: TIdIOHandlerWebsocket;
function IOHandler: TIdIOHandlerStack;
function WebsocketImpl: TWebsocketImplementationProxy;
public
function IsSocketIO: Boolean;
property SocketIO: TIdServerSocketIOHandling read FSocketIO write FSocketIO;
@ -67,9 +68,9 @@ begin
inherited;
end;
function TIdServerWSContext.IOHandler: TIdIOHandlerWebsocket;
function TIdServerWSContext.IOHandler: TIdIOHandlerStack;
begin
Result := Self.Connection.IOHandler as TIdIOHandlerWebsocket;
Result := Self.Connection.IOHandler as TIdIOHandlerStack;
end;
function TIdServerWSContext.IsSocketIO: Boolean;
@ -78,4 +79,9 @@ begin
Result := StartsText('/socket.io/1/websocket/', FPath);
end;
function TIdServerWSContext.WebsocketImpl: TWebsocketImplementationProxy;
begin
Result := (IOHandler as IWebsocketFunctions).WebsocketImpl;
end;
end.

39
IdServerWebsocketHandling.pas
View file

@ -16,6 +16,8 @@ type
TIdServerSocketIOHandling_Ext = class(TIdServerSocketIOHandling)
end;
TIdCustomHTTPServer_Ext = class(TIdCustomHTTPServer);
TIdServerWebsocketHandling = class(TIdServerBaseHandling)
protected
class procedure DoWSExecute(AThread: TIdContext; aSocketIOHandler: TIdServerSocketIOHandling_Ext);virtual;
@ -61,10 +63,10 @@ begin
try
context := AThread as TIdServerWSContext;
//todo: make seperate function + do it after first real write (not header!)
if context.IOHandler.BusyUpgrading then
if context.WebsocketImpl.BusyUpgrading then
begin
context.IOHandler.IsWebsocket := True;
context.IOHandler.BusyUpgrading := False;
context.WebsocketImpl.IsWebsocket := True;
context.WebsocketImpl.BusyUpgrading := False;
end;
//initial connect
if context.IsSocketIO then
@ -79,10 +81,13 @@ begin
while AThread.Connection.Connected do
begin
if context.IOHandler.HasData or
if context.WebsocketImpl.HasData or
(AThread.Connection.IOHandler.InputBuffer.Size > 0) or
AThread.Connection.IOHandler.Readable(1 * 1000) then //wait 5s, else ping the client(!)
begin
if not (context.WebsocketImpl.HasData or (context.IOHandler.InputBuffer.Size > 0)) then
Continue;
tstart := Now;
strmResponse := TMemoryStream.Create;
@ -99,7 +104,7 @@ begin
if wscode in [wdcPing, wdcPong] then
begin
if wscode = wdcPing then
context.IOHandler.WriteData(nil, wdcPong);
context.WebsocketImpl.WriteData(nil, wdcPong);
Continue;
end;
@ -108,18 +113,23 @@ begin
else
wstype := wdtBinary;
try
HandleWSMessage(context, wstype, strmRequest, strmResponse, aSocketIOHandler);
except
on E:Exception do
TIdCustomHTTPServer_Ext(context.Server as TIdCustomHTTPServer).DoCommandError(context, nil, nil, E);
end;
//write result back (of the same type: text or bin)
if strmResponse.Size > 0 then
begin
if wscode = wdcText then
context.IOHandler.Write(strmResponse, wdtText)
context.WebsocketImpl.Write(strmResponse, wdtText)
else
context.IOHandler.Write(strmResponse, wdtBinary)
context.WebsocketImpl.Write(strmResponse, wdtBinary)
end
else
context.IOHandler.WriteData(nil, wdcPing);
context.WebsocketImpl.WriteData(nil, wdcPing);
finally
strmRequest.Free;
strmResponse.Free;
@ -137,7 +147,7 @@ begin
aSocketIOHandler.WritePing(context);
end
else
context.IOHandler.WriteData(nil, wdcPing);
context.WebsocketImpl.WriteData(nil, wdcPing);
end;
end;
@ -147,7 +157,7 @@ begin
Assert(aSocketIOHandler <> nil);
aSocketIOHandler.WriteDisConnect(context);
end;
context.IOHandler.Clear;
context.WebsocketImpl.Clear;
AThread.Data := nil;
end;
end;
@ -194,7 +204,8 @@ begin
Sec-WebSocket-Version: 13 *)
//Connection: Upgrade
if not ContainsText(ARequestInfo.Connection, 'Upgrade') then //Firefox uses "keep-alive, Upgrade"
if not ( ContainsText(ARequestInfo.Connection, 'Upgrade') or //Firefox uses "keep-alive, Upgrade"
ContainsText(ARequestInfo.RawHeaders.Values['upgrade'], 'websocket') ) then //"connection" is empty in case of SSL? so check header too
begin
//initiele ondersteuning voor socket.io
if SameText(ARequestInfo.document , '/socket.io/1/') then
@ -325,17 +336,21 @@ begin
hash.Free;
end;
AResponseInfo.CustomHeaders.Values['Sec-WebSocket-Accept'] := sValue;
AResponseInfo.CustomHeaders.Values['Keep-alive'] := 'true';
//send same protocol back?
AResponseInfo.CustomHeaders.Values['Sec-WebSocket-Protocol'] := context.WebSocketProtocol;
//we do not support extensions yet (gzip deflate compression etc)
//AResponseInfo.CustomHeaders.Values['Sec-WebSocket-Extensions'] := context.WebSocketExtensions;
//http://www.lenholgate.com/blog/2011/07/websockets---the-deflate-stream-extension-is-broken-and-badly-designed.html
//but is could be done using idZlib.pas and DecompressGZipStream etc
AResponseInfo.CustomHeaders.Values['sec-websocket-extensions'] := '';
context.WebSocketExtensions := '';
//send response back
context.IOHandler.InputBuffer.Clear;
context.IOHandler.BusyUpgrading := True;
context.WebsocketImpl.BusyUpgrading := True;
AResponseInfo.WriteHeader;
//handle all WS communication in seperate loop

48
IdSocketIOHandling.pas
View file

@ -6,7 +6,7 @@ uses
Classes, Generics.Collections,
superobject,
IdServerBaseHandling, IdContext, IdException, IdIOHandlerWebsocket, IdHTTP,
SyncObjs, SysUtils;
SyncObjs, SysUtils, IdIOHandlerStack;
type
TSocketIOContext = class;
@ -74,7 +74,7 @@ type
protected
FHandling: TIdBaseSocketIOHandling;
FContext: TIdContext;
FIOHandler: TIdIOHandlerWebsocket;
FIOHandler: TIdIOHandlerStack;
FClient: TIdHTTP;
FEvent: TEvent;
FQueue: TList<string>;
@ -163,6 +163,7 @@ type
procedure ProcessEvent(const AContext: ISocketIOContext; const aText: string; aMsgNr: Integer; aHasCallback: Boolean);
private
FOnEventError: TSocketIOEventError;
FDefaultErrorCallback: TSocketIOError;
protected
type
TSocketIOCallback = procedure(const aData: string) of object;
@ -221,6 +222,7 @@ type
procedure OnConnection(const aCallback: TSocketIONotify);
procedure OnDisconnect(const aCallback: TSocketIONotify);
property OnEventError: TSocketIOEventError read FOnEventError write FOnEventError;
property DefaultErrorCallback: TSocketIOError read FDefaultErrorCallback write FDefaultErrorCallback;
procedure EnumerateSockets(const aEachSocketCallback: TSocketIONotify);
end;
@ -547,7 +549,7 @@ begin
if FOnEventList.TryGetValue(name, list) then
begin
if list.Count = 0 then
raise EIdSocketIoUnhandledMessage.Create(aText);
raise EIdSocketIoUnhandledMessage.Create('No listener available for event: ' + aText);
// socket := FConnections.Items[AContext];
if aHasCallback then
@ -560,7 +562,7 @@ begin
event(AContext, args, callback);
except on E:Exception do
if Assigned(OnEventError) then
OnEventError(AContext, callback, e)
OnEventError(AContext, callback, e) //custom error handling (send different error back to remote client/context/initiator?)
else
if callback <> nil then
callback.SendResponse( SO(['Error', SO(['msg', e.message])]).AsJSon );
@ -570,7 +572,7 @@ begin
end;
end
else
raise EIdSocketIoUnhandledMessage.Create(aText);
raise EIdSocketIoUnhandledMessage.Create('No listeners registered for event: ' + aText);
finally
// args.Free;
json := nil;
@ -742,6 +744,7 @@ procedure TIdBaseSocketIOHandling.ProcessSocketIORequest(
var
str, smsg, schannel, sdata: string;
sErrorType, sErrorMsg: string;
imsg: Integer;
bCallback: Boolean;
// socket: TSocketIOContext;
@ -878,7 +881,7 @@ begin
ProcessEvent(socket, sdata, imsg, bCallback);
except
on e:exception do
//
raise
end
end
//(6) ACK
@ -890,10 +893,7 @@ begin
imsg := StrToIntDef(smsg, 0);
sData := Copy(sdata, Pos('+', sData)+1, Length(sData));
TSocketIOContext(ASocket).FPendingMessages.Remove(imsg);
if FSocketIOErrorRef.TryGetValue(imsg, errorref) then
begin
FSocketIOErrorRef.Remove(imsg);
error := nil;
//'[{"Error":{"Message":"Operation aborted","Type":"EAbort"}}]'
if ContainsText(sdata, '{"Error":') then
begin
@ -901,16 +901,40 @@ begin
if error.IsType(stArray) then
error := error.O['0'];
error := error.O['Error'];
sErrorType := error.S['Type'];
if error.S['Message'] <> '' then
errorref(ASocket, error.S['Type'], error.S['Message'])
sErrorMsg := error.S['Message']
else if error.S['msg'] <> '' then
sErrorMsg := error.S['msg']
else
errorref(ASocket, 'Unknown', sdata);
begin
sErrorMsg := sdata;
sErrorType := 'Unknown';
end;
end;
TSocketIOContext(ASocket).FPendingMessages.Remove(imsg);
if FSocketIOErrorRef.TryGetValue(imsg, errorref) then
begin
FSocketIOErrorRef.Remove(imsg);
if error <> nil then
begin
errorref(ASocket, sErrorType, sErrorMsg);
FSocketIOEventCallback.Remove(imsg);
FSocketIOEventCallbackRef.Remove(imsg);
Exit;
end;
end;
//no error handler? than always raise an exception so programmer gets notified (and can log it using exception logger)
if error <> nil then
begin
if Assigned(DefaultErrorCallback) then
DefaultErrorCallback(ASocket, sErrorType, sErrorMsg)
else
raise ESocketIOException.CreateFmt('Server side error "%s": %s', [sErrorType, sErrorMsg]);
end;
if FSocketIOEventCallback.TryGetValue(imsg, callback) then
begin

78
IdWebsocketServer.pas
View file

@ -4,7 +4,7 @@ interface
uses
IdServerWebsocketHandling, IdServerSocketIOHandling, IdServerWebsocketContext,
IdHTTPServer, IdContext, IdCustomHTTPServer, Classes, IdIOHandlerWebsocket;
IdHTTPServer, IdContext, IdCustomHTTPServer, Classes, IdIOHandlerWebsocket, IdGlobal, IdServerIOHandler;
type
TWebsocketMessageText = procedure(const AContext: TIdServerWSContext; const aText: string) of object;
@ -16,9 +16,14 @@ type
FOnMessageText: TWebsocketMessageText;
FOnMessageBin: TWebsocketMessageBin;
FWriteTimeout: Integer;
FUseSSL: boolean;
function GetSocketIO: TIdServerSocketIOHandling;
procedure SetWriteTimeout(const Value: Integer);
function GetIOHandler: TIdServerIOHandler;
protected
procedure Startup; override;
procedure DetermineSSLforPort(APort: TIdPort; var VUseSSL: Boolean);
procedure DoCommandGet(AContext: TIdContext; ARequestInfo: TIdHTTPRequestInfo;
AResponseInfo: TIdHTTPResponseInfo); override;
procedure ContextCreated(AContext: TIdContext); override;
@ -35,15 +40,17 @@ type
property OnMessageText: TWebsocketMessageText read FOnMessageText write FOnMessageText;
property OnMessageBin : TWebsocketMessageBin read FOnMessageBin write FOnMessageBin;
property IOHandler: TIdServerIOHandler read GetIOHandler write SetIOHandler;
property SocketIO: TIdServerSocketIOHandling read GetSocketIO;
published
property WriteTimeout: Integer read FWriteTimeout write SetWriteTimeout default 2000;
property UseSSL: boolean read FUseSSL write FUseSSL;
end;
implementation
uses
IdServerIOHandlerWebsocket, IdStreamVCL, IdGlobal, Windows, IdWinsock2;
IdServerIOHandlerWebsocket, IdStreamVCL, Windows, IdWinsock2, IdSSLOpenSSL, IdSSL, IdThread; //, idIOHandler, idssl;
{ TIdWebsocketServer }
@ -54,9 +61,6 @@ begin
FSocketIO := TIdServerSocketIOHandling_Ext.Create;
ContextClass := TIdServerWSContext;
if IOHandler = nil then
IOHandler := TIdServerIOHandlerWebsocket.Create(Self);
FWriteTimeout := 2 * 1000; //2s
end;
@ -82,6 +86,28 @@ begin
FSocketIO.Free;
end;
procedure TIdWebsocketServer.DetermineSSLforPort(APort: TIdPort; var VUseSSL: Boolean);
//var
// thread: TIdThreadWithTask;
// ctx: TIdServerWSContext;
begin
VUseSSL := IOHandler.InheritsFrom(TIdServerIOHandlerSSLBase);
{$message warn 'todo: no ssl for localhost (testing, server IPC, etc)?'}
(*
//
if TThread.CurrentThread is TIdThreadWithTask then
begin
thread := TThread.CurrentThread as TIdThreadWithTask;
ctx := thread.Task as TIdServerWSContext;
//yarn := thread.Task.Yarn as TIdYarnOfThread;
if ctx.Binding.PeerIP = '127.0.0.1' then
VUseSSL := false;
end;
*)
end;
procedure TIdWebsocketServer.DoCommandGet(AContext: TIdContext;
ARequestInfo: TIdHTTPRequestInfo; AResponseInfo: TIdHTTPResponseInfo);
begin
@ -92,6 +118,39 @@ begin
inherited DoCommandGet(AContext, ARequestInfo, AResponseInfo);
end;
function TIdWebsocketServer.GetIOHandler: TIdServerIOHandler;
begin
Result := inherited IOHandler;
if Result = nil then
begin
if UseSSL then
begin
Result := TIdServerIOHandlerWebsocketSSL.Create(Self);
with Result as TIdServerIOHandlerWebsocketSSL do
begin
//note: custom certificate files must be set by user, e.g. in datamodule OnCreate:
//FHttpServer := TIdWebsocketServer.Create;
//FHttpServer.UseSSL := True;
//with FHttpServer.IOHandler as TIdServerIOHandlerWebsocketSSL do
// SSLOptions.RootCertFile := 'root.cer';
// SSLOptions.CertFile := 'your_cert.cer';
// SSLOptions.KeyFile := 'key.pem';
SSLOptions.Method := sslvSSLv23;
SSLOptions.Mode := sslmServer;
OnQuerySSLPort := DetermineSSLforPort;
end;
end
else
Result := TIdServerIOHandlerWebsocket.Create(Self);
inherited IOHandler := Result;
end;
end;
function TIdWebsocketServer.GetSocketIO: TIdServerSocketIOHandling;
begin
Result := FSocketIO;
@ -109,7 +168,7 @@ begin
begin
ctx := TIdServerWSContext(l.Items[i]);
Assert(ctx is TIdServerWSContext);
if ctx.IOHandler.IsWebsocket and
if ctx.WebsocketImpl.IsWebsocket and
not ctx.IsSocketIO
then
ctx.IOHandler.Write(aText);
@ -124,6 +183,11 @@ begin
FWriteTimeout := Value;
end;
procedure TIdWebsocketServer.Startup;
begin
inherited;
end;
procedure TIdWebsocketServer.WebsocketChannelRequest(
const AContext: TIdServerWSContext; aType: TWSDataType; const aStrmRequest,
aStrmResponse: TMemoryStream);
@ -158,7 +222,7 @@ begin
begin
ctx := TIdServerWSContext(l.Items[i]);
Assert(ctx is TIdServerWSContext);
if ctx.IOHandler.IsWebsocket and
if ctx.WebsocketImpl.IsWebsocket and
not ctx.IsSocketIO
then
ctx.IOHandler.Write(bytes);