Fix dynamic pipelining on 'Capability' type.

Fixes #336

c++/src/capnp/capability-test.c++

@@ -338,6 +338,49 @@ TEST(Capability, DynamicClientPipelining) {
   EXPECT_EQ(1, chainedCallCount);
 }
 
+TEST(Capability, DynamicClientPipelineAnyCap) {
+  kj::EventLoop loop;
+  kj::WaitScope waitScope(loop);
+
+  int callCount = 0;
+  int chainedCallCount = 0;
+  DynamicCapability::Client client =
+      test::TestPipeline::Client(kj::heap<TestPipelineImpl>(callCount));
+
+  auto request = client.newRequest("getAnyCap");
+  request.set("n", 234);
+  request.set("inCap", test::TestInterface::Client(kj::heap<TestInterfaceImpl>(chainedCallCount)));
+
+  auto promise = request.send();
+
+  auto outAnyCap = promise.get("outBox").releaseAs<DynamicStruct>()
+                          .get("cap").releaseAs<DynamicCapability>();
+
+  EXPECT_EQ(Schema::from<Capability>(), outAnyCap.getSchema());
+  auto outCap = outAnyCap.castAs<DynamicCapability>(Schema::from<test::TestInterface>());
+
+  auto pipelineRequest = outCap.newRequest("foo");
+  pipelineRequest.set("i", 321);
+  auto pipelinePromise = pipelineRequest.send();
+
+  auto pipelineRequest2 = outCap.castAs<test::TestExtends>().graultRequest();
+  auto pipelinePromise2 = pipelineRequest2.send();
+
+  promise = nullptr;  // Just to be annoying, drop the original promise.
+
+  EXPECT_EQ(0, callCount);
+  EXPECT_EQ(0, chainedCallCount);
+
+  auto response = pipelinePromise.wait(waitScope);
+  EXPECT_EQ("bar", response.get("x").as<Text>());
+
+  auto response2 = pipelinePromise2.wait(waitScope);
+  checkTestMessage(response2);
+
+  EXPECT_EQ(3, callCount);
+  EXPECT_EQ(1, chainedCallCount);
+}
+
 // =======================================================================================
 
 class TestInterfaceDynamicImpl final: public DynamicCapability::Server {

c++/src/capnp/dynamic.c++

@@ -388,6 +388,18 @@ DynamicValue::Pipeline DynamicStruct::Pipeline::get(StructSchema::Field field) {
           return DynamicCapability::Client(type.asInterface(),
               typeless.getPointerField(slot.getOffset()).asCap());
 
+        case schema::Type::ANY_POINTER:
+          switch (type.whichAnyPointerKind()) {
+            case schema::Type::AnyPointer::Unconstrained::STRUCT:
+              return DynamicStruct::Pipeline(StructSchema(),
+                  typeless.getPointerField(slot.getOffset()));
+            case schema::Type::AnyPointer::Unconstrained::CAPABILITY:
+              return DynamicCapability::Client(Capability::Client(
+                  typeless.getPointerField(slot.getOffset()).asCap()));
+            default:
+              KJ_FAIL_REQUIRE("Can only pipeline on struct and interface fields.");
+          }
+
         default:
           KJ_FAIL_REQUIRE("Can only pipeline on struct and interface fields.");
       }

c++/src/capnp/test-util.c++

@@ -952,6 +952,29 @@ kj::Promise<void> TestPipelineImpl::getCap(GetCapContext context) {
       });
 }
 
+kj::Promise<void> TestPipelineImpl::getAnyCap(GetAnyCapContext context) {
+  ++callCount;
+
+  auto params = context.getParams();
+  EXPECT_EQ(234, params.getN());
+
+  auto cap = params.getInCap();
+  context.releaseParams();
+
+  auto request = cap.castAs<test::TestInterface>().fooRequest();
+  request.setI(123);
+  request.setJ(true);
+
+  return request.send().then(
+      [this,KJ_CPCAP(context)](Response<test::TestInterface::FooResults>&& response) mutable {
+        EXPECT_EQ("foo", response.getX());
+
+        auto result = context.getResults();
+        result.setS("bar");
+        result.initOutBox().setCap(kj::heap<TestExtendsImpl>(callCount));
+      });
+}
+
 kj::Promise<void> TestCallOrderImpl::getCallSequence(GetCallSequenceContext context) {
   auto result = context.getResults();
   result.setN(count++);

c++/src/capnp/test-util.h

@@ -213,6 +213,7 @@ public:
   TestPipelineImpl(int& callCount);
 
   kj::Promise<void> getCap(GetCapContext context) override;
+  kj::Promise<void> getAnyCap(GetAnyCapContext context) override;
 
 private:
   int& callCount;

c++/src/capnp/test.capnp

@@ -782,10 +782,14 @@ interface TestExtends2 extends(TestExtends) {}
 interface TestPipeline {
   getCap @0 (n: UInt32, inCap :TestInterface) -> (s: Text, outBox :Box);
   testPointers @1 (cap :TestInterface, obj :AnyPointer, list :List(TestInterface)) -> ();
+  getAnyCap @2 (n: UInt32, inCap :Capability) -> (s: Text, outBox :AnyBox);
 
   struct Box {
     cap @0 :TestInterface;
   }
+  struct AnyBox {
+    cap @0 :Capability;
+  }
 }
 
 interface TestCallOrder {