ceph/src/jaegertracing/thrift/lib/delphi/src/Thrift.Transport.Pipes.pas

   1 (*
   2  * Licensed to the Apache Software Foundation (ASF) under one
   3  * or more contributor license agreements. See the NOTICE file
   4  * distributed with this work for additional information
   5  * regarding copyright ownership. The ASF licenses this file
   6  * to you under the Apache License, Version 2.0 (the
   7  * "License"); you may not use this file except in compliance
   8  * with the License. You may obtain a copy of the License at
   9  *
  10  *   http://www.apache.org/licenses/LICENSE-2.0
  11  *
  12  * Unless required by applicable law or agreed to in writing,
  13  * software distributed under the License is distributed on an
  14  * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
  15  * KIND, either express or implied. See the License for the
  16  * specific language governing permissions and limitations
  17  * under the License.
  18  *)
  19 unit Thrift.Transport.Pipes;
  20
  21 {$WARN SYMBOL_PLATFORM OFF}
  22 {$I Thrift.Defines.inc}
  23
  24 interface
  25
  26 uses
  27   {$IFDEF OLD_UNIT_NAMES}
  28   Windows, SysUtils, Math, AccCtrl, AclAPI, SyncObjs,
  29   {$ELSE}
  30   Winapi.Windows, System.SysUtils, System.Math, Winapi.AccCtrl, Winapi.AclAPI, System.SyncObjs,
  31   {$ENDIF}
  32   Thrift.Transport,
  33   Thrift.Utils,
  34   Thrift.Stream;
  35
  36 const
  37   DEFAULT_THRIFT_PIPE_OPEN_TIMEOUT = 10;  // default: fail fast on open
  38
  39
  40 type
  41   //--- Pipe Streams ---
  42
  43
  44   TPipeStreamBase = class( TThriftStreamImpl)
  45   strict protected
  46     FPipe    : THandle;
  47     FTimeout : DWORD;
  48     FOpenTimeOut : DWORD;  // separate value to allow for fail-fast-on-open scenarios
  49     FOverlapped : Boolean;
  50
  51     procedure Write( const pBuf : Pointer; offset, count : Integer); override;
  52     function  Read( const pBuf : Pointer; const buflen : Integer; offset: Integer; count: Integer): Integer; override;
  53     //procedure Open; override; - see derived classes
  54     procedure Close; override;
  55     procedure Flush; override;
  56
  57     function  ReadDirect(     const pBuf : Pointer; const buflen : Integer; offset: Integer; count: Integer): Integer;  overload;
  58     function  ReadOverlapped( const pBuf : Pointer; const buflen : Integer; offset: Integer; count: Integer): Integer;  overload;
  59     procedure WriteDirect(     const pBuf : Pointer; offset: Integer; count: Integer);  overload;
  60     procedure WriteOverlapped( const pBuf : Pointer; offset: Integer; count: Integer);  overload;
  61
  62     function IsOpen: Boolean; override;
  63     function ToArray: TBytes; override;
  64   public
  65     constructor Create( aEnableOverlapped : Boolean;
  66                         const aTimeOut : DWORD = DEFAULT_THRIFT_TIMEOUT;
  67                         const aOpenTimeOut : DWORD = DEFAULT_THRIFT_PIPE_OPEN_TIMEOUT);
  68     destructor Destroy;  override;
  69   end;
  70
  71
  72   TNamedPipeStreamImpl = class sealed( TPipeStreamBase)
  73   strict private
  74     FPipeName  : string;
  75     FShareMode : DWORD;
  76     FSecurityAttribs : PSecurityAttributes;
  77
  78   strict protected
  79     procedure Open; override;
  80
  81   public
  82     constructor Create( const aPipeName : string;
  83                         const aEnableOverlapped : Boolean;
  84                         const aShareMode: DWORD = 0;
  85                         const aSecurityAttributes: PSecurityAttributes = nil;
  86                         const aTimeOut : DWORD = DEFAULT_THRIFT_TIMEOUT;
  87                         const aOpenTimeOut : DWORD = DEFAULT_THRIFT_PIPE_OPEN_TIMEOUT);  overload;
  88   end;
  89
  90
  91   THandlePipeStreamImpl = class sealed( TPipeStreamBase)
  92   strict private
  93     FSrcHandle : THandle;
  94
  95   strict protected
  96     procedure Open; override;
  97
  98   public
  99     constructor Create( const aPipeHandle : THandle;
 100                         const aOwnsHandle, aEnableOverlapped : Boolean;
 101                         const aTimeOut : DWORD = DEFAULT_THRIFT_TIMEOUT);  overload;
 102     destructor Destroy;  override;
 103   end;
 104
 105
 106   //--- Pipe Transports ---
 107
 108
 109   IPipeTransport = interface( IStreamTransport)
 110     ['{5E05CC85-434F-428F-BFB2-856A168B5558}']
 111   end;
 112
 113
 114   TPipeTransportBase = class( TStreamTransportImpl, IPipeTransport)
 115   public
 116     // ITransport
 117     function  GetIsOpen: Boolean; override;
 118     procedure Open; override;
 119     procedure Close; override;
 120   end;
 121
 122
 123   TNamedPipeTransportClientEndImpl = class( TPipeTransportBase)
 124   public
 125     // Named pipe constructors
 126     constructor Create( aPipe : THandle; aOwnsHandle : Boolean;
 127                         const aTimeOut : DWORD); overload;
 128     constructor Create( const aPipeName : string;
 129                         const aShareMode: DWORD = 0;
 130                         const aSecurityAttributes: PSecurityAttributes = nil;
 131                         const aTimeOut : DWORD = DEFAULT_THRIFT_TIMEOUT;
 132                         const aOpenTimeOut : DWORD = DEFAULT_THRIFT_PIPE_OPEN_TIMEOUT);  overload;
 133   end;
 134
 135
 136   TNamedPipeTransportServerEndImpl = class( TNamedPipeTransportClientEndImpl)
 137   strict private
 138     FHandle : THandle;
 139   public
 140     // ITransport
 141     procedure Close; override;
 142     constructor Create( aPipe : THandle; aOwnsHandle : Boolean;
 143                         const aTimeOut : DWORD = DEFAULT_THRIFT_TIMEOUT); reintroduce;
 144   end;
 145
 146
 147   TAnonymousPipeTransportImpl = class( TPipeTransportBase)
 148   public
 149     // Anonymous pipe constructor
 150     constructor Create(const aPipeRead, aPipeWrite : THandle;
 151                        aOwnsHandles : Boolean;
 152                        const aTimeOut : DWORD = DEFAULT_THRIFT_TIMEOUT); overload;
 153   end;
 154
 155
 156   //--- Server Transports ---
 157
 158
 159   IAnonymousPipeServerTransport = interface( IServerTransport)
 160     ['{7AEE6793-47B9-4E49-981A-C39E9108E9AD}']
 161     // Server side anonymous pipe ends
 162     function ReadHandle : THandle;
 163     function WriteHandle : THandle;
 164     // Client side anonymous pipe ends
 165     function ClientAnonRead : THandle;
 166     function ClientAnonWrite  : THandle;
 167   end;
 168
 169
 170   INamedPipeServerTransport = interface( IServerTransport)
 171     ['{9DF9EE48-D065-40AF-8F67-D33037D3D960}']
 172     function Handle : THandle;
 173   end;
 174
 175
 176   TPipeServerTransportBase = class( TServerTransportImpl)
 177   strict protected
 178     FStopServer : TEvent;
 179     procedure InternalClose; virtual; abstract;
 180     function QueryStopServer : Boolean;
 181   public
 182     constructor Create;
 183     destructor Destroy;  override;
 184     procedure Listen; override;
 185     procedure Close; override;
 186   end;
 187
 188
 189   TAnonymousPipeServerTransportImpl = class( TPipeServerTransportBase, IAnonymousPipeServerTransport)
 190   strict private
 191     FBufSize      : DWORD;
 192
 193     // Server side anonymous pipe handles
 194     FReadHandle,
 195     FWriteHandle : THandle;
 196
 197     //Client side anonymous pipe handles
 198     FClientAnonRead,
 199     FClientAnonWrite  : THandle;
 200
 201     FTimeOut: DWORD;
 202   protected
 203     function Accept(const fnAccepting: TProc): ITransport; override;
 204
 205     function CreateAnonPipe : Boolean;
 206
 207     // IAnonymousPipeServerTransport
 208     function ReadHandle : THandle;
 209     function WriteHandle : THandle;
 210     function ClientAnonRead : THandle;
 211     function ClientAnonWrite  : THandle;
 212
 213     procedure InternalClose; override;
 214
 215   public
 216     constructor Create(aBufsize : Cardinal = 4096; aTimeOut : DWORD = DEFAULT_THRIFT_TIMEOUT);
 217   end;
 218
 219
 220   TNamedPipeServerTransportImpl = class( TPipeServerTransportBase, INamedPipeServerTransport)
 221   strict private
 222     FPipeName     : string;
 223     FMaxConns     : DWORD;
 224     FBufSize      : DWORD;
 225     FTimeout      : DWORD;
 226     FHandle       : THandle;
 227     FConnected    : Boolean;
 228
 229
 230   strict protected
 231     function Accept(const fnAccepting: TProc): ITransport; override;
 232     function CreateNamedPipe : THandle;
 233     function CreateTransportInstance : ITransport;
 234
 235     // INamedPipeServerTransport
 236     function Handle : THandle;
 237     procedure InternalClose; override;
 238
 239   public
 240     constructor Create( aPipename : string; aBufsize : Cardinal = 4096;
 241                         aMaxConns : Cardinal = PIPE_UNLIMITED_INSTANCES;
 242                         aTimeOut : Cardinal = INFINITE);
 243   end;
 244
 245
 246 implementation
 247
 248
 249 procedure ClosePipeHandle( var hPipe : THandle);
 250 begin
 251   if hPipe <> INVALID_HANDLE_VALUE
 252   then try
 253     CloseHandle( hPipe);
 254   finally
 255     hPipe := INVALID_HANDLE_VALUE;
 256   end;
 257 end;
 258
 259
 260 function DuplicatePipeHandle( const hSource : THandle) : THandle;
 261 begin
 262   if not DuplicateHandle( GetCurrentProcess, hSource,
 263                           GetCurrentProcess, @result,
 264                           0, FALSE, DUPLICATE_SAME_ACCESS)
 265   then raise TTransportExceptionNotOpen.Create('DuplicateHandle: '+SysErrorMessage(GetLastError));
 266 end;
 267
 268
 269
 270 { TPipeStreamBase }
 271
 272
 273 constructor TPipeStreamBase.Create( aEnableOverlapped : Boolean;
 274                                     const aTimeOut, aOpenTimeOut : DWORD);
 275 begin
 276   inherited Create;
 277   ASSERT( aTimeout > 0);  // aOpenTimeout may be 0
 278   FPipe        := INVALID_HANDLE_VALUE;
 279   FTimeout     := aTimeOut;
 280   FOpenTimeOut := aOpenTimeOut;
 281   FOverlapped  := aEnableOverlapped;
 282 end;
 283
 284
 285 destructor TPipeStreamBase.Destroy;
 286 begin
 287   try
 288     Close;
 289   finally
 290     inherited Destroy;
 291   end;
 292 end;
 293
 294
 295 procedure TPipeStreamBase.Close;
 296 begin
 297   ClosePipeHandle( FPipe);
 298 end;
 299
 300
 301 procedure TPipeStreamBase.Flush;
 302 begin
 303   FlushFileBuffers( FPipe);
 304 end;
 305
 306
 307 function TPipeStreamBase.IsOpen: Boolean;
 308 begin
 309   result := (FPipe <> INVALID_HANDLE_VALUE);
 310 end;
 311
 312
 313 procedure TPipeStreamBase.Write( const pBuf : Pointer; offset, count : Integer);
 314 begin
 315   if FOverlapped
 316   then WriteOverlapped( pBuf, offset, count)
 317   else WriteDirect( pBuf, offset, count);
 318 end;
 319
 320
 321 function TPipeStreamBase.Read( const pBuf : Pointer; const buflen : Integer; offset: Integer; count: Integer): Integer;
 322 begin
 323   if FOverlapped
 324   then result := ReadOverlapped( pBuf, buflen, offset, count)
 325   else result := ReadDirect( pBuf, buflen, offset, count);
 326 end;
 327
 328
 329 procedure TPipeStreamBase.WriteDirect( const pBuf : Pointer; offset: Integer; count: Integer);
 330 var cbWritten, nBytes : DWORD;
 331     pData : PByte;
 332 begin
 333   if not IsOpen
 334   then raise TTransportExceptionNotOpen.Create('Called write on non-open pipe');
 335
 336   // if necessary, send the data in chunks
 337   // there's a system limit around 0x10000 bytes that we hit otherwise
 338   // MSDN: "Pipe write operations across a network are limited to 65,535 bytes per write. For more information regarding pipes, see the Remarks section."
 339   nBytes := Min( 15*4096, count); // 16 would exceed the limit
 340   pData  := pBuf;
 341   Inc( pData, offset);
 342   while nBytes > 0 do begin
 343     if not WriteFile( FPipe, pData^, nBytes, cbWritten, nil)
 344     then raise TTransportExceptionNotOpen.Create('Write to pipe failed');
 345
 346     Inc( pData, cbWritten);
 347     Dec( count, cbWritten);
 348     nBytes := Min( nBytes, count);
 349   end;
 350 end;
 351
 352
 353 procedure TPipeStreamBase.WriteOverlapped( const pBuf : Pointer; offset: Integer; count: Integer);
 354 var cbWritten, dwWait, dwError, nBytes : DWORD;
 355     overlapped : IOverlappedHelper;
 356     pData : PByte;
 357 begin
 358   if not IsOpen
 359   then raise TTransportExceptionNotOpen.Create('Called write on non-open pipe');
 360
 361   // if necessary, send the data in chunks
 362   // there's a system limit around 0x10000 bytes that we hit otherwise
 363   // MSDN: "Pipe write operations across a network are limited to 65,535 bytes per write. For more information regarding pipes, see the Remarks section."
 364   nBytes := Min( 15*4096, count); // 16 would exceed the limit
 365   pData  := pBuf;
 366   Inc( pData, offset);
 367   while nBytes > 0 do begin
 368     overlapped := TOverlappedHelperImpl.Create;
 369     if not WriteFile( FPipe, pData^, nBytes, cbWritten, overlapped.OverlappedPtr)
 370     then begin
 371       dwError := GetLastError;
 372       case dwError of
 373         ERROR_IO_PENDING : begin
 374           dwWait := overlapped.WaitFor(FTimeout);
 375
 376           if (dwWait = WAIT_TIMEOUT) then begin
 377             CancelIo( FPipe);  // prevents possible AV on invalid overlapped ptr
 378             raise TTransportExceptionTimedOut.Create('Pipe write timed out');
 379           end;
 380
 381           if (dwWait <> WAIT_OBJECT_0)
 382           or not GetOverlappedResult( FPipe, overlapped.Overlapped, cbWritten, TRUE)
 383           then raise TTransportExceptionUnknown.Create('Pipe write error');
 384         end;
 385
 386       else
 387         raise TTransportExceptionUnknown.Create(SysErrorMessage(dwError));
 388       end;
 389     end;
 390
 391     ASSERT( DWORD(nBytes) = cbWritten);
 392
 393     Inc( pData, cbWritten);
 394     Dec( count, cbWritten);
 395     nBytes := Min( nBytes, count);
 396   end;
 397 end;
 398
 399
 400 function TPipeStreamBase.ReadDirect(     const pBuf : Pointer; const buflen : Integer; offset: Integer; count: Integer): Integer;
 401 var cbRead, dwErr, nRemaining  : DWORD;
 402     bytes, retries  : LongInt;
 403     bOk     : Boolean;
 404     pData   : PByte;
 405 const INTERVAL = 10;  // ms
 406 begin
 407   if not IsOpen
 408   then raise TTransportExceptionNotOpen.Create('Called read on non-open pipe');
 409
 410   // MSDN: Handle can be a handle to a named pipe instance,
 411   // or it can be a handle to the read end of an anonymous pipe,
 412   // The handle must have GENERIC_READ access to the pipe.
 413   if FTimeOut <> INFINITE then begin
 414     retries := Max( 1, Round( 1.0 * FTimeOut / INTERVAL));
 415     while TRUE do begin
 416       if not PeekNamedPipe( FPipe, nil, 0, nil, @bytes, nil) then begin
 417         dwErr := GetLastError;
 418         if (dwErr = ERROR_INVALID_HANDLE)
 419         or (dwErr = ERROR_BROKEN_PIPE)
 420         or (dwErr = ERROR_PIPE_NOT_CONNECTED)
 421         then begin
 422           result := 0;  // other side closed the pipe
 423           Exit;
 424         end;
 425       end
 426       else if bytes > 0 then begin
 427         Break;  // there are data
 428       end;
 429
 430       Dec( retries);
 431       if retries > 0
 432       then Sleep( INTERVAL)
 433       else raise TTransportExceptionTimedOut.Create('Pipe read timed out');
 434     end;
 435   end;
 436
 437   result := 0;
 438   nRemaining := count;
 439   pData := pBuf;
 440   Inc( pData, offset);
 441   while nRemaining > 0 do begin
 442     // read the data (or block INFINITE-ly)
 443     bOk := ReadFile( FPipe, pData^, nRemaining, cbRead, nil);
 444     if (not bOk) and (GetLastError() <> ERROR_MORE_DATA)
 445     then Break; // No more data, possibly because client disconnected.
 446
 447     Dec( nRemaining, cbRead);
 448     Inc( pData, cbRead);
 449     Inc( result, cbRead);
 450   end;
 451 end;
 452
 453
 454 function TPipeStreamBase.ReadOverlapped( const pBuf : Pointer; const buflen : Integer; offset: Integer; count: Integer): Integer;
 455 var cbRead, dwWait, dwError, nRemaining : DWORD;
 456     bOk     : Boolean;
 457     overlapped : IOverlappedHelper;
 458     pData   : PByte;
 459 begin
 460   if not IsOpen
 461   then raise TTransportExceptionNotOpen.Create('Called read on non-open pipe');
 462
 463   result := 0;
 464   nRemaining := count;
 465   pData := pBuf;
 466   Inc( pData, offset);
 467   while nRemaining > 0 do begin
 468     overlapped := TOverlappedHelperImpl.Create;
 469
 470      // read the data
 471     bOk := ReadFile( FPipe, pData^, nRemaining, cbRead, overlapped.OverlappedPtr);
 472     if not bOk then begin
 473       dwError := GetLastError;
 474       case dwError of
 475         ERROR_IO_PENDING : begin
 476           dwWait := overlapped.WaitFor(FTimeout);
 477
 478           if (dwWait = WAIT_TIMEOUT) then begin
 479             CancelIo( FPipe);  // prevents possible AV on invalid overlapped ptr
 480             raise TTransportExceptionTimedOut.Create('Pipe read timed out');
 481           end;
 482
 483           if (dwWait <> WAIT_OBJECT_0)
 484           or not GetOverlappedResult( FPipe, overlapped.Overlapped, cbRead, TRUE)
 485           then raise TTransportExceptionUnknown.Create('Pipe read error');
 486         end;
 487
 488       else
 489         raise TTransportExceptionUnknown.Create(SysErrorMessage(dwError));
 490       end;
 491     end;
 492
 493     ASSERT( cbRead > 0);  // see TTransportImpl.ReadAll()
 494     ASSERT( cbRead <= DWORD(nRemaining));
 495     Dec( nRemaining, cbRead);
 496     Inc( pData, cbRead);
 497     Inc( result, cbRead);
 498   end;
 499 end;
 500
 501
 502 function TPipeStreamBase.ToArray: TBytes;
 503 var bytes : LongInt;
 504 begin
 505   SetLength( result, 0);
 506   bytes := 0;
 507
 508   if  IsOpen
 509   and PeekNamedPipe( FPipe, nil, 0, nil, @bytes, nil)
 510   and (bytes > 0)
 511   then begin
 512     SetLength( result, bytes);
 513     Read( result, 0, bytes);
 514   end;
 515 end;
 516
 517
 518 { TNamedPipeStreamImpl }
 519
 520
 521 constructor TNamedPipeStreamImpl.Create( const aPipeName : string;
 522                                          const aEnableOverlapped : Boolean;
 523                                          const aShareMode: DWORD;
 524                                          const aSecurityAttributes: PSecurityAttributes;
 525                                          const aTimeOut, aOpenTimeOut : DWORD);
 526 begin
 527   inherited Create( aEnableOverlapped, aTimeout, aOpenTimeOut);
 528
 529   FPipeName        := aPipeName;
 530   FShareMode       := aShareMode;
 531   FSecurityAttribs := aSecurityAttributes;
 532
 533   if Copy(FPipeName,1,2) <> '\\'
 534   then FPipeName := '\\.\pipe\' + FPipeName;  // assume localhost
 535 end;
 536
 537
 538 procedure TNamedPipeStreamImpl.Open;
 539 var hPipe    : THandle;
 540     retries, timeout, dwErr : DWORD;
 541 const INTERVAL = 10; // ms
 542 begin
 543   if IsOpen then Exit;
 544
 545   retries := Max( 1, Round( 1.0 * FOpenTimeOut / INTERVAL));
 546   timeout := FOpenTimeOut;
 547
 548   // if the server hasn't gotten to the point where the pipe has been created, at least wait the timeout
 549   // According to MSDN, if no instances of the specified named pipe exist, the WaitNamedPipe function
 550   // returns IMMEDIATELY, regardless of the time-out value.
 551   // Always use INTERVAL, since WaitNamedPipe(0) defaults to some other value
 552   while not WaitNamedPipe( PChar(FPipeName), INTERVAL) do begin
 553     dwErr := GetLastError;
 554     if dwErr <> ERROR_FILE_NOT_FOUND
 555     then raise TTransportExceptionNotOpen.Create('Unable to open pipe, '+SysErrorMessage(dwErr));
 556
 557     if timeout <> INFINITE then begin
 558       if (retries > 0)
 559       then Dec(retries)
 560       else raise TTransportExceptionNotOpen.Create('Unable to open pipe, timed out');
 561     end;
 562
 563     Sleep(INTERVAL)
 564   end;
 565
 566   // open that thingy
 567   hPipe := CreateFile( PChar( FPipeName),
 568                        GENERIC_READ or GENERIC_WRITE,
 569                        FShareMode,        // sharing
 570                        FSecurityAttribs,  // security attributes
 571                        OPEN_EXISTING,     // opens existing pipe
 572                        FILE_FLAG_OVERLAPPED or FILE_FLAG_WRITE_THROUGH, // async+fast, please
 573                        0);                // no template file
 574
 575   if hPipe = INVALID_HANDLE_VALUE
 576   then raise TTransportExceptionNotOpen.Create('Unable to open pipe, '+SysErrorMessage(GetLastError));
 577
 578   // everything fine
 579   FPipe := hPipe;
 580 end;
 581
 582
 583 { THandlePipeStreamImpl }
 584
 585
 586 constructor THandlePipeStreamImpl.Create( const aPipeHandle : THandle;
 587                                           const aOwnsHandle, aEnableOverlapped : Boolean;
 588                                           const aTimeOut : DWORD);
 589 begin
 590   inherited Create( aEnableOverlapped, aTimeOut);
 591
 592   if aOwnsHandle
 593   then FSrcHandle := aPipeHandle
 594   else FSrcHandle := DuplicatePipeHandle( aPipeHandle);
 595
 596   Open;
 597 end;
 598
 599
 600 destructor THandlePipeStreamImpl.Destroy;
 601 begin
 602   try
 603     ClosePipeHandle( FSrcHandle);
 604   finally
 605     inherited Destroy;
 606   end;
 607 end;
 608
 609
 610 procedure THandlePipeStreamImpl.Open;
 611 begin
 612   if not IsOpen
 613   then FPipe := DuplicatePipeHandle( FSrcHandle);
 614 end;
 615
 616
 617 { TPipeTransportBase }
 618
 619
 620 function TPipeTransportBase.GetIsOpen: Boolean;
 621 begin
 622   result := (FInputStream <> nil)  and (FInputStream.IsOpen)
 623         and (FOutputStream <> nil) and (FOutputStream.IsOpen);
 624 end;
 625
 626
 627 procedure TPipeTransportBase.Open;
 628 begin
 629   FInputStream.Open;
 630   FOutputStream.Open;
 631 end;
 632
 633
 634 procedure TPipeTransportBase.Close;
 635 begin
 636   FInputStream.Close;
 637   FOutputStream.Close;
 638 end;
 639
 640
 641 { TNamedPipeTransportClientEndImpl }
 642
 643
 644 constructor TNamedPipeTransportClientEndImpl.Create( const aPipeName : string; const aShareMode: DWORD;
 645                                    const aSecurityAttributes: PSecurityAttributes;
 646                                    const aTimeOut, aOpenTimeOut : DWORD);
 647 // Named pipe constructor
 648 begin
 649   inherited Create( nil, nil);
 650   FInputStream  := TNamedPipeStreamImpl.Create( aPipeName, TRUE, aShareMode, aSecurityAttributes, aTimeOut, aOpenTimeOut);
 651   FOutputStream := FInputStream;  // true for named pipes
 652 end;
 653
 654
 655 constructor TNamedPipeTransportClientEndImpl.Create( aPipe : THandle; aOwnsHandle : Boolean;
 656                                                      const aTimeOut : DWORD);
 657 // Named pipe constructor
 658 begin
 659   inherited Create( nil, nil);
 660   FInputStream  := THandlePipeStreamImpl.Create( aPipe, TRUE, aOwnsHandle, aTimeOut);
 661   FOutputStream := FInputStream;  // true for named pipes
 662 end;
 663
 664
 665 { TNamedPipeTransportServerEndImpl }
 666
 667
 668 constructor TNamedPipeTransportServerEndImpl.Create( aPipe : THandle; aOwnsHandle : Boolean;
 669                                                      const aTimeOut : DWORD);
 670 // Named pipe constructor
 671 begin
 672   FHandle := DuplicatePipeHandle( aPipe);
 673   inherited Create( aPipe, aOwnsHandle, aTimeOut);
 674 end;
 675
 676
 677 procedure TNamedPipeTransportServerEndImpl.Close;
 678 begin
 679   FlushFileBuffers( FHandle);
 680   DisconnectNamedPipe( FHandle);  // force client off the pipe
 681   ClosePipeHandle( FHandle);
 682
 683   inherited Close;
 684 end;
 685
 686
 687 { TAnonymousPipeTransportImpl }
 688
 689
 690 constructor TAnonymousPipeTransportImpl.Create( const aPipeRead, aPipeWrite : THandle;
 691                                                 aOwnsHandles : Boolean;
 692                                                 const aTimeOut : DWORD = DEFAULT_THRIFT_TIMEOUT);
 693 // Anonymous pipe constructor
 694 begin
 695   inherited Create( nil, nil);
 696   // overlapped is not supported with AnonPipes, see MSDN
 697   FInputStream  := THandlePipeStreamImpl.Create( aPipeRead, aOwnsHandles, FALSE, aTimeOut);
 698   FOutputStream := THandlePipeStreamImpl.Create( aPipeWrite, aOwnsHandles, FALSE, aTimeOut);
 699 end;
 700
 701
 702 { TPipeServerTransportBase }
 703
 704
 705 constructor TPipeServerTransportBase.Create;
 706 begin
 707   inherited Create;
 708   FStopServer := TEvent.Create(nil,TRUE,FALSE,'');  // manual reset
 709 end;
 710
 711
 712 destructor TPipeServerTransportBase.Destroy;
 713 begin
 714   try
 715     FreeAndNil( FStopServer);
 716   finally
 717     inherited Destroy;
 718   end;
 719 end;
 720
 721
 722 function TPipeServerTransportBase.QueryStopServer : Boolean;
 723 begin
 724   result := (FStopServer = nil)
 725          or (FStopServer.WaitFor(0) <> wrTimeout);
 726 end;
 727
 728
 729 procedure TPipeServerTransportBase.Listen;
 730 begin
 731   FStopServer.ResetEvent;
 732 end;
 733
 734
 735 procedure TPipeServerTransportBase.Close;
 736 begin
 737   FStopServer.SetEvent;
 738   InternalClose;
 739 end;
 740
 741
 742 { TAnonymousPipeServerTransportImpl }
 743
 744
 745 constructor TAnonymousPipeServerTransportImpl.Create(aBufsize : Cardinal; aTimeOut : DWORD);
 746 // Anonymous pipe CTOR
 747 begin
 748   inherited Create;
 749   FBufsize  := aBufSize;
 750   FReadHandle := INVALID_HANDLE_VALUE;
 751   FWriteHandle := INVALID_HANDLE_VALUE;
 752   FClientAnonRead := INVALID_HANDLE_VALUE;
 753   FClientAnonWrite := INVALID_HANDLE_VALUE;
 754   FTimeOut := aTimeOut;
 755
 756   // The anonymous pipe needs to be created first so that the server can
 757   // pass the handles on to the client before the serve (acceptImpl)
 758   // blocking call.
 759   if not CreateAnonPipe
 760   then raise TTransportExceptionNotOpen.Create(ClassName+'.Create() failed');
 761 end;
 762
 763
 764 function TAnonymousPipeServerTransportImpl.Accept(const fnAccepting: TProc): ITransport;
 765 var buf    : Byte;
 766     br     : DWORD;
 767 begin
 768   if Assigned(fnAccepting)
 769   then fnAccepting();
 770
 771   // This 0-byte read serves merely as a blocking call.
 772   if not ReadFile( FReadHandle, buf, 0, br, nil)
 773   and (GetLastError() <> ERROR_MORE_DATA)
 774   then raise TTransportExceptionNotOpen.Create('TServerPipe unable to initiate pipe communication');
 775
 776   // create the transport impl
 777   result := TAnonymousPipeTransportImpl.Create( FReadHandle, FWriteHandle, FALSE, FTimeOut);
 778 end;
 779
 780
 781 procedure TAnonymousPipeServerTransportImpl.InternalClose;
 782 begin
 783   ClosePipeHandle( FReadHandle);
 784   ClosePipeHandle( FWriteHandle);
 785   ClosePipeHandle( FClientAnonRead);
 786   ClosePipeHandle( FClientAnonWrite);
 787 end;
 788
 789
 790 function TAnonymousPipeServerTransportImpl.ReadHandle : THandle;
 791 begin
 792   result := FReadHandle;
 793 end;
 794
 795
 796 function TAnonymousPipeServerTransportImpl.WriteHandle : THandle;
 797 begin
 798   result := FWriteHandle;
 799 end;
 800
 801
 802 function TAnonymousPipeServerTransportImpl.ClientAnonRead : THandle;
 803 begin
 804   result := FClientAnonRead;
 805 end;
 806
 807
 808 function TAnonymousPipeServerTransportImpl.ClientAnonWrite  : THandle;
 809 begin
 810   result := FClientAnonWrite;
 811 end;
 812
 813
 814 function TAnonymousPipeServerTransportImpl.CreateAnonPipe : Boolean;
 815 var sd           : PSECURITY_DESCRIPTOR;
 816     sa           : SECURITY_ATTRIBUTES; //TSecurityAttributes;
 817     hCAR, hPipeW, hCAW, hPipe : THandle;
 818 begin
 819   sd := PSECURITY_DESCRIPTOR( LocalAlloc( LPTR,SECURITY_DESCRIPTOR_MIN_LENGTH));
 820   try
 821     Win32Check( InitializeSecurityDescriptor( sd, SECURITY_DESCRIPTOR_REVISION));
 822     Win32Check( SetSecurityDescriptorDacl( sd, TRUE, nil, FALSE));
 823
 824     sa.nLength := sizeof( sa);
 825     sa.lpSecurityDescriptor := sd;
 826     sa.bInheritHandle       := TRUE; //allow passing handle to child
 827
 828     Result := CreatePipe( hCAR, hPipeW, @sa, FBufSize); //create stdin pipe
 829     if not Result then begin   //create stdin pipe
 830       raise TTransportExceptionNotOpen.Create('TServerPipe CreatePipe (anon) failed, '+SysErrorMessage(GetLastError));
 831       Exit;
 832     end;
 833
 834     Result := CreatePipe( hPipe, hCAW, @sa, FBufSize); //create stdout pipe
 835     if not Result then begin  //create stdout pipe
 836       CloseHandle( hCAR);
 837       CloseHandle( hPipeW);
 838       raise TTransportExceptionNotOpen.Create('TServerPipe CreatePipe (anon) failed, '+SysErrorMessage(GetLastError));
 839       Exit;
 840     end;
 841
 842     FClientAnonRead  := hCAR;
 843     FClientAnonWrite := hCAW;
 844     FReadHandle      := hPipe;
 845     FWriteHandle     := hPipeW;
 846   finally
 847     if sd <> nil then LocalFree( Cardinal(sd));
 848   end;
 849 end;
 850
 851
 852 { TNamedPipeServerTransportImpl }
 853
 854
 855 constructor TNamedPipeServerTransportImpl.Create( aPipename : string; aBufsize, aMaxConns, aTimeOut : Cardinal);
 856 // Named Pipe CTOR
 857 begin
 858   inherited Create;
 859   ASSERT( aTimeout > 0);
 860   FPipeName  := aPipename;
 861   FBufsize   := aBufSize;
 862   FMaxConns  := Max( 1, Min( PIPE_UNLIMITED_INSTANCES, aMaxConns));
 863   FHandle    := INVALID_HANDLE_VALUE;
 864   FTimeout   := aTimeOut;
 865   FConnected := FALSE;
 866
 867   if Copy(FPipeName,1,2) <> '\\'
 868   then FPipeName := '\\.\pipe\' + FPipeName;  // assume localhost
 869 end;
 870
 871
 872 function TNamedPipeServerTransportImpl.Accept(const fnAccepting: TProc): ITransport;
 873 var dwError, dwWait, dwDummy : DWORD;
 874     overlapped : IOverlappedHelper;
 875     handles : array[0..1] of THandle;
 876 begin
 877   overlapped := TOverlappedHelperImpl.Create;
 878
 879   ASSERT( not FConnected);
 880   CreateNamedPipe;
 881   while not FConnected do begin
 882
 883     if QueryStopServer then begin
 884       InternalClose;
 885       Abort;
 886     end;
 887
 888     if Assigned(fnAccepting)
 889     then fnAccepting();
 890
 891     // Wait for the client to connect; if it succeeds, the
 892     // function returns a nonzero value. If the function returns
 893     // zero, GetLastError should return ERROR_PIPE_CONNECTED.
 894     if ConnectNamedPipe( Handle, overlapped.OverlappedPtr) then begin
 895       FConnected := TRUE;
 896       Break;
 897     end;
 898
 899     // ConnectNamedPipe() returns FALSE for OverlappedIO, even if connected.
 900     // We have to check GetLastError() explicitly to find out
 901     dwError := GetLastError;
 902     case dwError of
 903       ERROR_PIPE_CONNECTED : begin
 904         FConnected := not QueryStopServer;  // special case: pipe immediately connected
 905       end;
 906
 907       ERROR_IO_PENDING : begin
 908         handles[0] := overlapped.WaitHandle;
 909         handles[1] := FStopServer.Handle;
 910         dwWait := WaitForMultipleObjects( 2, @handles, FALSE, FTimeout);
 911         FConnected := (dwWait = WAIT_OBJECT_0)
 912                   and GetOverlappedResult( Handle, overlapped.Overlapped, dwDummy, TRUE)
 913                   and not QueryStopServer;
 914       end;
 915
 916     else
 917       InternalClose;
 918       raise TTransportExceptionNotOpen.Create('Client connection failed');
 919     end;
 920   end;
 921
 922   // create the transport impl
 923   result := CreateTransportInstance;
 924 end;
 925
 926
 927 function TNamedPipeServerTransportImpl.CreateTransportInstance : ITransport;
 928 // create the transport impl
 929 var hPipe : THandle;
 930 begin
 931   hPipe := THandle( InterlockedExchangePointer( Pointer(FHandle), Pointer(INVALID_HANDLE_VALUE)));
 932   try
 933     FConnected := FALSE;
 934     result := TNamedPipeTransportServerEndImpl.Create( hPipe, TRUE, FTimeout);
 935   except
 936     ClosePipeHandle(hPipe);
 937     raise;
 938   end;
 939 end;
 940
 941
 942 procedure TNamedPipeServerTransportImpl.InternalClose;
 943 var hPipe : THandle;
 944 begin
 945   hPipe := THandle( InterlockedExchangePointer( Pointer(FHandle), Pointer(INVALID_HANDLE_VALUE)));
 946   if hPipe = INVALID_HANDLE_VALUE then Exit;
 947
 948   try
 949     if FConnected
 950     then FlushFileBuffers( hPipe)
 951     else CancelIo( hPipe);
 952     DisconnectNamedPipe( hPipe);
 953   finally
 954     ClosePipeHandle( hPipe);
 955     FConnected := FALSE;
 956   end;
 957 end;
 958
 959
 960 function TNamedPipeServerTransportImpl.Handle : THandle;
 961 begin
 962   {$IFDEF WIN64}
 963   result := THandle( InterlockedExchangeAdd64( Int64(FHandle), 0));
 964   {$ELSE}
 965   result := THandle( InterlockedExchangeAdd( Integer(FHandle), 0));
 966   {$ENDIF}
 967 end;
 968
 969
 970 function TNamedPipeServerTransportImpl.CreateNamedPipe : THandle;
 971 var SIDAuthWorld : SID_IDENTIFIER_AUTHORITY ;
 972     everyone_sid : PSID;
 973     ea           : EXPLICIT_ACCESS;
 974     acl          : PACL;
 975     sd           : PSECURITY_DESCRIPTOR;
 976     sa           : SECURITY_ATTRIBUTES;
 977 const
 978   SECURITY_WORLD_SID_AUTHORITY  : TSIDIdentifierAuthority = (Value : (0,0,0,0,0,1));
 979   SECURITY_WORLD_RID = $00000000;
 980 begin
 981   sd := nil;
 982   everyone_sid := nil;
 983   try
 984     ASSERT( (FHandle = INVALID_HANDLE_VALUE) and not FConnected);
 985
 986     // Windows - set security to allow non-elevated apps
 987     // to access pipes created by elevated apps.
 988     SIDAuthWorld := SECURITY_WORLD_SID_AUTHORITY;
 989     AllocateAndInitializeSid( SIDAuthWorld, 1, SECURITY_WORLD_RID, 0, 0, 0, 0, 0, 0, 0, everyone_sid);
 990
 991     ZeroMemory( @ea, SizeOf(ea));
 992     ea.grfAccessPermissions := GENERIC_ALL; //SPECIFIC_RIGHTS_ALL or STANDARD_RIGHTS_ALL;
 993     ea.grfAccessMode        := SET_ACCESS;
 994     ea.grfInheritance       := NO_INHERITANCE;
 995     ea.Trustee.TrusteeForm  := TRUSTEE_IS_SID;
 996     ea.Trustee.TrusteeType  := TRUSTEE_IS_WELL_KNOWN_GROUP;
 997     ea.Trustee.ptstrName    := PChar(everyone_sid);
 998
 999     acl := nil;
1000     SetEntriesInAcl( 1, @ea, nil, acl);
1001
1002     sd := PSECURITY_DESCRIPTOR( LocalAlloc( LPTR,SECURITY_DESCRIPTOR_MIN_LENGTH));
1003     Win32Check( InitializeSecurityDescriptor( sd, SECURITY_DESCRIPTOR_REVISION));
1004     Win32Check( SetSecurityDescriptorDacl( sd, TRUE, acl, FALSE));
1005
1006     sa.nLength := SizeOf(sa);
1007     sa.lpSecurityDescriptor := sd;
1008     sa.bInheritHandle       := FALSE;
1009
1010     // Create an instance of the named pipe
1011     {$IFDEF OLD_UNIT_NAMES}
1012     result := Windows.CreateNamedPipe(
1013     {$ELSE}
1014     result := Winapi.Windows.CreateNamedPipe(
1015     {$ENDIF}
1016         PChar( FPipeName),        // pipe name
1017         PIPE_ACCESS_DUPLEX or     // read/write access
1018         FILE_FLAG_OVERLAPPED,     // async mode
1019         PIPE_TYPE_BYTE or         // byte type pipe
1020         PIPE_READMODE_BYTE,       // byte read mode
1021         FMaxConns,                // max. instances
1022         FBufSize,                 // output buffer size
1023         FBufSize,                 // input buffer size
1024         FTimeout,                 // time-out, see MSDN
1025         @sa                       // default security attribute
1026     );
1027
1028     if( result <> INVALID_HANDLE_VALUE)
1029     then InterlockedExchangePointer( Pointer(FHandle), Pointer(result))
1030     else raise TTransportExceptionNotOpen.Create('CreateNamedPipe() failed ' + IntToStr(GetLastError));
1031
1032   finally
1033     if sd <> nil then LocalFree( Cardinal( sd));
1034     if acl <> nil then LocalFree( Cardinal( acl));
1035     if everyone_sid <> nil then FreeSid(everyone_sid);
1036   end;
1037 end;
1038
1039
1040
1041 end.
1042
1043
1044