Merge pull request #899 from capnproto/http-over-rpc-fixes

Various bug fixes

c++/src/capnp/compat/byte-stream.c++

@@ -603,6 +603,8 @@ protected:
       KJ_CASE_ONEOF(capnpStream, capnp::ByteStream::Client) {
         auto params = context.getParams();
         auto req = capnpStream.getSubstreamRequest(params.totalSize());
+        req.setCallback(params.getCallback());
+        req.setLimit(params.getLimit());
         return context.tailCall(kj::mv(req));
       }
       KJ_CASE_ONEOF(b, Borrowed) {

c++/src/capnp/compat/http-over-capnp.c++

@@ -449,12 +449,13 @@ public:
                            kj::HttpService& kjService)
       : factory(factory),
         method(validateMethod(request.getMethod())),
+        url(kj::str(request.getUrl())),
         headers(factory.headersToKj(request.getHeaders()).clone()),
         clientContext(kj::mv(clientContext)),
         // Note we attach `requestBodyIn` to `task` so that we will implicitly cancel reading
         // the request body as soon as the service returns. This is important in particular when
         // the request body is not fully consumed, in order to propagate cancellation.
-        task(kjService.request(method, request.getUrl(), headers, *requestBodyIn, *this)
+        task(kjService.request(method, url, headers, *requestBodyIn, *this)
                       .attach(kj::mv(requestBodyIn))) {}
 
   KJ_DISALLOW_COPY(ServerRequestContextImpl);
@@ -544,6 +545,7 @@ public:
 private:
   HttpOverCapnpFactory& factory;
   kj::HttpMethod method;
+  kj::String url;
   kj::HttpHeaders headers;
   capnp::HttpService::ClientRequestContext::Client clientContext;
   kj::Promise<void> task;

c++/src/capnp/membrane.c++

@@ -262,7 +262,7 @@ public:
   }
 
   void releaseParams() override {
-    KJ_REQUIRE(!releasedParams);
+    // Note that releaseParams() is idempotent -- it can be called multiple times.
     releasedParams = true;
     inner->releaseParams();
   }

c++/src/kj/compat/http.c++

@@ -3059,7 +3059,12 @@ private:
     }
     kj::Promise<void> close(uint16_t code, kj::StringPtr reason) override {
       KJ_REQUIRE(canceler.isEmpty(), "another message send is already in progress");
-      return canceler.wrap(output.close(code, reason));
+      return canceler.wrap(output.close(code, reason).then([this]() {
+        // A pump is expected to end upon seeing a Close message.
+        canceler.release();
+        pipe.endState(*this);
+        fulfiller.fulfill();
+      }));
     }
     kj::Promise<void> disconnect() override {
       KJ_REQUIRE(canceler.isEmpty(), "another message send is already in progress");