| (* |
| * Licensed to the Apache Software Foundation (ASF) under one |
| * or more contributor license agreements. See the NOTICE file |
| * distributed with this work for additional information |
| * regarding copyright ownership. The ASF licenses this file |
| * to you under the Apache License, Version 2.0 (the |
| * "License"); you may not use this file except in compliance |
| * with the License. You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, |
| * software distributed under the License is distributed on an |
| * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| * KIND, either express or implied. See the License for the |
| * specific language governing permissions and limitations |
| * under the License. |
| *) |
| unit Thrift.Transport.Pipes; |
| |
| {$WARN SYMBOL_PLATFORM OFF} |
| {$I Thrift.Defines.inc} |
| |
| interface |
| |
| uses |
| {$IFDEF OLD_UNIT_NAMES} |
| Windows, SysUtils, Math, AccCtrl, AclAPI, SyncObjs, |
| {$ELSE} |
| Winapi.Windows, System.SysUtils, System.Math, Winapi.AccCtrl, Winapi.AclAPI, System.SyncObjs, |
| {$ENDIF} |
| Thrift.Configuration, |
| Thrift.Transport, |
| Thrift.Utils, |
| Thrift.Stream; |
| |
| const |
| DEFAULT_THRIFT_PIPE_OPEN_TIMEOUT = 10; // default: fail fast on open |
| |
| |
| type |
| //--- Pipe Streams --- |
| |
| |
| TPipeStreamBase = class( TThriftStreamImpl) |
| strict protected |
| FPipe : THandle; |
| FTimeout : DWORD; |
| FOpenTimeOut : DWORD; // separate value to allow for fail-fast-on-open scenarios |
| FOverlapped : Boolean; |
| |
| procedure Write( const pBuf : Pointer; offset, count : Integer); override; |
| function Read( const pBuf : Pointer; const buflen : Integer; offset: Integer; count: Integer): Integer; override; |
| //procedure Open; override; - see derived classes |
| procedure Close; override; |
| procedure Flush; override; |
| |
| function ReadDirect( const pBuf : Pointer; const buflen : Integer; offset: Integer; count: Integer): Integer; overload; |
| function ReadOverlapped( const pBuf : Pointer; const buflen : Integer; offset: Integer; count: Integer): Integer; overload; |
| procedure WriteDirect( const pBuf : Pointer; offset: Integer; count: Integer); overload; |
| procedure WriteOverlapped( const pBuf : Pointer; offset: Integer; count: Integer); overload; |
| |
| function IsOpen: Boolean; override; |
| function ToArray: TBytes; override; |
| public |
| constructor Create( aEnableOverlapped : Boolean; |
| const aTimeOut : DWORD = DEFAULT_THRIFT_TIMEOUT; |
| const aOpenTimeOut : DWORD = DEFAULT_THRIFT_PIPE_OPEN_TIMEOUT |
| ); reintroduce; overload; |
| |
| destructor Destroy; override; |
| end; |
| |
| |
| TNamedPipeStreamImpl = class sealed( TPipeStreamBase) |
| strict private |
| FPipeName : string; |
| FShareMode : DWORD; |
| FSecurityAttribs : PSecurityAttributes; |
| |
| strict protected |
| procedure Open; override; |
| |
| public |
| constructor Create( const aPipeName : string; |
| const aEnableOverlapped : Boolean; |
| const aShareMode: DWORD = 0; |
| const aSecurityAttributes: PSecurityAttributes = nil; |
| const aTimeOut : DWORD = DEFAULT_THRIFT_TIMEOUT; |
| const aOpenTimeOut : DWORD = DEFAULT_THRIFT_PIPE_OPEN_TIMEOUT |
| ); reintroduce; overload; |
| end; |
| |
| |
| THandlePipeStreamImpl = class sealed( TPipeStreamBase) |
| strict private |
| FSrcHandle : THandle; |
| |
| strict protected |
| procedure Open; override; |
| |
| public |
| constructor Create( const aPipeHandle : THandle; |
| const aOwnsHandle, aEnableOverlapped : Boolean; |
| const aTimeOut : DWORD = DEFAULT_THRIFT_TIMEOUT |
| ); reintroduce; overload; |
| |
| destructor Destroy; override; |
| end; |
| |
| |
| //--- Pipe Transports --- |
| |
| |
| IPipeTransport = interface( IStreamTransport) |
| ['{5E05CC85-434F-428F-BFB2-856A168B5558}'] |
| end; |
| |
| |
| TPipeTransportBase = class( TStreamTransportImpl, IPipeTransport) |
| strict protected |
| // ITransport |
| function GetIsOpen: Boolean; override; |
| procedure Open; override; |
| procedure Close; override; |
| end; |
| |
| |
| TNamedPipeTransportClientEndImpl = class( TPipeTransportBase) |
| public |
| // Named pipe constructors |
| constructor Create( const aPipe : THandle; |
| const aOwnsHandle : Boolean; |
| const aTimeOut : DWORD; |
| const aConfig : IThriftConfiguration = nil |
| ); reintroduce; overload; |
| |
| constructor Create( const aPipeName : string; |
| const aShareMode: DWORD = 0; |
| const aSecurityAttributes: PSecurityAttributes = nil; |
| const aTimeOut : DWORD = DEFAULT_THRIFT_TIMEOUT; |
| const aOpenTimeOut : DWORD = DEFAULT_THRIFT_PIPE_OPEN_TIMEOUT; |
| const aConfig : IThriftConfiguration = nil |
| ); reintroduce; overload; |
| end; |
| |
| |
| TNamedPipeTransportServerEndImpl = class( TNamedPipeTransportClientEndImpl) |
| strict private |
| FHandle : THandle; |
| strict protected |
| // ITransport |
| procedure Close; override; |
| public |
| constructor Create( const aPipe : THandle; |
| const aOwnsHandle : Boolean; |
| const aTimeOut : DWORD = DEFAULT_THRIFT_TIMEOUT; |
| const aConfig : IThriftConfiguration = nil |
| ); reintroduce; overload; |
| |
| end; |
| |
| |
| TAnonymousPipeTransportImpl = class( TPipeTransportBase) |
| public |
| // Anonymous pipe constructor |
| constructor Create( const aPipeRead, aPipeWrite : THandle; |
| const aOwnsHandles : Boolean; |
| const aTimeOut : DWORD = DEFAULT_THRIFT_TIMEOUT; |
| const aConfig : IThriftConfiguration = nil |
| ); reintroduce; overload; |
| end; |
| |
| |
| //--- Server Transports --- |
| |
| |
| IAnonymousPipeServerTransport = interface( IServerTransport) |
| ['{7AEE6793-47B9-4E49-981A-C39E9108E9AD}'] |
| // Server side anonymous pipe ends |
| function ReadHandle : THandle; |
| function WriteHandle : THandle; |
| // Client side anonymous pipe ends |
| function ClientAnonRead : THandle; |
| function ClientAnonWrite : THandle; |
| end; |
| |
| |
| INamedPipeServerTransport = interface( IServerTransport) |
| ['{9DF9EE48-D065-40AF-8F67-D33037D3D960}'] |
| function Handle : THandle; |
| end; |
| |
| |
| TPipeServerTransportBase = class( TServerTransportImpl) |
| strict protected |
| FStopServer : TEvent; |
| procedure InternalClose; virtual; abstract; |
| function QueryStopServer : Boolean; |
| public |
| constructor Create( const aConfig : IThriftConfiguration); |
| destructor Destroy; override; |
| procedure Listen; override; |
| procedure Close; override; |
| end; |
| |
| |
| TAnonymousPipeServerTransportImpl = class( TPipeServerTransportBase, IAnonymousPipeServerTransport) |
| strict private |
| FBufSize : DWORD; |
| |
| // Server side anonymous pipe handles |
| FReadHandle, |
| FWriteHandle : THandle; |
| |
| //Client side anonymous pipe handles |
| FClientAnonRead, |
| FClientAnonWrite : THandle; |
| |
| FTimeOut: DWORD; |
| strict protected |
| function Accept(const fnAccepting: TProc): ITransport; override; |
| |
| function CreateAnonPipe : Boolean; |
| |
| // IAnonymousPipeServerTransport |
| function ReadHandle : THandle; |
| function WriteHandle : THandle; |
| function ClientAnonRead : THandle; |
| function ClientAnonWrite : THandle; |
| |
| procedure InternalClose; override; |
| |
| public |
| constructor Create( const aBufsize : Cardinal = 4096; |
| const aTimeOut : DWORD = DEFAULT_THRIFT_TIMEOUT; |
| const aConfig : IThriftConfiguration = nil |
| ); reintroduce; overload; |
| end; |
| |
| |
| TNamedPipeServerTransportImpl = class( TPipeServerTransportBase, INamedPipeServerTransport) |
| strict private |
| FPipeName : string; |
| FMaxConns : DWORD; |
| FBufSize : DWORD; |
| FTimeout : DWORD; |
| FHandle : THandle; |
| FConnected : Boolean; |
| |
| |
| strict protected |
| function Accept(const fnAccepting: TProc): ITransport; override; |
| function CreateNamedPipe : THandle; |
| function CreateTransportInstance : ITransport; |
| |
| // INamedPipeServerTransport |
| function Handle : THandle; |
| procedure InternalClose; override; |
| |
| public |
| constructor Create( const aPipename : string; |
| const aBufsize : Cardinal = 4096; |
| const aMaxConns : Cardinal = PIPE_UNLIMITED_INSTANCES; |
| const aTimeOut : Cardinal = INFINITE; |
| const aConfig : IThriftConfiguration = nil |
| ); reintroduce; overload; |
| end; |
| |
| |
| implementation |
| |
| |
| procedure ClosePipeHandle( var hPipe : THandle); |
| begin |
| if hPipe <> INVALID_HANDLE_VALUE |
| then try |
| CloseHandle( hPipe); |
| finally |
| hPipe := INVALID_HANDLE_VALUE; |
| end; |
| end; |
| |
| |
| function DuplicatePipeHandle( const hSource : THandle) : THandle; |
| begin |
| if not DuplicateHandle( GetCurrentProcess, hSource, |
| GetCurrentProcess, @result, |
| 0, FALSE, DUPLICATE_SAME_ACCESS) |
| then raise TTransportExceptionNotOpen.Create('DuplicateHandle: '+SysErrorMessage(GetLastError)); |
| end; |
| |
| |
| |
| { TPipeStreamBase } |
| |
| |
| constructor TPipeStreamBase.Create( aEnableOverlapped : Boolean; const aTimeOut, aOpenTimeOut : DWORD); |
| begin |
| inherited Create; |
| FPipe := INVALID_HANDLE_VALUE; |
| FTimeout := aTimeOut; |
| FOpenTimeOut := aOpenTimeOut; |
| FOverlapped := aEnableOverlapped; |
| ASSERT( FTimeout > 0); // FOpenTimeout may be 0 |
| end; |
| |
| |
| destructor TPipeStreamBase.Destroy; |
| begin |
| try |
| Close; |
| finally |
| inherited Destroy; |
| end; |
| end; |
| |
| |
| procedure TPipeStreamBase.Close; |
| begin |
| ClosePipeHandle( FPipe); |
| end; |
| |
| |
| procedure TPipeStreamBase.Flush; |
| begin |
| FlushFileBuffers( FPipe); |
| end; |
| |
| |
| function TPipeStreamBase.IsOpen: Boolean; |
| begin |
| result := (FPipe <> INVALID_HANDLE_VALUE); |
| end; |
| |
| |
| procedure TPipeStreamBase.Write( const pBuf : Pointer; offset, count : Integer); |
| begin |
| if FOverlapped |
| then WriteOverlapped( pBuf, offset, count) |
| else WriteDirect( pBuf, offset, count); |
| end; |
| |
| |
| function TPipeStreamBase.Read( const pBuf : Pointer; const buflen : Integer; offset: Integer; count: Integer): Integer; |
| begin |
| if FOverlapped |
| then result := ReadOverlapped( pBuf, buflen, offset, count) |
| else result := ReadDirect( pBuf, buflen, offset, count); |
| end; |
| |
| |
| procedure TPipeStreamBase.WriteDirect( const pBuf : Pointer; offset: Integer; count: Integer); |
| var cbWritten, nBytes : DWORD; |
| pData : PByte; |
| begin |
| if not IsOpen |
| then raise TTransportExceptionNotOpen.Create('Called write on non-open pipe'); |
| |
| // if necessary, send the data in chunks |
| // there's a system limit around 0x10000 bytes that we hit otherwise |
| // MSDN: "Pipe write operations across a network are limited to 65,535 bytes per write. For more information regarding pipes, see the Remarks section." |
| nBytes := Min( 15*4096, count); // 16 would exceed the limit |
| pData := pBuf; |
| Inc( pData, offset); |
| while nBytes > 0 do begin |
| if not WriteFile( FPipe, pData^, nBytes, cbWritten, nil) |
| then raise TTransportExceptionNotOpen.Create('Write to pipe failed'); |
| |
| Inc( pData, cbWritten); |
| Dec( count, cbWritten); |
| nBytes := Min( nBytes, count); |
| end; |
| end; |
| |
| |
| procedure TPipeStreamBase.WriteOverlapped( const pBuf : Pointer; offset: Integer; count: Integer); |
| var cbWritten, dwWait, dwError, nBytes : DWORD; |
| overlapped : IOverlappedHelper; |
| pData : PByte; |
| begin |
| if not IsOpen |
| then raise TTransportExceptionNotOpen.Create('Called write on non-open pipe'); |
| |
| // if necessary, send the data in chunks |
| // there's a system limit around 0x10000 bytes that we hit otherwise |
| // MSDN: "Pipe write operations across a network are limited to 65,535 bytes per write. For more information regarding pipes, see the Remarks section." |
| nBytes := Min( 15*4096, count); // 16 would exceed the limit |
| pData := pBuf; |
| Inc( pData, offset); |
| while nBytes > 0 do begin |
| overlapped := TOverlappedHelperImpl.Create; |
| if not WriteFile( FPipe, pData^, nBytes, cbWritten, overlapped.OverlappedPtr) |
| then begin |
| dwError := GetLastError; |
| case dwError of |
| ERROR_IO_PENDING : begin |
| dwWait := overlapped.WaitFor(FTimeout); |
| |
| if (dwWait = WAIT_TIMEOUT) then begin |
| CancelIo( FPipe); // prevents possible AV on invalid overlapped ptr |
| raise TTransportExceptionTimedOut.Create('Pipe write timed out'); |
| end; |
| |
| if (dwWait <> WAIT_OBJECT_0) |
| or not GetOverlappedResult( FPipe, overlapped.Overlapped, cbWritten, TRUE) |
| then raise TTransportExceptionUnknown.Create('Pipe write error'); |
| end; |
| |
| else |
| raise TTransportExceptionUnknown.Create(SysErrorMessage(dwError)); |
| end; |
| end; |
| |
| ASSERT( DWORD(nBytes) = cbWritten); |
| |
| Inc( pData, cbWritten); |
| Dec( count, cbWritten); |
| nBytes := Min( nBytes, count); |
| end; |
| end; |
| |
| |
| function TPipeStreamBase.ReadDirect( const pBuf : Pointer; const buflen : Integer; offset: Integer; count: Integer): Integer; |
| var cbRead, dwErr, nRemaining : DWORD; |
| bytes, retries : LongInt; |
| bOk : Boolean; |
| pData : PByte; |
| const INTERVAL = 10; // ms |
| begin |
| if not IsOpen |
| then raise TTransportExceptionNotOpen.Create('Called read on non-open pipe'); |
| |
| // MSDN: Handle can be a handle to a named pipe instance, |
| // or it can be a handle to the read end of an anonymous pipe, |
| // The handle must have GENERIC_READ access to the pipe. |
| if FTimeOut <> INFINITE then begin |
| retries := Max( 1, Round( 1.0 * FTimeOut / INTERVAL)); |
| while TRUE do begin |
| if not PeekNamedPipe( FPipe, nil, 0, nil, @bytes, nil) then begin |
| dwErr := GetLastError; |
| if (dwErr = ERROR_INVALID_HANDLE) |
| or (dwErr = ERROR_BROKEN_PIPE) |
| or (dwErr = ERROR_PIPE_NOT_CONNECTED) |
| then begin |
| result := 0; // other side closed the pipe |
| Exit; |
| end; |
| end |
| else if bytes > 0 then begin |
| Break; // there are data |
| end; |
| |
| Dec( retries); |
| if retries > 0 |
| then Sleep( INTERVAL) |
| else raise TTransportExceptionTimedOut.Create('Pipe read timed out'); |
| end; |
| end; |
| |
| result := 0; |
| nRemaining := count; |
| pData := pBuf; |
| Inc( pData, offset); |
| while nRemaining > 0 do begin |
| // read the data (or block INFINITE-ly) |
| bOk := ReadFile( FPipe, pData^, nRemaining, cbRead, nil); |
| if (not bOk) and (GetLastError() <> ERROR_MORE_DATA) |
| then Break; // No more data, possibly because client disconnected. |
| |
| Dec( nRemaining, cbRead); |
| Inc( pData, cbRead); |
| Inc( result, cbRead); |
| end; |
| end; |
| |
| |
| function TPipeStreamBase.ReadOverlapped( const pBuf : Pointer; const buflen : Integer; offset: Integer; count: Integer): Integer; |
| var cbRead, dwWait, dwError, nRemaining : DWORD; |
| bOk : Boolean; |
| overlapped : IOverlappedHelper; |
| pData : PByte; |
| begin |
| if not IsOpen |
| then raise TTransportExceptionNotOpen.Create('Called read on non-open pipe'); |
| |
| result := 0; |
| nRemaining := count; |
| pData := pBuf; |
| Inc( pData, offset); |
| while nRemaining > 0 do begin |
| overlapped := TOverlappedHelperImpl.Create; |
| |
| // read the data |
| bOk := ReadFile( FPipe, pData^, nRemaining, cbRead, overlapped.OverlappedPtr); |
| if not bOk then begin |
| dwError := GetLastError; |
| case dwError of |
| ERROR_IO_PENDING : begin |
| dwWait := overlapped.WaitFor(FTimeout); |
| |
| if (dwWait = WAIT_TIMEOUT) then begin |
| CancelIo( FPipe); // prevents possible AV on invalid overlapped ptr |
| raise TTransportExceptionTimedOut.Create('Pipe read timed out'); |
| end; |
| |
| if (dwWait <> WAIT_OBJECT_0) |
| or not GetOverlappedResult( FPipe, overlapped.Overlapped, cbRead, TRUE) |
| then raise TTransportExceptionUnknown.Create('Pipe read error'); |
| end; |
| |
| else |
| raise TTransportExceptionUnknown.Create(SysErrorMessage(dwError)); |
| end; |
| end; |
| |
| ASSERT( cbRead > 0); // see TTransportImpl.ReadAll() |
| ASSERT( cbRead <= DWORD(nRemaining)); |
| Dec( nRemaining, cbRead); |
| Inc( pData, cbRead); |
| Inc( result, cbRead); |
| end; |
| end; |
| |
| |
| function TPipeStreamBase.ToArray: TBytes; |
| var bytes : LongInt; |
| begin |
| SetLength( result, 0); |
| bytes := 0; |
| |
| if IsOpen |
| and PeekNamedPipe( FPipe, nil, 0, nil, @bytes, nil) |
| and (bytes > 0) |
| then begin |
| SetLength( result, bytes); |
| Read( result, 0, bytes); |
| end; |
| end; |
| |
| |
| { TNamedPipeStreamImpl } |
| |
| |
| constructor TNamedPipeStreamImpl.Create( const aPipeName : string; |
| const aEnableOverlapped : Boolean; |
| const aShareMode: DWORD; |
| const aSecurityAttributes: PSecurityAttributes; |
| const aTimeOut, aOpenTimeOut : DWORD); |
| begin |
| inherited Create( aEnableOverlapped, aTimeOut, aOpenTimeOut); |
| |
| FPipeName := aPipeName; |
| FShareMode := aShareMode; |
| FSecurityAttribs := aSecurityAttributes; |
| |
| if Copy(FPipeName,1,2) <> '\\' |
| then FPipeName := '\\.\pipe\' + FPipeName; // assume localhost |
| end; |
| |
| |
| procedure TNamedPipeStreamImpl.Open; |
| var hPipe : THandle; |
| retries, timeout, dwErr : DWORD; |
| const INTERVAL = 10; // ms |
| begin |
| if IsOpen then Exit; |
| |
| retries := Max( 1, Round( 1.0 * FOpenTimeOut / INTERVAL)); |
| timeout := FOpenTimeOut; |
| |
| // if the server hasn't gotten to the point where the pipe has been created, at least wait the timeout |
| // According to MSDN, if no instances of the specified named pipe exist, the WaitNamedPipe function |
| // returns IMMEDIATELY, regardless of the time-out value. |
| // Always use INTERVAL, since WaitNamedPipe(0) defaults to some other value |
| while not WaitNamedPipe( PChar(FPipeName), INTERVAL) do begin |
| dwErr := GetLastError; |
| if dwErr <> ERROR_FILE_NOT_FOUND |
| then raise TTransportExceptionNotOpen.Create('Unable to open pipe, '+SysErrorMessage(dwErr)); |
| |
| if timeout <> INFINITE then begin |
| if (retries > 0) |
| then Dec(retries) |
| else raise TTransportExceptionNotOpen.Create('Unable to open pipe, timed out'); |
| end; |
| |
| Sleep(INTERVAL) |
| end; |
| |
| // open that thingy |
| hPipe := CreateFile( PChar( FPipeName), |
| GENERIC_READ or GENERIC_WRITE, |
| FShareMode, // sharing |
| FSecurityAttribs, // security attributes |
| OPEN_EXISTING, // opens existing pipe |
| FILE_FLAG_OVERLAPPED or FILE_FLAG_WRITE_THROUGH, // async+fast, please |
| 0); // no template file |
| |
| if hPipe = INVALID_HANDLE_VALUE |
| then raise TTransportExceptionNotOpen.Create('Unable to open pipe, '+SysErrorMessage(GetLastError)); |
| |
| // everything fine |
| FPipe := hPipe; |
| end; |
| |
| |
| { THandlePipeStreamImpl } |
| |
| |
| constructor THandlePipeStreamImpl.Create( const aPipeHandle : THandle; |
| const aOwnsHandle, aEnableOverlapped : Boolean; |
| const aTimeOut : DWORD); |
| begin |
| inherited Create( aEnableOverlapped, aTimeout, aTimeout); |
| |
| if aOwnsHandle |
| then FSrcHandle := aPipeHandle |
| else FSrcHandle := DuplicatePipeHandle( aPipeHandle); |
| |
| Open; |
| end; |
| |
| |
| destructor THandlePipeStreamImpl.Destroy; |
| begin |
| try |
| ClosePipeHandle( FSrcHandle); |
| finally |
| inherited Destroy; |
| end; |
| end; |
| |
| |
| procedure THandlePipeStreamImpl.Open; |
| begin |
| if not IsOpen |
| then FPipe := DuplicatePipeHandle( FSrcHandle); |
| end; |
| |
| |
| { TPipeTransportBase } |
| |
| |
| function TPipeTransportBase.GetIsOpen: Boolean; |
| begin |
| result := (FInputStream <> nil) and (FInputStream.IsOpen) |
| and (FOutputStream <> nil) and (FOutputStream.IsOpen); |
| end; |
| |
| |
| procedure TPipeTransportBase.Open; |
| begin |
| FInputStream.Open; |
| FOutputStream.Open; |
| end; |
| |
| |
| procedure TPipeTransportBase.Close; |
| begin |
| FInputStream.Close; |
| FOutputStream.Close; |
| end; |
| |
| |
| { TNamedPipeTransportClientEndImpl } |
| |
| |
| constructor TNamedPipeTransportClientEndImpl.Create( const aPipeName : string; |
| const aShareMode: DWORD; |
| const aSecurityAttributes: PSecurityAttributes; |
| const aTimeOut, aOpenTimeOut : DWORD; |
| const aConfig : IThriftConfiguration); |
| // Named pipe constructor |
| begin |
| inherited Create( nil, nil, aConfig); |
| FInputStream := TNamedPipeStreamImpl.Create( aPipeName, TRUE, aShareMode, aSecurityAttributes, aTimeOut, aOpenTimeOut); |
| FOutputStream := FInputStream; // true for named pipes |
| end; |
| |
| |
| constructor TNamedPipeTransportClientEndImpl.Create( const aPipe : THandle; |
| const aOwnsHandle : Boolean; |
| const aTimeOut : DWORD; |
| const aConfig : IThriftConfiguration); |
| // Named pipe constructor |
| begin |
| inherited Create( nil, nil, aConfig); |
| FInputStream := THandlePipeStreamImpl.Create( aPipe, aOwnsHandle, TRUE, aTimeOut); |
| FOutputStream := FInputStream; // true for named pipes |
| end; |
| |
| |
| { TNamedPipeTransportServerEndImpl } |
| |
| |
| constructor TNamedPipeTransportServerEndImpl.Create( const aPipe : THandle; |
| const aOwnsHandle : Boolean; |
| const aTimeOut : DWORD; |
| const aConfig : IThriftConfiguration); |
| // Named pipe constructor |
| begin |
| FHandle := DuplicatePipeHandle( aPipe); |
| inherited Create( aPipe, aOwnsHandle, aTimeout, aConfig); |
| end; |
| |
| |
| procedure TNamedPipeTransportServerEndImpl.Close; |
| begin |
| FlushFileBuffers( FHandle); |
| DisconnectNamedPipe( FHandle); // force client off the pipe |
| ClosePipeHandle( FHandle); |
| |
| inherited Close; |
| end; |
| |
| |
| { TAnonymousPipeTransportImpl } |
| |
| |
| constructor TAnonymousPipeTransportImpl.Create( const aPipeRead, aPipeWrite : THandle; |
| const aOwnsHandles : Boolean; |
| const aTimeOut : DWORD; |
| const aConfig : IThriftConfiguration); |
| // Anonymous pipe constructor |
| begin |
| inherited Create( nil, nil, aConfig); |
| // overlapped is not supported with AnonPipes, see MSDN |
| FInputStream := THandlePipeStreamImpl.Create( aPipeRead, aOwnsHandles, FALSE, aTimeout); |
| FOutputStream := THandlePipeStreamImpl.Create( aPipeWrite, aOwnsHandles, FALSE, aTimeout); |
| end; |
| |
| |
| { TPipeServerTransportBase } |
| |
| |
| constructor TPipeServerTransportBase.Create( const aConfig : IThriftConfiguration); |
| begin |
| inherited Create( aConfig); |
| FStopServer := TEvent.Create(nil,TRUE,FALSE,''); // manual reset |
| end; |
| |
| |
| destructor TPipeServerTransportBase.Destroy; |
| begin |
| try |
| FreeAndNil( FStopServer); |
| finally |
| inherited Destroy; |
| end; |
| end; |
| |
| |
| function TPipeServerTransportBase.QueryStopServer : Boolean; |
| begin |
| result := (FStopServer = nil) |
| or (FStopServer.WaitFor(0) <> wrTimeout); |
| end; |
| |
| |
| procedure TPipeServerTransportBase.Listen; |
| begin |
| FStopServer.ResetEvent; |
| end; |
| |
| |
| procedure TPipeServerTransportBase.Close; |
| begin |
| FStopServer.SetEvent; |
| InternalClose; |
| end; |
| |
| |
| { TAnonymousPipeServerTransportImpl } |
| |
| constructor TAnonymousPipeServerTransportImpl.Create( const aBufsize : Cardinal; |
| const aTimeOut : DWORD; |
| const aConfig : IThriftConfiguration); |
| // Anonymous pipe CTOR |
| begin |
| inherited Create(aConfig); |
| FBufsize := aBufSize; |
| FReadHandle := INVALID_HANDLE_VALUE; |
| FWriteHandle := INVALID_HANDLE_VALUE; |
| FClientAnonRead := INVALID_HANDLE_VALUE; |
| FClientAnonWrite := INVALID_HANDLE_VALUE; |
| FTimeOut := aTimeOut; |
| |
| // The anonymous pipe needs to be created first so that the server can |
| // pass the handles on to the client before the serve (acceptImpl) |
| // blocking call. |
| if not CreateAnonPipe |
| then raise TTransportExceptionNotOpen.Create(ClassName+'.Create() failed'); |
| end; |
| |
| |
| function TAnonymousPipeServerTransportImpl.Accept(const fnAccepting: TProc): ITransport; |
| var buf : Byte; |
| br : DWORD; |
| begin |
| if Assigned(fnAccepting) |
| then fnAccepting(); |
| |
| // This 0-byte read serves merely as a blocking call. |
| if not ReadFile( FReadHandle, buf, 0, br, nil) |
| and (GetLastError() <> ERROR_MORE_DATA) |
| then raise TTransportExceptionNotOpen.Create('TServerPipe unable to initiate pipe communication'); |
| |
| // create the transport impl |
| result := TAnonymousPipeTransportImpl.Create( FReadHandle, FWriteHandle, FALSE, FTimeOut, Configuration); |
| end; |
| |
| |
| procedure TAnonymousPipeServerTransportImpl.InternalClose; |
| begin |
| ClosePipeHandle( FReadHandle); |
| ClosePipeHandle( FWriteHandle); |
| ClosePipeHandle( FClientAnonRead); |
| ClosePipeHandle( FClientAnonWrite); |
| end; |
| |
| |
| function TAnonymousPipeServerTransportImpl.ReadHandle : THandle; |
| begin |
| result := FReadHandle; |
| end; |
| |
| |
| function TAnonymousPipeServerTransportImpl.WriteHandle : THandle; |
| begin |
| result := FWriteHandle; |
| end; |
| |
| |
| function TAnonymousPipeServerTransportImpl.ClientAnonRead : THandle; |
| begin |
| result := FClientAnonRead; |
| end; |
| |
| |
| function TAnonymousPipeServerTransportImpl.ClientAnonWrite : THandle; |
| begin |
| result := FClientAnonWrite; |
| end; |
| |
| |
| function TAnonymousPipeServerTransportImpl.CreateAnonPipe : Boolean; |
| var sd : PSECURITY_DESCRIPTOR; |
| sa : SECURITY_ATTRIBUTES; //TSecurityAttributes; |
| hCAR, hPipeW, hCAW, hPipe : THandle; |
| begin |
| sd := PSECURITY_DESCRIPTOR( LocalAlloc( LPTR,SECURITY_DESCRIPTOR_MIN_LENGTH)); |
| try |
| Win32Check( InitializeSecurityDescriptor( sd, SECURITY_DESCRIPTOR_REVISION)); |
| Win32Check( SetSecurityDescriptorDacl( sd, TRUE, nil, FALSE)); |
| |
| sa.nLength := sizeof( sa); |
| sa.lpSecurityDescriptor := sd; |
| sa.bInheritHandle := TRUE; //allow passing handle to child |
| |
| Result := CreatePipe( hCAR, hPipeW, @sa, FBufSize); //create stdin pipe |
| if not Result then begin //create stdin pipe |
| raise TTransportExceptionNotOpen.Create('TServerPipe CreatePipe (anon) failed, '+SysErrorMessage(GetLastError)); |
| Exit; |
| end; |
| |
| Result := CreatePipe( hPipe, hCAW, @sa, FBufSize); //create stdout pipe |
| if not Result then begin //create stdout pipe |
| CloseHandle( hCAR); |
| CloseHandle( hPipeW); |
| raise TTransportExceptionNotOpen.Create('TServerPipe CreatePipe (anon) failed, '+SysErrorMessage(GetLastError)); |
| Exit; |
| end; |
| |
| FClientAnonRead := hCAR; |
| FClientAnonWrite := hCAW; |
| FReadHandle := hPipe; |
| FWriteHandle := hPipeW; |
| finally |
| if sd <> nil then LocalFree( Cardinal(sd)); |
| end; |
| end; |
| |
| |
| { TNamedPipeServerTransportImpl } |
| |
| |
| constructor TNamedPipeServerTransportImpl.Create( const aPipename : string; |
| const aBufsize, aMaxConns, aTimeOut : Cardinal; |
| const aConfig : IThriftConfiguration); |
| // Named Pipe CTOR |
| begin |
| inherited Create( aConfig); |
| FPipeName := aPipename; |
| FBufsize := aBufSize; |
| FMaxConns := Max( 1, Min( PIPE_UNLIMITED_INSTANCES, aMaxConns)); |
| FHandle := INVALID_HANDLE_VALUE; |
| FTimeout := aTimeOut; |
| FConnected := FALSE; |
| ASSERT( FTimeout > 0); |
| |
| if Copy(FPipeName,1,2) <> '\\' |
| then FPipeName := '\\.\pipe\' + FPipeName; // assume localhost |
| end; |
| |
| |
| function TNamedPipeServerTransportImpl.Accept(const fnAccepting: TProc): ITransport; |
| var dwError, dwWait, dwDummy : DWORD; |
| overlapped : IOverlappedHelper; |
| handles : array[0..1] of THandle; |
| begin |
| overlapped := TOverlappedHelperImpl.Create; |
| |
| ASSERT( not FConnected); |
| CreateNamedPipe; |
| while not FConnected do begin |
| |
| if QueryStopServer then begin |
| InternalClose; |
| Abort; |
| end; |
| |
| if Assigned(fnAccepting) |
| then fnAccepting(); |
| |
| // Wait for the client to connect; if it succeeds, the |
| // function returns a nonzero value. If the function returns |
| // zero, GetLastError should return ERROR_PIPE_CONNECTED. |
| if ConnectNamedPipe( Handle, overlapped.OverlappedPtr) then begin |
| FConnected := TRUE; |
| Break; |
| end; |
| |
| // ConnectNamedPipe() returns FALSE for OverlappedIO, even if connected. |
| // We have to check GetLastError() explicitly to find out |
| dwError := GetLastError; |
| case dwError of |
| ERROR_PIPE_CONNECTED : begin |
| FConnected := not QueryStopServer; // special case: pipe immediately connected |
| end; |
| |
| ERROR_IO_PENDING : begin |
| handles[0] := overlapped.WaitHandle; |
| handles[1] := FStopServer.Handle; |
| dwWait := WaitForMultipleObjects( 2, @handles, FALSE, FTimeout); |
| FConnected := (dwWait = WAIT_OBJECT_0) |
| and GetOverlappedResult( Handle, overlapped.Overlapped, dwDummy, TRUE) |
| and not QueryStopServer; |
| end; |
| |
| else |
| InternalClose; |
| raise TTransportExceptionNotOpen.Create('Client connection failed'); |
| end; |
| end; |
| |
| // create the transport impl |
| result := CreateTransportInstance; |
| end; |
| |
| |
| function TNamedPipeServerTransportImpl.CreateTransportInstance : ITransport; |
| // create the transport impl |
| var hPipe : THandle; |
| begin |
| hPipe := THandle( InterlockedExchangePointer( Pointer(FHandle), Pointer(INVALID_HANDLE_VALUE))); |
| try |
| FConnected := FALSE; |
| result := TNamedPipeTransportServerEndImpl.Create( hPipe, TRUE, FTimeout, Configuration); |
| except |
| ClosePipeHandle(hPipe); |
| raise; |
| end; |
| end; |
| |
| |
| procedure TNamedPipeServerTransportImpl.InternalClose; |
| var hPipe : THandle; |
| begin |
| hPipe := THandle( InterlockedExchangePointer( Pointer(FHandle), Pointer(INVALID_HANDLE_VALUE))); |
| if hPipe = INVALID_HANDLE_VALUE then Exit; |
| |
| try |
| if FConnected |
| then FlushFileBuffers( hPipe) |
| else CancelIo( hPipe); |
| DisconnectNamedPipe( hPipe); |
| finally |
| ClosePipeHandle( hPipe); |
| FConnected := FALSE; |
| end; |
| end; |
| |
| |
| function TNamedPipeServerTransportImpl.Handle : THandle; |
| begin |
| {$IFDEF WIN64} |
| result := THandle( InterlockedExchangeAdd64( Int64(FHandle), 0)); |
| {$ELSE} |
| result := THandle( InterlockedExchangeAdd( Integer(FHandle), 0)); |
| {$ENDIF} |
| end; |
| |
| |
| function TNamedPipeServerTransportImpl.CreateNamedPipe : THandle; |
| var SIDAuthWorld : SID_IDENTIFIER_AUTHORITY ; |
| everyone_sid : PSID; |
| ea : EXPLICIT_ACCESS; |
| acl : PACL; |
| sd : PSECURITY_DESCRIPTOR; |
| sa : SECURITY_ATTRIBUTES; |
| const |
| SECURITY_WORLD_SID_AUTHORITY : TSIDIdentifierAuthority = (Value : (0,0,0,0,0,1)); |
| SECURITY_WORLD_RID = $00000000; |
| begin |
| sd := nil; |
| everyone_sid := nil; |
| try |
| ASSERT( (FHandle = INVALID_HANDLE_VALUE) and not FConnected); |
| |
| // Windows - set security to allow non-elevated apps |
| // to access pipes created by elevated apps. |
| SIDAuthWorld := SECURITY_WORLD_SID_AUTHORITY; |
| AllocateAndInitializeSid( SIDAuthWorld, 1, SECURITY_WORLD_RID, 0, 0, 0, 0, 0, 0, 0, everyone_sid); |
| |
| ZeroMemory( @ea, SizeOf(ea)); |
| ea.grfAccessPermissions := GENERIC_ALL; //SPECIFIC_RIGHTS_ALL or STANDARD_RIGHTS_ALL; |
| ea.grfAccessMode := SET_ACCESS; |
| ea.grfInheritance := NO_INHERITANCE; |
| ea.Trustee.TrusteeForm := TRUSTEE_IS_SID; |
| ea.Trustee.TrusteeType := TRUSTEE_IS_WELL_KNOWN_GROUP; |
| ea.Trustee.ptstrName := PChar(everyone_sid); |
| |
| acl := nil; |
| SetEntriesInAcl( 1, @ea, nil, acl); |
| |
| sd := PSECURITY_DESCRIPTOR( LocalAlloc( LPTR,SECURITY_DESCRIPTOR_MIN_LENGTH)); |
| Win32Check( InitializeSecurityDescriptor( sd, SECURITY_DESCRIPTOR_REVISION)); |
| Win32Check( SetSecurityDescriptorDacl( sd, TRUE, acl, FALSE)); |
| |
| sa.nLength := SizeOf(sa); |
| sa.lpSecurityDescriptor := sd; |
| sa.bInheritHandle := FALSE; |
| |
| // Create an instance of the named pipe |
| {$IFDEF OLD_UNIT_NAMES} |
| result := Windows.CreateNamedPipe( |
| {$ELSE} |
| result := Winapi.Windows.CreateNamedPipe( |
| {$ENDIF} |
| PChar( FPipeName), // pipe name |
| PIPE_ACCESS_DUPLEX or // read/write access |
| FILE_FLAG_OVERLAPPED, // async mode |
| PIPE_TYPE_BYTE or // byte type pipe |
| PIPE_READMODE_BYTE, // byte read mode |
| FMaxConns, // max. instances |
| FBufSize, // output buffer size |
| FBufSize, // input buffer size |
| FTimeout, // time-out, see MSDN |
| @sa // default security attribute |
| ); |
| |
| if( result <> INVALID_HANDLE_VALUE) |
| then InterlockedExchangePointer( Pointer(FHandle), Pointer(result)) |
| else raise TTransportExceptionNotOpen.Create('CreateNamedPipe() failed ' + IntToStr(GetLastError)); |
| |
| finally |
| if sd <> nil then LocalFree( Cardinal( sd)); |
| if acl <> nil then LocalFree( Cardinal( acl)); |
| if everyone_sid <> nil then FreeSid(everyone_sid); |
| end; |
| end; |
| |
| |
| |
| end. |
| |
| |
| |
