gorealis v2 refactor (#5)

* Changing default timeout for start maintenance.

* Upgrading dependencies to gorealis v2 and thrift  0.12.0

* Refactored to update to gorealis v2.

vendor/git.apache.org/thrift.git/lib/d/test/Makefile.am

@@ -24,8 +24,6 @@ BUILT_SOURCES = trusted-ca-certificate.pem server-certificate.pem
 
 # Thrift compiler rules
 
-THRIFT = $(top_builddir)/compiler/cpp/thrift
-
 debug_proto_gen = $(addprefix gen-d/, DebugProtoTest_types.d)
 
 $(debug_proto_gen): $(top_srcdir)/test/DebugProtoTest.thrift

vendor/git.apache.org/thrift.git/lib/d/test/client_pool_test.d

@@ -18,6 +18,7 @@
  */
 module client_pool_test;
 
+import core.sync.semaphore : Semaphore;
 import core.time : Duration, dur;
 import core.thread : Thread;
 import std.algorithm;
@@ -28,6 +29,7 @@ import std.getopt;
 import std.range;
 import std.stdio;
 import std.typecons;
+import std.variant : Variant;
 import thrift.base;
 import thrift.async.libevent;
 import thrift.async.socket;
@@ -37,9 +39,12 @@ import thrift.codegen.async_client_pool;
 import thrift.codegen.client;
 import thrift.codegen.client_pool;
 import thrift.codegen.processor;
+import thrift.protocol.base;
 import thrift.protocol.binary;
+import thrift.server.base;
 import thrift.server.simple;
 import thrift.server.transport.socket;
+import thrift.transport.base;
 import thrift.transport.buffered;
 import thrift.transport.socket;
 import thrift.util.cancellation;
@@ -108,11 +113,29 @@ private:
   }
 }
 
+class ServerPreServeHandler : TServerEventHandler {
+  this(Semaphore sem) {
+    sem_ = sem;
+  }
+
+  override void preServe() {
+    sem_.notify();
+  }
+
+  Variant createContext(TProtocol input, TProtocol output) { return Variant.init; }
+  void deleteContext(Variant serverContext, TProtocol input, TProtocol output) {}
+  void preProcess(Variant serverContext, TTransport transport) {}
+
+private:
+  Semaphore sem_;
+}
+
 class ServerThread : Thread {
-  this(ExTestHandler handler, TCancellation cancellation) {
+  this(ExTestHandler handler, ServerPreServeHandler serverHandler, TCancellation cancellation) {
     super(&run);
     handler_ = handler;
     cancellation_ = cancellation;
+    serverHandler_ = serverHandler;
   }
 private:
   void run() {
@@ -123,16 +146,17 @@ private:
       serverTransport.recvTimeout = dur!"seconds"(3);
       auto transportFactory = new TBufferedTransportFactory;
 
-      auto server = new TSimpleServer(
-        processor, serverTransport, transportFactory, protocolFactory);
+      auto server = new TSimpleServer(processor, serverTransport, transportFactory, protocolFactory);
+      server.eventHandler = serverHandler_;
       server.serve(cancellation_);
     } catch (Exception e) {
       writefln("Server thread on port %s failed: %s", handler_.port, e);
     }
   }
 
-  TCancellation cancellation_;
   ExTestHandler handler_;
+  ServerPreServeHandler serverHandler_;
+  TCancellation cancellation_;
 }
 
 void main(string[] args) {
@@ -145,6 +169,9 @@ void main(string[] args) {
 
   immutable ports = cast(immutable)array(map!"cast(ushort)a"(iota(port, port + 6)));
 
+  // semaphore that will be incremented whenever each server thread has bound and started listening
+  Semaphore sem = new Semaphore(0);
+
 version (none) {
   // Cannot use this due to multiple DMD @@BUG@@s:
   // 1. »function D main is a nested function and cannot be accessed from array«
@@ -174,11 +201,10 @@ version (none) {
 }
 
   // Fire up the server threads.
-  foreach (h; handlers) (new ServerThread(h, serverCancellation)).start();
+  foreach (h; handlers) (new ServerThread(h, new ServerPreServeHandler(sem), serverCancellation)).start();
 
-  // Give the servers some time to get up. This should really be accomplished
-  // via a barrier here and in the preServe() hook.
-  Thread.sleep(dur!"msecs"(10));
+  // wait until all the handlers signal that they're ready to serve
+  foreach (h; handlers) (sem.wait(dur!`seconds`(1)));
 
   syncClientPoolTest(ports, handlers);
   asyncClientPoolTest(ports, handlers);
@@ -409,8 +435,8 @@ void asyncAggregatorTest(const(ushort)[] ports, ExTestHandler[] handlers) {
     )();
     Thread.sleep(dur!"msecs"(20));
     auto resultTuple = partialResult.finishGet();
-    enforce(resultTuple._0 == ports[0 .. 2]);
-    enforce(equal(map!"a.port"(cast(TestServiceException[])resultTuple._1),
+    enforce(resultTuple[0] == ports[0 .. 2]);
+    enforce(equal(map!"a.port"(cast(TestServiceException[])resultTuple[1]),
       ports[3 .. $ - 1]));
   }
 }

vendor/git.apache.org/thrift.git/lib/d/test/serialization_benchmark.d

@@ -13,7 +13,7 @@
  */
 module serialization_benchmark;
 
-import std.datetime : AutoStart, StopWatch;
+import std.datetime.stopwatch : AutoStart, StopWatch;
 import std.math : PI;
 import std.stdio;
 import thrift.protocol.binary;
@@ -46,7 +46,7 @@ void main() {
     }
     sw.stop();
 
-    auto msecs = sw.peek().msecs;
+    auto msecs = sw.peek().total!"msecs";
     writefln("Write: %s ms (%s kHz)", msecs, ITERATIONS / msecs);
   }
 
@@ -64,7 +64,7 @@ void main() {
     }
     sw.stop();
 
-    auto msecs = sw.peek().msecs;
+    auto msecs = sw.peek().total!"msecs";
     writefln(" Read: %s ms (%s kHz)", msecs, ITERATIONS / msecs);
   }
 }

vendor/git.apache.org/thrift.git/lib/d/test/thrift_test_server.d

@@ -16,8 +16,11 @@
  * specific language governing permissions and limitations
  * under the License.
  */
+
 module thrift_test_server;
 
+import core.stdc.errno : errno;
+import core.stdc.signal : signal, sigfn_t, SIGINT, SIG_DFL, SIG_ERR;
 import core.thread : dur, Thread;
 import std.algorithm;
 import std.exception : enforce;
@@ -40,6 +43,7 @@ import thrift.transport.buffered;
 import thrift.transport.framed;
 import thrift.transport.http;
 import thrift.transport.ssl;
+import thrift.util.cancellation;
 import thrift.util.hashset;
 import test_utils;
 
@@ -205,14 +209,44 @@ private:
   bool trace_;
 }
 
+shared(bool) gShutdown = false;
+
+nothrow @nogc extern(C) void handleSignal(int sig) {
+  gShutdown = true;
+}
+
+// Runs a thread that waits for shutdown to be
+// signaled and then triggers cancellation,
+// causing the server to stop.  While we could
+// use a signalfd for this purpose, we are instead
+// opting for a busy waiting scheme for maximum
+// portability since signalfd is a linux thing.
+
+class ShutdownThread : Thread {
+  this(TCancellationOrigin cancellation) {
+    cancellation_ = cancellation;
+    super(&run);
+  }
+  
+private:
+  void run() {
+    while (!gShutdown) {
+      Thread.sleep(dur!("msecs")(25));
+    }
+    cancellation_.trigger();
+  }
+  
+  TCancellationOrigin cancellation_;
+}
+
 void main(string[] args) {
   ushort port = 9090;
   ServerType serverType;
   ProtocolType protocolType;
   size_t numIOThreads = 1;
   TransportType transportType;
-  bool ssl;
-  bool trace;
+  bool ssl = false;
+  bool trace = true;
   size_t taskPoolSize = totalCPUs;
 
   getopt(args, "port", &port, "protocol", &protocolType, "server-type",
@@ -279,8 +313,26 @@ void main(string[] args) {
   auto server = createServer(serverType, numIOThreads, taskPoolSize,
     processor, serverSocket, transportFactory, protocolFactory);
 
+  // Set up SIGINT signal handling
+  sigfn_t oldHandler = signal(SIGINT, &handleSignal);
+  enforce(oldHandler != SIG_ERR,
+    "Could not replace the SIGINT signal handler: errno {0}".format(errno()));
+  
+  // Set up a server cancellation trigger
+  auto cancel = new TCancellationOrigin();
+
+  // Set up a listener for the shutdown condition - this will
+  // wake up when the signal occurs and trigger cancellation.
+  auto shutdown = new ShutdownThread(cancel);
+  shutdown.start();
+  
+  // Serve from this thread; the signal will stop the server
+  // and control will return here
   writefln("Starting %s/%s %s ThriftTest server %son port %s...", protocolType,
     transportType, serverType, ssl ? "(using SSL) ": "", port);
-  server.serve();
+  server.serve(cancel);
+  shutdown.join();
+  signal(SIGINT, SIG_DFL);
+    
   writeln("done.");
 }