Handle incoming receiverAnswer cap descriptors.

answer.go

@@ -181,6 +181,18 @@ func (p *Promise) Reject(e error) {
 	p.resolve(Ptr{}, e)
 }
 
+// Resolve resolves the promise.
+//
+// If e != nil, then this is equivalent to p.Reject(e).
+// Otherwise, it is equivalent to p.Fulfill(r).
+func (p *Promise) Resolve(r Ptr, e error) {
+	if e != nil {
+		p.Reject(e)
+	} else {
+		p.Fulfill(r)
+	}
+}
+
 // resolve moves p into the resolved state from unresolved or pending
 // join.  The caller must be holding onto p.mu.
 func (p *Promise) resolve(r Ptr, e error) {

rpc/answer.go

@@ -60,6 +60,9 @@ type answer struct {
 	// pcall is the PipelineCaller returned by RecvCall.  It will be set
 	// to nil once results are ready.
 	pcall capnp.PipelineCaller
+	// promise is a promise wrapping pcall. It will be resolved, and
+	// subsequently set to nil, once the results are ready.
+	promise *capnp.Promise
 
 	// pcalls is added to for every pending RecvCall and subtracted from
 	// for every RecvCall return (delivery acknowledgement).  This is used
@@ -109,10 +112,13 @@ func (c *Conn) newReturn(ctx context.Context) (rpccp.Return, func() error, capnp
 // setPipelineCaller sets ans.pcall to pcall if the answer has not
 // already returned.  The caller MUST NOT be holding onto ans.c.mu
 // or the sender lock.
-func (ans *answer) setPipelineCaller(pcall capnp.PipelineCaller) {
+//
+// This also sets ans.promise to a new promise, wrapping pcall.
+func (ans *answer) setPipelineCaller(m capnp.Method, pcall capnp.PipelineCaller) {
 	syncutil.With(&ans.c.mu, func() {
 		if ans.flags&resultsReady == 0 {
 			ans.pcall = pcall
+			ans.promise = capnp.NewPromise(m, pcall)
 		}
 	})
 }
@@ -202,11 +208,12 @@ func (ans *answer) Return(e error) {
 //
 // The caller must be holding onto ans.c.mu and the sender lock.
 // The result's capability table must have been extracted into
-// ans.resultsCapTable before calling sendReturn. Only one of
+// ans.resultCapTable before calling sendReturn. Only one of
 // sendReturn or sendException should be called.
 func (ans *answer) sendReturn() (releaseList, error) {
 	ans.pcall = nil
 	ans.flags |= resultsReady
+
 	var err error
 	ans.exportRefs, err = ans.c.fillPayloadCapTable(ans.results, ans.resultCapTable)
 	if err != nil {
@@ -218,6 +225,18 @@ func (ans *answer) sendReturn() (releaseList, error) {
 	case <-ans.c.bgctx.Done():
 	default:
 		fin := ans.flags&finishReceived != 0
+		if ans.promise != nil {
+			if fin {
+				// Can't use ans.result after a finish, but it's
+				// ok to return an error if the finish comes in
+				// before the return. Possible enhancement: use
+				// the cancel variant of return.
+				ans.promise.Reject(failedf("Received finish before return"))
+			} else {
+				ans.promise.Resolve(ans.results.Content())
+			}
+			ans.promise = nil
+		}
 		ans.c.mu.Unlock()
 		if err := ans.sendMsg(); err != nil {
 			ans.c.report(failedf("send return: %w", err))
@@ -244,13 +263,18 @@ func (ans *answer) sendReturn() (releaseList, error) {
 //
 // The caller must be holding onto ans.c.mu and the sender lock.
 // The result's capability table must have been extracted into
-// ans.resultsCapTable before calling sendException. Only one of
+// ans.resultCapTable before calling sendException. Only one of
 // sendReturn or sendException should be called.
 func (ans *answer) sendException(e error) releaseList {
 	ans.err = e
 	ans.pcall = nil
 	ans.flags |= resultsReady
 
+	if ans.promise != nil {
+		ans.promise.Reject(e)
+		ans.promise = nil
+	}
+
 	select {
 	case <-ans.c.bgctx.Done():
 	default:

rpc/internal/testcapnp/test.capnp

@@ -13,3 +13,8 @@ interface PingPong {
 interface StreamTest {
   push @0 (data :Data) -> stream;
 }
+
+interface CapArgsTest {
+  call @0 (cap :Capability);
+  self @1 () -> (self :CapArgsTest);
+}