paweld/lazarus-ccr
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

TCP protocole now supports block type prefixed messages.

Browse Source 
git-svn-id: https://svn.code.sf.net/p/lazarus-ccr/svn@1783 8e941d3f-bd1b-0410-a28a-d453659cc2b4
This commit is contained in:
inoussa 2011-08-08 02:24:18 +00:00
parent d1ae85de1d
commit 0a24ccf684
4 changed files with 124 additions and 44 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
wst/trunk/client_utils.pas
View File

@ -54,6 +54,7 @@ Type
FContentType : string;
FFormat : string;
FTarget : string;
FUseBlockType : Boolean;
protected
procedure DoSend(const AData; const ALength : Int64); virtual; abstract;
function DoReceive(var AData; const ALength : Int64) : Int64; virtual; abstract;
@ -63,6 +64,7 @@ Type
property Target : string Read FTarget Write FTarget;
property ContentType : string Read FContentType Write FContentType;
property Format : string read FFormat write FFormat;
property UseBlockType : Boolean read FUseBlockType write FUseBlockType default false;
end;
{$M+}
@ -113,6 +115,8 @@ begin
buffStream := TMemoryStream.Create();
Try
wrtr := CreateBinaryWriter(buffStream);
if UseBlockType then
wrtr.WriteInt32S(WST_BLOCK_TYPE);
wrtr.WriteInt32S(0);
wrtr.WriteAnsiStr(Target);
wrtr.WriteAnsiStr(ContentType);
@ -134,8 +138,13 @@ begin
end;
wrtr.WriteBinary(binBuff);
SetLength(binBuff,0);
buffStream.Position := 0;
wrtr.WriteInt32S(buffStream.Size-4);
if UseBlockType then begin
buffStream.Position := 4;
wrtr.WriteInt32S(buffStream.Size-({BlockType}4+4));
end else begin
buffStream.Position := 0;
wrtr.WriteInt32S(buffStream.Size-4);
end;
buffStream.Position := 0;
DoSend(buffStream.Memory^,buffStream.Size);

98
wst/trunk/fpc_tcp_server.pas
View File

@ -19,8 +19,7 @@ uses
Classes, SysUtils, ssockets, server_listener, wst_types;
const
sSERVER_PORT = 1234;
sSERVER_PORT = 1234;
type
TwstFPCTcpListener = class;
@ -33,7 +32,7 @@ type
FSocket : TSocketStream;
FOwner : TwstFPCTcpListener;
private
function ReadRequest(ARequest : TStream):Integer;
function ReadRequest(ARequest : TStream; var ABlockType : LongInt):Integer;
procedure SendResponse(AResponse : TMemoryStream);
public
constructor Create (ASocket : TSocketStream; AOwner : TwstFPCTcpListener);
@ -58,11 +57,11 @@ type
procedure SuspendAsSoonAsPossible();
procedure ResumeListening();
property DefaultTimeOut : Integer read FDefaultTimeOut write FDefaultTimeOut;
end;
end;
{ TwstFPCTcpListener }
TwstFPCTcpListener = class(TwstListener)
TwstFPCTcpListener = class(TwstBaseTcpListener)
private
FServerThread : TServerListnerThread;
FPort : Integer;
@ -76,27 +75,38 @@ type
implementation
uses binary_streamer, server_service_intf, server_service_imputils, math;
resourcestring
SErrReadingFromSocket = 'Error %d reading data from socket';
SErrWritingToSocket = 'Error %d writing data to socket';
uses
wst_consts, binary_streamer, server_service_intf, server_service_imputils, math;
{ TClientHandlerThread }
function TClientHandlerThread.ReadRequest(ARequest : TStream): Integer;
function TClientHandlerThread.ReadRequest(
ARequest : TStream;
var ABlockType : LongInt
): Integer;
var
binBuff : TByteDynArray;
bufferLen : LongInt;
i, j, c, readBufferLen : PtrInt;
bufferLen, bktype : TInt32S;
i, j, c : PtrInt;
begin
Result := 0;
if (tloHandleBlockType in FOwner.Options) then begin
bktype := 0;
j:=FSocket.Read(bktype,SizeOf(bktype));
if (j<0) then
raise Exception.CreateFmt(SERR_ErrorReadindDataToSocket,[FSocket.LastError]);
if (j=0) then
Exit(0) // Closed gracefully
else
bktype:=Reverse_32(bktype);
end;
bufferLen := 0;
j:=FSocket.Read(bufferLen,SizeOf(bufferLen));
if (j<0) then
Raise Exception.CreateFmt(SErrReadingFromSocket,[FSocket.LastError]);
Raise Exception.CreateFmt(SERR_ErrorReadindDataToSocket,[FSocket.LastError]);
if (j=0) then
Exit(0) // Closed gracefully
else
@ -110,7 +120,7 @@ begin
repeat
j:=FSocket.Read(binBuff[0],i);
If (J<=0) then
Raise Exception.CreateFmt(SErrReadingFromSocket,[FSocket.LastError]);
Raise Exception.CreateFmt(SERR_ErrorReadindDataToSocket,[FSocket.LastError]);
ARequest.Write(binBuff[0],j);
Inc(c,j);
I:=Min(1024,(bufferLen - c ))
@ -120,6 +130,9 @@ begin
if C<ARequest.Size then
ARequest.Size:=C;
end;
if (tloHandleBlockType in FOwner.Options) then
ABlockType := bktype;
end;
procedure TClientHandlerThread.SendResponse(AResponse : TMemoryStream);
@ -134,7 +147,7 @@ begin
Repeat
W:=FSocket.Write(P^,C);
if (W<0) then
Raise Exception.CreateFmt(SErrWritingToSocket,[FSocket.LastError]);
Raise Exception.CreateFmt(SERR_ErrorSendindDataToSocket,[FSocket.LastError]);
Inc(P,W);
Dec(C,W);
Until (C=0) or (w=0);
@ -174,26 +187,33 @@ var
rqst : IRequestBuffer;
ARequest,AResponse : TMemoryStream;
i : PtrUInt;
blocktype : TInt32S;
begin
while not Terminated do
while not Terminated do begin
Try
ARequest:=TMemoryStream.Create;
blocktype := 0;
AResponse := nil;
ARequest := TMemoryStream.Create;
try
if ReadRequest(ARequest)>SizeOf(LongInt) then
begin
rdr := CreateBinaryReader(ARequest);
trgt := rdr.ReadAnsiStr();
ctntyp := rdr.ReadAnsiStr();
frmt := rdr.ReadAnsiStr();
buff := rdr.ReadBinary();
rdr := nil;
ARequest.Size := 0;
ARequest.Write(buff[0],Length(buff));
SetLength(buff,0);
ARequest.Position := 0;
AResponse:=TMemoryStream.Create;
try
if ReadRequest(ARequest,blocktype)>SizeOf(LongInt) then begin
AResponse := TMemoryStream.Create();
if (tloHandleBlockType in FOwner.Options) and
(blocktype <> WST_BLOCK_TYPE)
then begin
if (FOwner.UnknownBlockHandler <> nil) then
FOwner.UnknownBlockHandler.Execute(blocktype,ARequest,AResponse);
end else begin
rdr := CreateBinaryReader(ARequest);
trgt := rdr.ReadAnsiStr();
ctntyp := rdr.ReadAnsiStr();
frmt := rdr.ReadAnsiStr();
buff := rdr.ReadBinary();
rdr := nil;
ARequest.Size := 0;
ARequest.Write(buff[0],Length(buff));
SetLength(buff,0);
ARequest.Position := 0;
rqst := TRequestBuffer.Create(trgt,ctntyp,ARequest,AResponse,frmt);
//rqst.GetPropertyManager().SetProperty(sREMOTE_IP,FSocketObject.GetRemoteSinIP());
//rqst.GetPropertyManager().SetProperty(sREMOTE_PORT,IntToStr(FSocketObject.GetRemoteSinPort()));
@ -206,21 +226,21 @@ begin
wrtr := CreateBinaryWriter(AResponse);
wrtr.WriteBinary(buff);
SetLength(buff,0);
end;
if (AResponse.Size > 0) then
SendResponse(AResponse);
finally
AResponse.Free;
end;
end;
end;
finally
AResponse.Free;
ARequest.Free;
end;
except
on e : Exception do
begin
on e : Exception do begin
Terminate;
FOwner.NotifyMessage(Format('Error : ThreadID = %d; Message = %s',[Self.ThreadID,e.Message]));
end;
end;
end;
end;
end;
{ TServerListnerThread }

54
wst/trunk/server_listener.pas
View File

@ -23,9 +23,20 @@ const
sSERVICES_PREFIXE = 'services';
sWSDL = 'WSDL';
type
type
TTcpListenerOption = (tloHandleBlockType);
TTcpListenerOptions = set of TTcpListenerOption;
TListnerNotifyMessage = procedure(Sender : TObject; const AMsg : string) of object;
IBlockHandler = interface
['{E0C50F08-A2C3-41D7-ACD5-E7867DD9F981}']
procedure Execute(
const ABlockType : LongInt;
ARequestBlock,
AResponseBlock : TStream
);
end;
{ TwstListener }
@ -43,10 +54,25 @@ type
property OnNotifyMessage : TListnerNotifyMessage read FOnNotifyMessage write SetOnNotifyMessage;
end;
{ TwstBaseTcpListener }
TwstBaseTcpListener = class(TwstListener)
private
FOptions : TTcpListenerOptions;
FUnknownBlockHandler : IBlockHandler;
protected
procedure CheckActive(const AActive : Boolean; ACaller : string);
procedure SetOptions(const AValue : TTcpListenerOptions);
procedure SetUnknownBlockHandler(const AValue : IBlockHandler);
public
property Options : TTcpListenerOptions read FOptions write SetOptions;
property UnknownBlockHandler : IBlockHandler read FUnknownBlockHandler write SetUnknownBlockHandler;
end;
function GenerateWSDLHtmlTable(const AServicesModulePath : string=''): string;
implementation
uses base_service_intf, metadata_repository,
uses wst_consts, base_service_intf, metadata_repository,
metadata_service, metadata_service_binder, metadata_service_imp ;
@ -89,6 +115,30 @@ begin
'</html>';
end;
{ TwstBaseTcpListener }
procedure TwstBaseTcpListener.CheckActive(const AActive : Boolean; ACaller : string);
begin
if (IsActive() <> AActive) then
raise Exception.CreateFmt(SERR_ObjectStateDoesNotAllowOperation,[ACaller]);
end;
procedure TwstBaseTcpListener.SetOptions(const AValue : TTcpListenerOptions);
begin
CheckActive(False,'SetOptions');
if (FOptions=AValue) then
exit;
FOptions:=AValue;
end;
procedure TwstBaseTcpListener.SetUnknownBlockHandler(const AValue : IBlockHandler);
begin
CheckActive(False,'SetUnknownBlockHandler');
if (FUnknownBlockHandler = AValue) then
exit;
FUnknownBlockHandler := AValue;
end;
{ TwstListener }
procedure TwstListener.SetOnNotifyMessage(const AValue : TListnerNotifyMessage);

3
wst/trunk/wst_consts.pas
View File

@ -16,7 +16,8 @@ unit wst_consts;
interface
const
const
WST_BLOCK_TYPE = LongInt(56789);
sWST_SIGNATURE = 'WST_METADATA_0.6';
resourcestring