Merge pull request #741 from capnproto/harris/tee

Implement kj::newTee()

.travis.yml

@@ -24,10 +24,10 @@ matrix:
           sources:
             - ubuntu-toolchain-r-test
           packages:
-            - g++-4.9
+            - g++-5
       env:
-        - MATRIX_CC=gcc-4.9
-        - MATRIX_CXX=g++-4.9
+        - MATRIX_CC=gcc-5
+        - MATRIX_CXX=g++-5
 
     # New GCC
     - os: linux

c++/src/kj/array-test.c++

@@ -447,6 +447,7 @@ TEST(Array, AttachNested) {
   Array<Own<DestructionOrderRecorder>> combined = arr.attach(kj::mv(obj2)).attach(kj::mv(obj3));
 
   KJ_EXPECT(combined.begin() == ptr);
+  KJ_EXPECT(combined.size() == 1);
 
   KJ_EXPECT(obj1.get() == nullptr);
   KJ_EXPECT(obj2.get() == nullptr);

c++/src/kj/array.h

@@ -862,6 +862,7 @@ template <typename T>
 template <typename... Attachments>
 Array<T> Array<T>::attach(Attachments&&... attachments) {
   T* ptrCopy = ptr;
+  auto sizeCopy = size_;
 
   KJ_IREQUIRE(ptrCopy != nullptr, "cannot attach to null pointer");
 
@@ -872,7 +873,7 @@ Array<T> Array<T>::attach(Attachments&&... attachments) {
 
   auto bundle = new _::ArrayDisposableOwnedBundle<Array<T>, Attachments...>(
       kj::mv(*this), kj::fwd<Attachments>(attachments)...);
-  return Array<T>(ptrCopy, size_, *bundle);
+  return Array<T>(ptrCopy, sizeCopy, *bundle);
 }
 
 template <typename T>

c++/src/kj/async-io.h

@@ -203,6 +203,30 @@ struct CapabilityPipe {
   Own<AsyncCapabilityStream> ends[2];
 };
 
+struct Tee {
+  // Two AsyncInputStreams which each read the same data from some wrapped inner AsyncInputStream.
+
+  Own<AsyncInputStream> branches[2];
+};
+
+Tee newTee(Own<AsyncInputStream> input, uint64_t limit = kj::maxValue);
+// Constructs a Tee that operates in-process. The tee buffers data if any read or pump operations is
+// called on one of the two input ends. If a read or pump operation is subsequently called on the
+// other input end, the buffered data is consumed.
+//
+// `pumpTo()` operations on the input ends will proactively read from the inner stream and block
+// while writing to the output stream. While one branch has an active `pumpTo()` operation, any
+// `tryRead()` operation on the other branch will not be allowed to read faster than allowed by the
+// pump's backpressure. (In other words, it will never cause buffering on the pump.) Similarly, if
+// there are `pumpTo()` operations active on both branches, the greater of the two backpressures is
+// respected -- the two pumps progress in lockstep, and there is no buffering.
+//
+// At no point will a branch's buffer be allowed to grow beyond `limit` bytes. If the buffer would
+// grow beyond the limit, an exception is generated, which both branches see once they have
+// exhausted their buffers.
+//
+// It is recommended that you use a more conservative value for `limit` than the default.
+
 class ConnectionReceiver {
   // Represents a server socket listening on a port.
 

c++/src/kj/compat/url.h

@@ -37,12 +37,6 @@ struct UrlOptions {
   bool percentDecode = true;
   // True if URL components should be automatically percent-decoded during parsing, and
   // percent-encoded during serialization.
-
-#if __cplusplus < 201402L
-  inline constexpr UrlOptions(bool percentDecode = true): percentDecode(percentDecode) {}
-  // TODO(cleanup): This constructor is only here to support brace initialization in C++11. It
-  //   should be removed once we upgrade to C++14.
-#endif
 };
 
 struct Url {
@@ -103,17 +97,6 @@ struct Url {
   ~Url() noexcept(false);
   Url& operator=(Url&&) = default;
 
-#if __cplusplus < 201402L
-  inline Url(String&& scheme, Maybe<UserInfo>&& userInfo, String&& host, Vector<String>&& path,
-             bool hasTrailingSlash, Vector<QueryParam>&& query, Maybe<String>&& fragment,
-             UrlOptions options)
-      : scheme(kj::mv(scheme)), userInfo(kj::mv(userInfo)), host(kj::mv(host)), path(kj::mv(path)),
-        hasTrailingSlash(hasTrailingSlash), query(kj::mv(query)), fragment(kj::mv(fragment)),
-        options(options) {}
-  // TODO(cleanup): This constructor is only here to support brace initialization in C++11. It
-  //   should be removed once we upgrade to C++14.
-#endif
-
   Url clone() const;
 
   enum Context {