mirror of
https://github.com/zap-proto/go.git
synced 2026-10-30 15:05:00 +00:00
The call envelope has always carried Target @8, but zapgen hardcoded Target: rpc.NoTarget in every generated method, so Go could DECODE a pipelined envelope yet never ORIGINATE or RESOLVE one — pipelining was a no-op. Make the canonical Target model real on both sides. Client (origination): rpc.Session allocates unique non-zero PromiseIDs and builds Calls — Origin (Target=NoTarget) and Pipeline (Target=prior.ID). zapgen now emits a *rpc.Session-backed client: M(...) originates, MOn(on rpc.Promise) pipelines on a prior answer (payload supplied server-side). NoTarget is no longer hardcoded — it flows through one shared invoke<M> path. Both forms return the call's own rpc.Promise so chains compose. Server (resolution): rpc.Pipeliner wraps a generated Dispatch<Iface> and is the per-connection promise table: records each OK answer under its PromiseID, substitutes a resolved Target's Body for the dependent's Payload before dispatch, QUEUES a dependent whose Target hasn't resolved yet until it does, and REFUSES (StatusBadRequest, never hangs) a dependent whose Target answered non-OK or was Finished. Finish(id) bounds the table. Wire-compatible: Target=NoTarget is the non-pipelined call, so a Pipeliner and a plain dispatcher are byte-identical on the wire; a non-pipelining peer (any other runtime) interoperates trivially. Proven end-to-end in rpc/pipeline_test.go (auth->getResource): single-pass resolution, server-side queuing when the dependent arrives first, a 3-deep A->B->C chain, non-OK/Finished targets refused, Finish waking a parked dependent, and Target riding on the wire byte-for-byte. Full suite green under -race. Docs: README + LLM.md describe the one canonical model and the Rust capnp PromisedAnswer boundary honestly.
199 lines
6.0 KiB
Go
199 lines
6.0 KiB
Go
// Code generated by zapgen; DO NOT EDIT.
|
|
// source: echo.zap
|
|
|
|
package echo
|
|
|
|
import (
|
|
"fmt"
|
|
"github.com/zap-proto/go/rpc"
|
|
)
|
|
|
|
// Method ordinals for the Echo service (stable 1-based wire ids).
|
|
const (
|
|
EchoPingOrdinal uint32 = 1
|
|
EchoNotifyOrdinal uint32 = 2
|
|
EchoHealthOrdinal uint32 = 3
|
|
EchoShutdownOrdinal uint32 = 4
|
|
)
|
|
|
|
// EchoChannel ships one Call envelope and awaits its correlated Response.
|
|
type EchoChannel interface {
|
|
Call(envelope []byte) (rpc.Response, error)
|
|
}
|
|
|
|
// EchoClient is a typed RPC client for the Echo service over a ZAP call channel.
|
|
// Each call takes a fresh PromiseID from sess; the pipelined "On" form of a
|
|
// method sets Target to a prior call's Promise so the server chains them.
|
|
type EchoClient struct {
|
|
ch EchoChannel
|
|
cap []byte
|
|
sess *rpc.Session
|
|
}
|
|
|
|
// NewEchoClient returns a client that issues calls over ch, attaching cap
|
|
// (which may be nil) to every request.
|
|
func NewEchoClient(ch EchoChannel, capability []byte) *EchoClient {
|
|
return &EchoClient{ch: ch, cap: capability, sess: rpc.NewSession()}
|
|
}
|
|
|
|
func (c *EchoClient) Ping(req []byte) (rpc.Promise, []byte, error) {
|
|
return c.invokePing(rpc.NoTarget, req)
|
|
}
|
|
|
|
// PingOn issues Ping as a dependent call pipelined on the answer of on:
|
|
// the server substitutes on's resolved result for this call's payload
|
|
// before dispatch, so it ships without waiting for on to round-trip.
|
|
func (c *EchoClient) PingOn(on rpc.Promise) (rpc.Promise, []byte, error) {
|
|
return c.invokePing(on.ID, nil)
|
|
}
|
|
|
|
func (c *EchoClient) invokePing(target uint32, payload []byte) (rpc.Promise, []byte, error) {
|
|
p := c.sess.Next()
|
|
resp, err := c.ch.Call(rpc.BuildRequest(rpc.Call{
|
|
Method: EchoPingOrdinal,
|
|
PromiseID: p.ID,
|
|
Target: target,
|
|
Cap: c.cap,
|
|
Payload: payload,
|
|
}))
|
|
if err != nil {
|
|
return p, nil, err
|
|
}
|
|
if resp.Status != rpc.StatusOK {
|
|
return p, nil, fmt.Errorf("Echo.Ping: status %d", resp.Status)
|
|
}
|
|
return p, resp.Body, nil
|
|
}
|
|
|
|
func (c *EchoClient) Notify(req []byte) (rpc.Promise, error) {
|
|
return c.invokeNotify(rpc.NoTarget, req)
|
|
}
|
|
|
|
// NotifyOn issues Notify as a dependent call pipelined on the answer of on:
|
|
// the server substitutes on's resolved result for this call's payload
|
|
// before dispatch, so it ships without waiting for on to round-trip.
|
|
func (c *EchoClient) NotifyOn(on rpc.Promise) (rpc.Promise, error) {
|
|
return c.invokeNotify(on.ID, nil)
|
|
}
|
|
|
|
func (c *EchoClient) invokeNotify(target uint32, payload []byte) (rpc.Promise, error) {
|
|
p := c.sess.Next()
|
|
resp, err := c.ch.Call(rpc.BuildRequest(rpc.Call{
|
|
Method: EchoNotifyOrdinal,
|
|
PromiseID: p.ID,
|
|
Target: target,
|
|
Cap: c.cap,
|
|
Payload: payload,
|
|
}))
|
|
if err != nil {
|
|
return p, err
|
|
}
|
|
if resp.Status != rpc.StatusOK {
|
|
return p, fmt.Errorf("Echo.Notify: status %d", resp.Status)
|
|
}
|
|
return p, nil
|
|
}
|
|
|
|
func (c *EchoClient) Health() (rpc.Promise, []byte, error) {
|
|
return c.invokeHealth(rpc.NoTarget, nil)
|
|
}
|
|
|
|
// HealthOn issues Health as a dependent call pipelined on the answer of on:
|
|
// the server substitutes on's resolved result for this call's payload
|
|
// before dispatch, so it ships without waiting for on to round-trip.
|
|
func (c *EchoClient) HealthOn(on rpc.Promise) (rpc.Promise, []byte, error) {
|
|
return c.invokeHealth(on.ID, nil)
|
|
}
|
|
|
|
func (c *EchoClient) invokeHealth(target uint32, payload []byte) (rpc.Promise, []byte, error) {
|
|
p := c.sess.Next()
|
|
resp, err := c.ch.Call(rpc.BuildRequest(rpc.Call{
|
|
Method: EchoHealthOrdinal,
|
|
PromiseID: p.ID,
|
|
Target: target,
|
|
Cap: c.cap,
|
|
Payload: payload,
|
|
}))
|
|
if err != nil {
|
|
return p, nil, err
|
|
}
|
|
if resp.Status != rpc.StatusOK {
|
|
return p, nil, fmt.Errorf("Echo.Health: status %d", resp.Status)
|
|
}
|
|
return p, resp.Body, nil
|
|
}
|
|
|
|
func (c *EchoClient) Shutdown() (rpc.Promise, error) {
|
|
return c.invokeShutdown(rpc.NoTarget, nil)
|
|
}
|
|
|
|
// ShutdownOn issues Shutdown as a dependent call pipelined on the answer of on:
|
|
// the server substitutes on's resolved result for this call's payload
|
|
// before dispatch, so it ships without waiting for on to round-trip.
|
|
func (c *EchoClient) ShutdownOn(on rpc.Promise) (rpc.Promise, error) {
|
|
return c.invokeShutdown(on.ID, nil)
|
|
}
|
|
|
|
func (c *EchoClient) invokeShutdown(target uint32, payload []byte) (rpc.Promise, error) {
|
|
p := c.sess.Next()
|
|
resp, err := c.ch.Call(rpc.BuildRequest(rpc.Call{
|
|
Method: EchoShutdownOrdinal,
|
|
PromiseID: p.ID,
|
|
Target: target,
|
|
Cap: c.cap,
|
|
Payload: payload,
|
|
}))
|
|
if err != nil {
|
|
return p, err
|
|
}
|
|
if resp.Status != rpc.StatusOK {
|
|
return p, fmt.Errorf("Echo.Shutdown: status %d", resp.Status)
|
|
}
|
|
return p, nil
|
|
}
|
|
|
|
// EchoHandler is the server contract for the Echo service. Implement each
|
|
// method, then route requests to it with DispatchEcho.
|
|
type EchoHandler interface {
|
|
Ping(req []byte) ([]byte, error)
|
|
Notify(req []byte) error
|
|
Health() ([]byte, error)
|
|
Shutdown() error
|
|
}
|
|
|
|
// DispatchEcho decodes a Call envelope, routes it by method ordinal to h,
|
|
// and returns the response envelope. An unknown ordinal yields a
|
|
// StatusNotFound response; a handler error yields StatusInternal.
|
|
func DispatchEcho(h EchoHandler, envelope []byte) ([]byte, error) {
|
|
call, err := rpc.ParseRequest(envelope)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
switch call.Method {
|
|
case EchoPingOrdinal:
|
|
body, err := h.Ping(call.Payload)
|
|
if err != nil {
|
|
return rpc.BuildResponse(rpc.StatusInternal, call.PromiseID, nil), nil
|
|
}
|
|
return rpc.BuildResponse(rpc.StatusOK, call.PromiseID, body), nil
|
|
case EchoNotifyOrdinal:
|
|
if err := h.Notify(call.Payload); err != nil {
|
|
return rpc.BuildResponse(rpc.StatusInternal, call.PromiseID, nil), nil
|
|
}
|
|
return rpc.BuildResponse(rpc.StatusOK, call.PromiseID, nil), nil
|
|
case EchoHealthOrdinal:
|
|
body, err := h.Health()
|
|
if err != nil {
|
|
return rpc.BuildResponse(rpc.StatusInternal, call.PromiseID, nil), nil
|
|
}
|
|
return rpc.BuildResponse(rpc.StatusOK, call.PromiseID, body), nil
|
|
case EchoShutdownOrdinal:
|
|
if err := h.Shutdown(); err != nil {
|
|
return rpc.BuildResponse(rpc.StatusInternal, call.PromiseID, nil), nil
|
|
}
|
|
return rpc.BuildResponse(rpc.StatusOK, call.PromiseID, nil), nil
|
|
default:
|
|
return rpc.BuildResponse(rpc.StatusNotFound, call.PromiseID, nil), nil
|
|
}
|
|
}
|