THRIFT-1874 timeout for the server-side end of a named pipe

Patch: Jens Geyer

diff --git a/lib/delphi/src/Thrift.Transport.Pipes.pas b/lib/delphi/src/Thrift.Transport.Pipes.pas
index 66db240..54e00a4 100644 (file)
--- a/lib/delphi/src/Thrift.Transport.Pipes.pas
+++ b/lib/delphi/src/Thrift.Transport.Pipes.pas
@@ -191,6 +191,7 @@ type
     FPipeName     : string;
     FMaxConns     : DWORD;
     FBufSize      : DWORD;
+    FTimeout      : DWORD;
 
     FHandle : THandle;
 
@@ -203,7 +204,8 @@ type
 
   public
     constructor Create( aPipename : string; aBufsize : Cardinal = 4096;
-                        aMaxConns : Cardinal = PIPE_UNLIMITED_INSTANCES);
+                        aMaxConns : Cardinal = PIPE_UNLIMITED_INSTANCES;
+                        aTimeOut : Cardinal = 0);
 
     procedure Close; override;
   end;
@@ -439,7 +441,8 @@ end;
 
 function TPipeTransportBaseImpl.GetIsOpen: Boolean;
 begin
-  result := (FInputStream <> nil);
+  result := (FInputStream <> nil)  and (FInputStream.IsOpen)
+        and (FOutputStream <> nil) and (FOutputStream.IsOpen);
 end;
 
 
@@ -634,7 +637,7 @@ end;
 { TNamedServerPipeImpl }
 
 
-constructor TNamedServerPipeImpl.Create( aPipename : string; aBufsize, aMaxConns : Cardinal);
+constructor TNamedServerPipeImpl.Create( aPipename : string; aBufsize, aMaxConns, aTimeOut : Cardinal);
 // Named Pipe CTOR
 begin
   inherited Create;
@@ -642,6 +645,7 @@ begin
   FBufsize  := aBufSize;
   FMaxConns := Max( 1, Min( PIPE_UNLIMITED_INSTANCES, aMaxConns));
   FHandle   := INVALID_HANDLE_VALUE;
+  FTimeout  := aTimeOut;
 
   if Copy(FPipeName,1,2) <> '\\'
   then FPipeName := '\\.\pipe\' + FPipeName;  // assume localhost
@@ -735,7 +739,7 @@ begin
                                       FMaxConns,                // max. instances
                                       FBufSize,                 // output buffer size
                                       FBufSize,                 // input buffer size
-                                      0,                        // client time-out
+                                      FTimeout,                 // time-out, see MSDN
                                       @sa);                     // default security attribute
 
     FHandle := hPipe;

diff --git a/lib/delphi/test/TestClient.pas b/lib/delphi/test/TestClient.pas
index e72775e..2f77de8 100644 (file)
--- a/lib/delphi/test/TestClient.pas
+++ b/lib/delphi/test/TestClient.pas
@@ -116,6 +116,11 @@ var
   streamtrans : IStreamTransport;
   http : IHTTPClient;
   protType, p : TKnownProtocol;
+const
+  // pipe timeouts to be used
+  DEBUG_TIMEOUT   = 30 * 1000;
+  RELEASE_TIMEOUT = DEFAULT_THRIFT_PIPE_TIMEOUT;
+  TIMEOUT         = RELEASE_TIMEOUT;
 begin
   bBuffered := False;;
   bFramed := False;
@@ -233,7 +238,7 @@ begin
       begin
         if sPipeName <> '' then begin
           Console.WriteLine('Using named pipe ('+sPipeName+')');
-          streamtrans := TNamedPipeImpl.Create( sPipeName);
+          streamtrans := TNamedPipeImpl.Create( sPipeName, 0, nil, TIMEOUT);
         end
         else if bAnonPipe then begin
           Console.WriteLine('Using anonymous pipes ('+IntToStr(Integer(hAnonRead))+' and '+IntToStr(Integer(hAnonWrite))+')');
@@ -247,7 +252,7 @@ begin
         trans := streamtrans;
 
         if bBuffered then begin
-          trans := TBufferedTransportImpl.Create( streamtrans);
+          trans := TBufferedTransportImpl.Create( streamtrans, 32);  // small buffer to test read()
           Console.WriteLine('Using buffered transport');
         end;
 
@@ -265,11 +270,11 @@ begin
 
       // create protocol instance, default to BinaryProtocol
       case protType of
-        prot_Binary:  prot := TBinaryProtocolImpl.Create( trans);
+        prot_Binary:  prot := TBinaryProtocolImpl.Create( trans, BINARY_STRICT_READ, BINARY_STRICT_WRITE);
         prot_JSON  :  prot := TJSONProtocolImpl.Create( trans);
       else
         ASSERT( FALSE);  // unhandled case!
-        prot := TBinaryProtocolImpl.Create( trans);  // use default
+        prot := TBinaryProtocolImpl.Create( trans, BINARY_STRICT_READ, BINARY_STRICT_WRITE);  // use default
       end;
 
       thread := TClientThread.Create( trans, prot, FNumIteration);

diff --git a/lib/delphi/test/TestConstants.pas b/lib/delphi/test/TestConstants.pas
index 9cb85ab..b6664ef 100644 (file)
--- a/lib/delphi/test/TestConstants.pas
+++ b/lib/delphi/test/TestConstants.pas
@@ -29,6 +29,10 @@ const
   KNOWN_PROTOCOLS : array[TKnownProtocol] of string
                   = ('binary', 'JSON');
 
+  // defaults are: read=false, write=true
+  BINARY_STRICT_READ  = FALSE;
+  BINARY_STRICT_WRITE = FALSE;
+
 implementation
 
 // nothing

diff --git a/lib/delphi/test/TestServer.pas b/lib/delphi/test/TestServer.pas
index 7048a20..791468b 100644 (file)
--- a/lib/delphi/test/TestServer.pas
+++ b/lib/delphi/test/TestServer.pas
@@ -464,6 +464,11 @@ var
   i : Integer;
   s : string;
   protType, p : TKnownProtocol;
+const
+  // pipe timeouts to be used
+  DEBUG_TIMEOUT   = 30 * 1000;
+  RELEASE_TIMEOUT = 0;  // server-side default
+  TIMEOUT         = RELEASE_TIMEOUT;
 begin
   try
     UseBufferedSockets := False;
@@ -525,11 +530,11 @@ begin
 
     // create protocol factory, default to BinaryProtocol
     case protType of
-      prot_Binary:  ProtocolFactory := TBinaryProtocolImpl.TFactory.Create;
+      prot_Binary:  ProtocolFactory := TBinaryProtocolImpl.TFactory.Create( BINARY_STRICT_READ, BINARY_STRICT_WRITE);
       prot_JSON  :  ProtocolFactory := TJSONProtocolImpl.TFactory.Create;
     else
       ASSERT( FALSE);  // unhandled case!
-      ProtocolFactory := TBinaryProtocolImpl.TFactory.Create;
+      ProtocolFactory := TBinaryProtocolImpl.TFactory.Create( BINARY_STRICT_READ, BINARY_STRICT_WRITE);
     end;
     ASSERT( ProtocolFactory <> nil);
     Console.WriteLine('- '+KNOWN_PROTOCOLS[protType]+' protocol');
@@ -537,7 +542,7 @@ begin
 
     if sPipeName <> '' then begin
       Console.WriteLine('- named pipe ('+sPipeName+')');
-      namedpipe   := TNamedServerPipeImpl.Create( sPipeName);
+      namedpipe   := TNamedServerPipeImpl.Create( sPipeName, 4096, PIPE_UNLIMITED_INSTANCES, TIMEOUT);
       servertrans := namedpipe;
     end
     else if AnonPipe then begin

diff --git a/lib/delphi/test/server.dpr b/lib/delphi/test/server.dpr
index a5568c4..5fad6eb 100644 (file)
--- a/lib/delphi/test/server.dpr
+++ b/lib/delphi/test/server.dpr
@@ -24,7 +24,7 @@ program server;
 uses
   SysUtils,
   TestServer in 'TestServer.pas',
-  Thrift.Test in 'gen-delphi\Thrift.Test.pas',
+  Thrift.Test,  // in gen-delphi folder
   Thrift in '..\src\Thrift.pas',
   Thrift.Transport in '..\src\Thrift.Transport.pas',
   Thrift.Transport.Pipes in '..\src\Thrift.Transport.Pipes.pas',