1 files changed, 803 insertions, 0 deletions
diff --git a/src/jaegertracing/thrift/lib/d/test/transport_test.d b/src/jaegertracing/thrift/lib/d/test/transport_test.d
new file mode 100644
index 000000000..623e03f0e
--- /dev/null
+++ b/src/jaegertracing/thrift/lib/d/test/transport_test.d
@@ -0,0 +1,803 @@
+/*
+ * 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.
+ */
+
+/**
+ * Exercises various transports, combined with the buffered/framed wrappers.
+ *
+ * Originally ported from the C++ version, with Windows support code added.
+ */
+module transport_test;
+
+import core.atomic;
+import core.time : Duration;
+import core.thread : Thread;
+import std.conv : to;
+import std.datetime;
+import std.exception : enforce;
+static import std.file;
+import std.getopt;
+import std.random : rndGen, uniform, unpredictableSeed;
+import std.socket;
+import std.stdio;
+import std.string;
+import std.typetuple;
+import thrift.transport.base;
+import thrift.transport.buffered;
+import thrift.transport.framed;
+import thrift.transport.file;
+import thrift.transport.http;
+import thrift.transport.memory;
+import thrift.transport.socket;
+import thrift.transport.zlib;
+
+/*
+ * Size generation helpers – used to be able to run the same testing code
+ * with both constant and random total/chunk sizes.
+ */
+
+interface SizeGenerator {
+  size_t nextSize();
+  string toString();
+}
+
+class ConstantSizeGenerator : SizeGenerator {
+  this(size_t value) {
+    value_ = value;
+  }
+
+  override size_t nextSize() {
+    return value_;
+  }
+
+  override string toString() const {
+    return to!string(value_);
+  }
+
+private:
+  size_t value_;
+}
+
+class RandomSizeGenerator : SizeGenerator {
+  this(size_t min, size_t max) {
+    min_ = min;
+    max_ = max;
+  }
+
+  override size_t nextSize() {
+    return uniform!"[]"(min_, max_);
+  }
+
+  override string toString() const {
+    return format("rand(%s, %s)", min_, max_);
+  }
+
+  size_t min() const @property {
+    return min_;
+  }
+
+  size_t max() const @property {
+    return max_;
+  }
+
+private:
+  size_t min_;
+  size_t max_;
+}
+
+
+/*
+ * Classes to set up coupled transports
+ */
+
+/**
+ * Helper class to represent a coupled pair of transports.
+ *
+ * Data written to the output transport can be read from the input transport.
+ *
+ * This is used as the base class for the various coupled transport
+ * implementations. It shouldn't be used directly.
+ */
+class CoupledTransports(Transport) if (isTTransport!Transport) {
+  Transport input;
+  Transport output;
+}
+
+template isCoupledTransports(T) {
+  static if (is(T _ : CoupledTransports!U, U)) {
+    enum isCoupledTransports = true;
+  } else {
+    enum isCoupledTransports = false;
+  }
+}
+
+/**
+ * Helper template class for creating coupled transports that wrap
+ * another transport.
+ */
+class CoupledWrapperTransports(WrapperTransport, InnerCoupledTransports) if (
+  isTTransport!WrapperTransport && isCoupledTransports!InnerCoupledTransports
+) : CoupledTransports!WrapperTransport {
+  this() {
+    inner_ = new InnerCoupledTransports();
+    if (inner_.input) {
+      input = new WrapperTransport(inner_.input);
+    }
+    if (inner_.output) {
+      output = new WrapperTransport(inner_.output);
+    }
+  }
+
+  ~this() {
+    destroy(inner_);
+  }
+
+private:
+  InnerCoupledTransports inner_;
+}
+
+import thrift.internal.codegen : PApply;
+alias PApply!(CoupledWrapperTransports, TBufferedTransport) CoupledBufferedTransports;
+alias PApply!(CoupledWrapperTransports, TFramedTransport) CoupledFramedTransports;
+alias PApply!(CoupledWrapperTransports, TZlibTransport) CoupledZlibTransports;
+
+/**
+ * Coupled TMemoryBuffers.
+ */
+class CoupledMemoryBuffers : CoupledTransports!TMemoryBuffer {
+  this() {
+    buf = new TMemoryBuffer;
+    input = buf;
+    output = buf;
+  }
+
+  TMemoryBuffer buf;
+}
+
+/**
+ * Coupled TSockets.
+ */
+class CoupledSocketTransports : CoupledTransports!TSocket {
+  this() {
+    auto sockets = socketPair();
+    input = new TSocket(sockets[0]);
+    output = new TSocket(sockets[1]);
+  }
+
+  ~this() {
+    input.close();
+    output.close();
+  }
+}
+
+/**
+ * Coupled TFileTransports
+ */
+class CoupledFileTransports : CoupledTransports!TTransport {
+  this() {
+    // We actually need the file name of the temp file here, so we can't just
+    // use the usual tempfile facilities.
+    do {
+      fileName_ = tmpDir ~ "/thrift.transport_test." ~ to!string(rndGen().front);
+      rndGen().popFront();
+    } while (std.file.exists(fileName_));
+
+    writefln("Using temp file: %s", fileName_);
+
+    auto writer = new TFileWriterTransport(fileName_);
+    writer.open();
+    output = writer;
+
+    // Wait until the file has been created.
+    writer.flush();
+
+    auto reader = new TFileReaderTransport(fileName_);
+    reader.open();
+    reader.readTimeout(dur!"msecs"(-1));
+    input = reader;
+  }
+
+  ~this() {
+    input.close();
+    output.close();
+    std.file.remove(fileName_);
+  }
+
+  static string tmpDir;
+
+private:
+  string fileName_;
+}
+
+
+/*
+ * Test functions
+ */
+
+/**
+ * Test interleaved write and read calls.
+ *
+ * Generates a buffer totalSize bytes long, then writes it to the transport,
+ * and verifies the written data can be read back correctly.
+ *
+ * Mode of operation:
+ * - call wChunkGenerator to figure out how large of a chunk to write
+ *   - call wSizeGenerator to get the size for individual write() calls,
+ *     and do this repeatedly until the entire chunk is written.
+ * - call rChunkGenerator to figure out how large of a chunk to read
+ *   - call rSizeGenerator to get the size for individual read() calls,
+ *     and do this repeatedly until the entire chunk is read.
+ * - repeat until the full buffer is written and read back,
+ *   then compare the data read back against the original buffer
+ *
+ *
+ * - If any of the size generators return 0, this means to use the maximum
+ *   possible size.
+ *
+ * - If maxOutstanding is non-zero, write chunk sizes will be chosen such that
+ *   there are never more than maxOutstanding bytes waiting to be read back.
+ */
+void testReadWrite(CoupledTransports)(
+  size_t totalSize,
+  SizeGenerator wSizeGenerator,
+  SizeGenerator rSizeGenerator,
+  SizeGenerator wChunkGenerator,
+  SizeGenerator rChunkGenerator,
+  size_t maxOutstanding
+) if (
+  isCoupledTransports!CoupledTransports
+) {
+  scope transports = new CoupledTransports;
+  assert(transports.input);
+  assert(transports.output);
+
+  auto wbuf = new ubyte[totalSize];
+  auto rbuf = new ubyte[totalSize];
+
+  // Store some data in wbuf.
+  foreach (i, ref b; wbuf) {
+    b = i & 0xff;
+  }
+
+  size_t totalWritten;
+  size_t totalRead;
+  while (totalRead < totalSize) {
+    // Determine how large a chunk of data to write.
+    auto wChunkSize = wChunkGenerator.nextSize();
+    if (wChunkSize == 0 || wChunkSize > totalSize - totalWritten) {
+      wChunkSize = totalSize - totalWritten;
+    }
+
+    // Make sure (totalWritten - totalRead) + wChunkSize is less than
+    // maxOutstanding.
+    if (maxOutstanding > 0 &&
+        wChunkSize > maxOutstanding - (totalWritten - totalRead)) {
+      wChunkSize = maxOutstanding - (totalWritten - totalRead);
+    }
+
+    // Write the chunk.
+    size_t chunkWritten = 0;
+    while (chunkWritten < wChunkSize) {
+      auto writeSize = wSizeGenerator.nextSize();
+      if (writeSize == 0 || writeSize > wChunkSize - chunkWritten) {
+        writeSize = wChunkSize - chunkWritten;
+      }
+
+      transports.output.write(wbuf[totalWritten .. totalWritten + writeSize]);
+      chunkWritten += writeSize;
+      totalWritten += writeSize;
+    }
+
+    // Flush the data, so it will be available in the read transport
+    // Don't flush if wChunkSize is 0. (This should only happen if
+    // totalWritten == totalSize already, and we're only reading now.)
+    if (wChunkSize > 0) {
+      transports.output.flush();
+    }
+
+    // Determine how large a chunk of data to read back.
+    auto rChunkSize = rChunkGenerator.nextSize();
+    if (rChunkSize == 0 || rChunkSize > totalWritten - totalRead) {
+      rChunkSize = totalWritten - totalRead;
+    }
+
+    // Read the chunk.
+    size_t chunkRead;
+    while (chunkRead < rChunkSize) {
+      auto readSize = rSizeGenerator.nextSize();
+      if (readSize == 0 || readSize > rChunkSize - chunkRead) {
+        readSize = rChunkSize - chunkRead;
+      }
+
+      size_t bytesRead;
+      try {
+        bytesRead = transports.input.read(
+          rbuf[totalRead .. totalRead + readSize]);
+      } catch (TTransportException e) {
+        throw new Exception(format(`read(pos = %s, size = %s) threw ` ~
+          `exception "%s"; written so far: %s/%s bytes`, totalRead, readSize,
+          e.msg, totalWritten, totalSize));
+      }
+
+      enforce(bytesRead > 0, format(`read(pos = %s, size = %s) returned %s; ` ~
+        `written so far: %s/%s bytes`, totalRead, readSize, bytesRead,
+        totalWritten, totalSize));
+
+      chunkRead += bytesRead;
+      totalRead += bytesRead;
+    }
+  }
+
+  // make sure the data read back is identical to the data written
+  if (rbuf != wbuf) {
+    stderr.writefln("%s vs. %s", wbuf[$ - 4 .. $], rbuf[$ - 4 .. $]);
+    stderr.writefln("rbuf: %s vs. wbuf: %s", rbuf.length, wbuf.length);
+  }
+  enforce(rbuf == wbuf);
+}
+
+void testReadPartAvailable(CoupledTransports)() if (
+  isCoupledTransports!CoupledTransports
+) {
+  scope transports = new CoupledTransports;
+  assert(transports.input);
+  assert(transports.output);
+
+  ubyte[10] writeBuf = 'a';
+  ubyte[10] readBuf;
+
+  // Attemping to read 10 bytes when only 9 are available should return 9
+  // immediately.
+  transports.output.write(writeBuf[0 .. 9]);
+  transports.output.flush();
+
+  auto t = Trigger(dur!"seconds"(3), transports.output, 1);
+  auto bytesRead = transports.input.read(readBuf);
+  enforce(t.fired == 0);
+  enforce(bytesRead == 9);
+}
+
+void testReadPartialMidframe(CoupledTransports)() if (
+  isCoupledTransports!CoupledTransports
+) {
+  scope transports = new CoupledTransports;
+  assert(transports.input);
+  assert(transports.output);
+
+  ubyte[13] writeBuf = 'a';
+  ubyte[14] readBuf;
+
+  // Attempt to read 10 bytes, when only 9 are available, but after we have
+  // already read part of the data that is available.  This exercises a
+  // different code path for several of the transports.
+  //
+  // For transports that add their own framing (e.g., TFramedTransport and
+  // TFileTransport), the two flush calls break up the data in to a 10 byte
+  // frame and a 3 byte frame.  The first read then puts us partway through the
+  // first frame, and then we attempt to read past the end of that frame, and
+  // through the next frame, too.
+  //
+  // For buffered transports that perform read-ahead (e.g.,
+  // TBufferedTransport), the read-ahead will most likely see all 13 bytes
+  // written on the first read.  The next read will then attempt to read past
+  // the end of the read-ahead buffer.
+  //
+  // Flush 10 bytes, then 3 bytes.  This creates 2 separate frames for
+  // transports that track framing internally.
+  transports.output.write(writeBuf[0 .. 10]);
+  transports.output.flush();
+  transports.output.write(writeBuf[10 .. 13]);
+  transports.output.flush();
+
+  // Now read 4 bytes, so that we are partway through the written data.
+  auto bytesRead = transports.input.read(readBuf[0 .. 4]);
+  enforce(bytesRead == 4);
+
+  // Now attempt to read 10 bytes.  Only 9 more are available.
+  //
+  // We should be able to get all 9 bytes, but it might take multiple read
+  // calls, since it is valid for read() to return fewer bytes than requested.
+  // (Most transports do immediately return 9 bytes, but the framing transports
+  // tend to only return to the end of the current frame, which is 6 bytes in
+  // this case.)
+  size_t totalRead = 0;
+  while (totalRead < 9) {
+    auto t = Trigger(dur!"seconds"(3), transports.output, 1);
+    bytesRead = transports.input.read(readBuf[4 + totalRead .. 14]);
+    enforce(t.fired == 0);
+    enforce(bytesRead > 0);
+    totalRead += bytesRead;
+    enforce(totalRead <= 9);
+  }
+
+  enforce(totalRead == 9);
+}
+
+void testBorrowPartAvailable(CoupledTransports)() if (
+  isCoupledTransports!CoupledTransports
+) {
+  scope transports = new CoupledTransports;
+  assert(transports.input);
+  assert(transports.output);
+
+  ubyte[9] writeBuf = 'a';
+  ubyte[10] readBuf;
+
+  // Attemping to borrow 10 bytes when only 9 are available should return NULL
+  // immediately.
+  transports.output.write(writeBuf);
+  transports.output.flush();
+
+  auto t = Trigger(dur!"seconds"(3), transports.output, 1);
+  auto borrowLen = readBuf.length;
+  auto borrowedBuf = transports.input.borrow(readBuf.ptr, borrowLen);
+  enforce(t.fired == 0);
+  enforce(borrowedBuf is null);
+}
+
+void testReadNoneAvailable(CoupledTransports)() if (
+  isCoupledTransports!CoupledTransports
+) {
+  scope transports = new CoupledTransports;
+  assert(transports.input);
+  assert(transports.output);
+
+  // Attempting to read when no data is available should either block until
+  // some data is available, or fail immediately.  (e.g., TSocket blocks,
+  // TMemoryBuffer just fails.)
+  //
+  // If the transport blocks, it should succeed once some data is available,
+  // even if less than the amount requested becomes available.
+  ubyte[10] readBuf;
+
+  auto t = Trigger(dur!"seconds"(1), transports.output, 2);
+  t.add(dur!"seconds"(1), transports.output, 8);
+
+  auto bytesRead = transports.input.read(readBuf);
+  if (bytesRead == 0) {
+    enforce(t.fired == 0);
+  } else {
+    enforce(t.fired == 1);
+    enforce(bytesRead == 2);
+  }
+}
+
+void testBorrowNoneAvailable(CoupledTransports)() if (
+  isCoupledTransports!CoupledTransports
+) {
+  scope transports = new CoupledTransports;
+  assert(transports.input);
+  assert(transports.output);
+
+  ubyte[16] writeBuf = 'a';
+
+  // Attempting to borrow when no data is available should fail immediately
+  auto t = Trigger(dur!"seconds"(1), transports.output, 10);
+
+  auto borrowLen = 10;
+  auto borrowedBuf = transports.input.borrow(null, borrowLen);
+  enforce(borrowedBuf is null);
+  enforce(t.fired == 0);
+}
+
+
+void doRwTest(CoupledTransports)(
+  size_t totalSize,
+  SizeGenerator wSizeGen,
+  SizeGenerator rSizeGen,
+  SizeGenerator wChunkSizeGen = new ConstantSizeGenerator(0),
+  SizeGenerator rChunkSizeGen = new ConstantSizeGenerator(0),
+  size_t maxOutstanding = 0
+) if (
+  isCoupledTransports!CoupledTransports
+) {
+  totalSize = cast(size_t)(totalSize * g_sizeMultiplier);
+
+  scope(failure) {
+    writefln("Test failed for %s: testReadWrite(%s, %s, %s, %s, %s, %s)",
+      CoupledTransports.stringof, totalSize, wSizeGen, rSizeGen,
+      wChunkSizeGen, rChunkSizeGen, maxOutstanding);
+  }
+
+  testReadWrite!CoupledTransports(totalSize, wSizeGen, rSizeGen,
+    wChunkSizeGen, rChunkSizeGen, maxOutstanding);
+}
+
+void doBlockingTest(CoupledTransports)() if (
+  isCoupledTransports!CoupledTransports
+) {
+  void writeFailure(string name) {
+    writefln("Test failed for %s: %s()", CoupledTransports.stringof, name);
+  }
+
+  {
+    scope(failure) writeFailure("testReadPartAvailable");
+    testReadPartAvailable!CoupledTransports();
+  }
+
+  {
+    scope(failure) writeFailure("testReadPartialMidframe");
+    testReadPartialMidframe!CoupledTransports();
+  }
+
+  {
+    scope(failure) writeFailure("testReadNoneAvaliable");
+    testReadNoneAvailable!CoupledTransports();
+  }
+
+  {
+    scope(failure) writeFailure("testBorrowPartAvailable");
+    testBorrowPartAvailable!CoupledTransports();
+  }
+
+  {
+    scope(failure) writeFailure("testBorrowNoneAvailable");
+    testBorrowNoneAvailable!CoupledTransports();
+  }
+}
+
+SizeGenerator getGenerator(T)(T t) {
+  static if (is(T : SizeGenerator)) {
+    return t;
+  } else {
+    return new ConstantSizeGenerator(t);
+  }
+}
+
+template WrappedTransports(T) if (isCoupledTransports!T) {
+  alias TypeTuple!(
+    T,
+    CoupledBufferedTransports!T,
+    CoupledFramedTransports!T,
+    CoupledZlibTransports!T
+  ) WrappedTransports;
+}
+
+void testRw(C, R, S)(
+  size_t totalSize,
+  R wSize,
+  S rSize
+) if (
+  isCoupledTransports!C && is(typeof(getGenerator(wSize))) &&
+  is(typeof(getGenerator(rSize)))
+) {
+  testRw!C(totalSize, wSize, rSize, 0, 0, 0);
+}
+
+void testRw(C, R, S, T, U)(
+  size_t totalSize,
+  R wSize,
+  S rSize,
+  T wChunkSize,
+  U rChunkSize,
+  size_t maxOutstanding = 0
+) if (
+  isCoupledTransports!C && is(typeof(getGenerator(wSize))) &&
+  is(typeof(getGenerator(rSize))) && is(typeof(getGenerator(wChunkSize))) &&
+  is(typeof(getGenerator(rChunkSize)))
+) {
+  foreach (T; WrappedTransports!C) {
+    doRwTest!T(
+      totalSize,
+      getGenerator(wSize),
+      getGenerator(rSize),
+      getGenerator(wChunkSize),
+      getGenerator(rChunkSize),
+      maxOutstanding
+    );
+  }
+}
+
+void testBlocking(C)() if (isCoupledTransports!C) {
+  foreach (T; WrappedTransports!C) {
+    doBlockingTest!T();
+  }
+}
+
+// A quick hack, for the sake of brevity…
+float g_sizeMultiplier = 1;
+
+version (Posix) {
+  immutable defaultTempDir = "/tmp";
+} else version (Windows) {
+  import core.sys.windows.windows;
+  extern(Windows) DWORD GetTempPathA(DWORD nBufferLength, LPTSTR lpBuffer);
+
+  string defaultTempDir() @property {
+    char[MAX_PATH + 1] dir;
+    enforce(GetTempPathA(dir.length, dir.ptr));
+    return to!string(dir.ptr)[0 .. $ - 1];
+  }
+} else static assert(false);
+
+void main(string[] args) {
+  int seed = unpredictableSeed();
+  string tmpDir = defaultTempDir;
+
+  getopt(args, "seed", &seed, "size-multiplier", &g_sizeMultiplier,
+    "tmp-dir", &tmpDir);
+  enforce(g_sizeMultiplier >= 0, "Size multiplier must not be negative.");
+
+  writefln("Using seed: %s", seed);
+  rndGen().seed(seed);
+  CoupledFileTransports.tmpDir = tmpDir;
+
+  auto rand4k = new RandomSizeGenerator(1, 4096);
+
+  /*
+   * We do the basically the same set of tests for each transport type,
+   * although we tweak the parameters in some places.
+   */
+
+  // TMemoryBuffer tests
+  testRw!CoupledMemoryBuffers(1024 * 1024, 0, 0);
+  testRw!CoupledMemoryBuffers(1024 * 256, rand4k, rand4k);
+  testRw!CoupledMemoryBuffers(1024 * 256, 167, 163);
+  testRw!CoupledMemoryBuffers(1024 * 16, 1, 1);
+
+  testRw!CoupledMemoryBuffers(1024 * 256, 0, 0, rand4k, rand4k);
+  testRw!CoupledMemoryBuffers(1024 * 256, rand4k, rand4k, rand4k, rand4k);
+  testRw!CoupledMemoryBuffers(1024 * 256, 167, 163, rand4k, rand4k);
+  testRw!CoupledMemoryBuffers(1024 * 16, 1, 1, rand4k, rand4k);
+
+  testBlocking!CoupledMemoryBuffers();
+
+  // TSocket tests
+  enum socketMaxOutstanding = 4096;
+  testRw!CoupledSocketTransports(1024 * 1024, 0, 0,
+          0, 0, socketMaxOutstanding);
+  testRw!CoupledSocketTransports(1024 * 256, rand4k, rand4k,
+          0, 0, socketMaxOutstanding);
+  testRw!CoupledSocketTransports(1024 * 256, 167, 163,
+          0, 0, socketMaxOutstanding);
+  // Doh.  Apparently writing to a socket has some additional overhead for
+  // each send() call.  If we have more than ~400 outstanding 1-byte write
+  // requests, additional send() calls start blocking.
+  testRw!CoupledSocketTransports(1024 * 16, 1, 1,
+          0, 0, 250);
+  testRw!CoupledSocketTransports(1024 * 256, 0, 0,
+          rand4k, rand4k, socketMaxOutstanding);
+  testRw!CoupledSocketTransports(1024 * 256, rand4k, rand4k,
+          rand4k, rand4k, socketMaxOutstanding);
+  testRw!CoupledSocketTransports(1024 * 256, 167, 163,
+          rand4k, rand4k, socketMaxOutstanding);
+  testRw!CoupledSocketTransports(1024 * 16, 1, 1,
+          rand4k, rand4k, 250);
+
+  testBlocking!CoupledSocketTransports();
+
+  // File transport tests.
+
+  // Cannot write more than the frame size at once.
+  enum maxWriteAtOnce = 1024 * 1024 * 16 - 4;
+
+  testRw!CoupledFileTransports(1024 * 1024, maxWriteAtOnce, 0);
+  testRw!CoupledFileTransports(1024 * 256, rand4k, rand4k);
+  testRw!CoupledFileTransports(1024 * 256, 167, 163);
+  testRw!CoupledFileTransports(1024 * 16, 1, 1);
+
+  testRw!CoupledFileTransports(1024 * 256, 0, 0, rand4k, rand4k);
+  testRw!CoupledFileTransports(1024 * 256, rand4k, rand4k, rand4k, rand4k);
+  testRw!CoupledFileTransports(1024 * 256, 167, 163, rand4k, rand4k);
+  testRw!CoupledFileTransports(1024 * 16, 1, 1, rand4k, rand4k);
+
+  testBlocking!CoupledFileTransports();
+}
+
+
+/*
+ * Timer handling code for use in tests that check the transport blocking
+ * semantics.
+ *
+ * The implementation has been hacked together in a hurry and wastes a lot of
+ * threads, but speed should not be the concern here.
+ */
+
+struct Trigger {
+  this(Duration timeout, TTransport transport, size_t writeLength) {
+    mutex_ = new Mutex;
+    cancelCondition_ = new Condition(mutex_);
+    info_ = new Info(timeout, transport, writeLength);
+    startThread();
+  }
+
+  ~this() {
+    synchronized (mutex_) {
+      info_ = null;
+      cancelCondition_.notifyAll();
+    }
+    if (thread_) thread_.join();
+  }
+
+  @disable this(this) { assert(0); }
+
+  void add(Duration timeout, TTransport transport, size_t writeLength) {
+    synchronized (mutex_) {
+      auto info = new Info(timeout, transport, writeLength);
+      if (info_) {
+        auto prev = info_;
+        while (prev.next) prev = prev.next;
+        prev.next = info;
+      } else {
+        info_ = info;
+        startThread();
+      }
+    }
+  }
+
+  @property short fired() {
+    return atomicLoad(fired_);
+  }
+
+private:
+  void timerThread() {
+    // KLUDGE: Make sure the std.concurrency mbox is initialized on the timer
+    // thread to be able to unblock the file transport.
+    import std.concurrency;
+    thisTid;
+
+    synchronized (mutex_) {
+      while (info_) {
+        auto cancelled = cancelCondition_.wait(info_.timeout);
+        if (cancelled) {
+          info_ = null;
+          break;
+        }
+
+        atomicOp!"+="(fired_, 1);
+
+        // Write some data to the transport to unblock it.
+        auto buf = new ubyte[info_.writeLength];
+        buf[] = 'b';
+        info_.transport.write(buf);
+        info_.transport.flush();
+
+        info_ = info_.next;
+      }
+    }
+
+    thread_ = null;
+  }
+
+  void startThread() {
+    thread_ = new Thread(&timerThread);
+    thread_.start();
+  }
+
+  struct Info {
+    this(Duration timeout, TTransport transport, size_t writeLength) {
+      this.timeout = timeout;
+      this.transport = transport;
+      this.writeLength = writeLength;
+    }
+
+    Duration timeout;
+    TTransport transport;
+    size_t writeLength;
+    Info* next;
+  }
+
+  Info* info_;
+  Thread thread_;
+  shared short fired_;
+
+  import core.sync.mutex;
+  Mutex mutex_;
+  import core.sync.condition;
+  Condition cancelCondition_;
+}