capnweb

type AbortMsg ¶

type AbortMsg struct {
	Expr json.RawMessage
}

AbortMsg is a fatal error that terminates the session. No further messages are sent or received after this.

Wire format: ["abort", expression]

type ArrayExpr ¶

type ArrayExpr struct{ Elements []Expr }

ArrayExpr wraps a plain array of expressions. On the wire actual arrays are encoded as [elem0, elem1, ...] — the encoder handles distinguishing them from typed expression arrays.

type BatchClient ¶

type BatchClient struct {
	URL        string
	HTTPClient *http.Client
}

BatchClient sends batches of messages to an HTTP batch endpoint. BatchClient sends batches of RPC messages to an HTTP batch endpoint. Set URL to the batch endpoint. HTTPClient is optional (defaults to http.DefaultClient).

func (c *BatchClient) Do(ctx context.Context, messages []Message) ([]Message, error)

type BigIntExpr ¶

type BigIntExpr struct{ Value *big.Int }

BigIntExpr represents an arbitrary-precision integer — ["bigint", decimal].

type BytesExpr ¶

type BytesExpr struct{ Data []byte }

BytesExpr represents a byte slice — ["bytes", base64].

type DateExpr ¶

type DateExpr struct{ Time time.Time }

DateExpr represents a timestamp — ["date", unixMillis].

type ErrorExpr ¶

type ErrorExpr struct {
	Type    string // "Error", "TypeError", "RangeError", etc.
	Message string
	Stack   string // optional
}

ErrorExpr represents a remote error — ["error", type, message, stack?].

func NewRangeError(message string) *ErrorExpr

func NewReferenceError(message string) *ErrorExpr

func NewTypeError(message string) *ErrorExpr

func (e ErrorExpr) Error() string

type ExportEntry ¶

type ExportEntry struct {
	ID       int64
	Target   any
	RefCount int64 // times we've exported this ID
}

ExportEntry represents a single export table slot.

type ExportExpr ¶

type ExportExpr struct{ ExportID int64 }

ExportExpr exports a local object — ["export", id].

type ExportTable ¶

type ExportTable struct {
	// contains filtered or unexported fields
}

ExportTable tracks objects exported to the remote endpoint. The exporting side allocates negative IDs starting from -1.

func NewExportTable(main any) *ExportTable

func (t *ExportTable) Export(target any) *ExportEntry

func (t *ExportTable) ExportWithID(id int64, target any) *ExportEntry

func (t *ExportTable) Get(id int64) *ExportEntry

func (t *ExportTable) HandleRelease(id, refcount int64) bool

type Expr ¶

type Expr interface {
	// contains filtered or unexported methods
}

Expr represents a capnweb expression value. All values transmitted over the protocol are expressions — either literal JSON values or typed arrays like ["bytes", "..."], ["date", 123], ["import", 1, "method", [args]], etc.

The array wrapping rule: non-array JSON values are interpreted literally. Arrays where element 0 is a recognized string tag are typed expressions. Actual plain arrays must be wrapped: [elem0, elem1] on the wire represents a two-element array, distinguished from expressions by the first element not being a known tag string (or by being an ArrayExpr during encoding).

func DecodeExpr(data json.RawMessage) (Expr, error)

type Future ¶

type Future struct {
	// contains filtered or unexported fields
}

Future represents a pending result that will be resolved or rejected.

func NewFuture() *Future

func (f *Future) Await(ctx context.Context) (any, error)

func (f *Future) Done() <-chan struct{}

func (f *Future) Reject(err error)

func (f *Future) Resolve(val any)

func (f *Future) Settled() bool

type HeadersExpr ¶

type HeadersExpr struct{ Header http.Header }

HeadersExpr represents HTTP headers — ["headers", [[name, value], ...]].

type ImportEntry ¶

type ImportEntry struct {
	ID       int64
	RefCount int64 // times this import was "introduced" to us
	Resolved bool
	Value    any // resolved value, or nil if pending
}

ImportEntry represents a single import table slot.

type ImportExpr ¶

type ImportExpr struct {
	ImportID int64
	Path     []string
	Args     []Expr // nil = no call; empty = call with zero args
}

ImportExpr references an import table entry — ["import", id, path?, args?]. Evaluates to a stub.

type ImportTable ¶

type ImportTable struct {
	// contains filtered or unexported fields
}

ImportTable tracks objects imported from the remote endpoint. The importing side allocates positive IDs starting from 1.

func NewImportTable() *ImportTable

func (t *ImportTable) AddRef(id int64)

func (t *ImportTable) Allocate() *ImportEntry

func (t *ImportTable) Get(id int64) *ImportEntry

func (t *ImportTable) Insert(id int64) *ImportEntry

func (t *ImportTable) Release(id, count int64) bool

func (t *ImportTable) Remove(id int64)

type InfExpr ¶

type InfExpr struct{}

InfExpr represents +Infinity — ["inf"].

type LiteralExpr ¶

type LiteralExpr struct{ Value any }

LiteralExpr wraps a plain JSON value (string, number, bool, null, or object).

type Message ¶

type Message interface {
	// contains filtered or unexported methods
}

Message is a capnweb protocol message. Each concrete type corresponds to one of the 8 wire message types.

func ReadNDJSON(r io.Reader) ([]Message, error)

func UnmarshalMessage(data []byte) (Message, error)

type NaNExpr ¶

type NaNExpr struct{}

NaNExpr represents NaN — ["nan"].

type NegInfExpr ¶

type NegInfExpr struct{}

NegInfExpr represents -Infinity — ["-inf"].

type PipeMsg ¶

type PipeMsg struct{}

PipeMsg creates a bidirectional pipe on the remote end. The sender implicitly assigns the next positive import ID, usable as a WritableStream.

Wire format: ["pipe"]

type PipelineExpr ¶

type PipelineExpr struct {
	ImportID int64
	Path     []string
	Args     []Expr
}

PipelineExpr is like ImportExpr but evaluates to a promise — ["pipeline", id, path?, args?].

type PromiseExpr ¶

type PromiseExpr struct{ ExportID int64 }

PromiseExpr exports a promise — ["promise", id].

type PullMsg ¶

type PullMsg struct {
	ImportID int64
}

PullMsg signals that the sender wants a resolve/reject for a promise import.

Wire format: ["pull", importId]

type PushMsg ¶

type PushMsg struct {
	Expr json.RawMessage
}

PushMsg requests the recipient to evaluate an expression. The sender implicitly assigns the next positive import ID to the result.

Wire format: ["push", expression]

type ReadableExpr ¶

type ReadableExpr struct{ ImportID int64 }

ReadableExpr references the readable end of a pipe — ["readable", importId].

type RejectMsg ¶

type RejectMsg struct {
	ExportID int64
	Expr     json.RawMessage
}

RejectMsg delivers a rejection for a promise export. The expression must not contain stubs — it typically evaluates to an error.

Wire format: ["reject", exportId, expression]

type ReleaseMsg ¶

type ReleaseMsg struct {
	ImportID int64
	RefCount int64
}

ReleaseMsg releases an import table entry.

Wire format: ["release", importId, refcount]

type RemapExpr ¶

type RemapExpr struct {
	ImportID     int64
	Path         []string
	Captures     []Expr
	Instructions []Expr
}

RemapExpr represents a server-side .map() — ["remap", importId, path, captures, instructions].

type RequestExpr ¶

type RequestExpr struct {
	URL     string
	Method  string
	Headers http.Header
	Body    Expr // may be nil
}

RequestExpr represents an HTTP request — ["request", url, init].

type ResolveMsg ¶

type ResolveMsg struct {
	ExportID int64
	Expr     json.RawMessage
}

ResolveMsg delivers the resolution of a promise export.

Wire format: ["resolve", exportId, expression]

type ResponseExpr ¶

type ResponseExpr struct {
	Status     int
	StatusText string
	Headers    http.Header
	Body       Expr // may be nil
}

ResponseExpr represents an HTTP response — ["response", body, init].

type RpcTarget ¶

type RpcTarget interface {
	IsRpcTarget()
}

RpcTarget is implemented by types that should be passed by reference over the wire. Instead of being serialized as JSON, RpcTarget values are exported into the session's export table and sent as ["export", id] expressions.

Embed RpcTargetBase in your struct to implement this interface:

type MyService struct {
    capnweb.RpcTargetBase
}

type RpcTargetBase ¶

type RpcTargetBase struct{}

RpcTargetBase is embedded in structs to mark them as pass-by-reference RPC targets. This is the Go equivalent of extending the RpcTarget class in the TypeScript reference implementation.

func (RpcTargetBase) IsRpcTarget()

type Session ¶

type Session struct {
	// contains filtered or unexported fields
}

Session manages a single capnweb RPC connection. It is fully bidirectional — either side can call methods on objects exported by the other.

func NewSession(transport Transport, main any) *Session

package main

import (
	"context"
	"fmt"
	"net/http"
	"net/http/httptest"
	"strings"
	"time"

	capnweb "github.com/flaticols/capnweb-go"
)

// A Greeter is a simple RPC service. Embed RpcTargetBase to mark it
// as pass-by-reference when returned from other methods.
type Greeter struct {
	capnweb.RpcTargetBase
}

func (g *Greeter) Greet(_ context.Context, name string) (string, error) {
	return "Hello, " + name + "!", nil
}

func (g *Greeter) Add(_ context.Context, a, b float64) (float64, error) {
	return a + b, nil
}

func (g *Greeter) Fail(_ context.Context) (any, error) {
	return nil, fmt.Errorf("something went wrong")
}

func main() {
	// Start a Go server with a Greeter service.
	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		tr, _ := capnweb.WSAccept(w, r, &capnweb.WSAcceptOptions{Origins: []string{"*"}})
		sess := capnweb.NewSession(tr, &Greeter{})
		sess.Run(r.Context())
	}))
	defer srv.Close()

	wsURL := "ws" + strings.TrimPrefix(srv.URL, "http")

	// Connect a Go client.
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()

	tr, _ := capnweb.WSDial(ctx, wsURL, nil)
	client := capnweb.NewSession(tr, nil)
	go client.Run(ctx)
	defer client.Close()

	// Call the Greet method.
	main := client.Main()
	result, _ := capnweb.Call[string](ctx, main, "Greet", "World")
	fmt.Println(result)
}

Output:
Hello, World!

func (s *Session) Abort(reason error) error

func (s *Session) Call(ctx context.Context, targetID int64, method string, args ...any) (any, error)

func (s *Session) Close() error

func (s *Session) CreatePipe(ctx context.Context) (*StreamWriter, Expr, error)

package main

import (
	"context"
	"fmt"
	"io"
	"net/http"
	"net/http/httptest"
	"strings"
	"time"

	capnweb "github.com/flaticols/capnweb-go"
)

// StreamService collects chunks from a stream into a single string.
type StreamService struct {
	capnweb.RpcTargetBase
}

func (s *StreamService) Collect(_ context.Context, reader *capnweb.StreamReader) (string, error) {
	var sb strings.Builder
	for {
		chunk, err := reader.Read(context.Background())
		if err == io.EOF {
			break
		}
		if err != nil {
			return "", err
		}
		sb.WriteString(chunk.(string))
	}
	return sb.String(), nil
}

func main() {
	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		tr, _ := capnweb.WSAccept(w, r, &capnweb.WSAcceptOptions{Origins: []string{"*"}})
		sess := capnweb.NewSession(tr, &StreamService{})
		sess.Run(r.Context())
	}))
	defer srv.Close()

	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()

	tr, _ := capnweb.WSDial(ctx, "ws"+strings.TrimPrefix(srv.URL, "http"), nil)
	client := capnweb.NewSession(tr, nil)
	go client.Run(ctx)
	defer client.Close()

	// Create a pipe and pass the readable end to the server.
	writer, readable, _ := client.CreatePipe(ctx)

	resultCh := make(chan string, 1)
	go func() {
		main := client.Main()
		r, _ := capnweb.Call[string](ctx, main, "Collect", readable)
		resultCh <- r
	}()

	// Write chunks through the pipe.
	writer.Write(ctx, "Hello")
	writer.Write(ctx, ", ")
	writer.Write(ctx, "World!")
	writer.Close(ctx)

	fmt.Println(<-resultCh)
}

Output:
Hello, World!

func (s *Session) Done() <-chan struct{}

func (s *Session) Err() error

func (s *Session) Main() *Stub

func (s *Session) Release(ctx context.Context, importID, refCount int64) error

func (s *Session) Run(ctx context.Context) error

type StreamMsg ¶

type StreamMsg struct {
	Expr json.RawMessage
}

StreamMsg is like PushMsg but optimized for streaming: no pipelining on the result, auto-pulled, and implicitly released with refcount 1 on resolve/reject.

Wire format: ["stream", expression]

type StreamReader ¶

type StreamReader struct {
	// contains filtered or unexported fields
}

StreamReader reads chunks from a pipe. Obtained from a ReadableExpr when a method receives a stream argument.

func (r *StreamReader) Read(ctx context.Context) (any, error)

type StreamWriter ¶

type StreamWriter struct {
	// contains filtered or unexported fields
}

StreamWriter writes chunks to a remote pipe via stream messages. Each write sends a ["stream", ...] message and waits for the resolve (backpressure).

func (w *StreamWriter) Abort(ctx context.Context, reason error) error

func (w *StreamWriter) Close(ctx context.Context) error

func (w *StreamWriter) Write(ctx context.Context, chunk any) error

type Stub ¶

type Stub struct {
	// contains filtered or unexported fields
}

Stub represents a remote object accessible through a session. It wraps an import table entry and provides methods to call remote methods and release the reference.

Stubs are created via Session.Main (for the bootstrap interface) or returned from Stub.Call when the remote returns a pass-by-reference object.

When done with a stub, call Stub.Release to send a release message to the remote. A runtime finalizer is set as a safety net, but explicit release is preferred for deterministic cleanup.

func (s *Stub) Call(ctx context.Context, method string, args ...any) (any, error)

func (s *Stub) ID() int64

func (s *Stub) Pipeline(ctx context.Context, method string, args ...any) (*Stub, error)

auth, _ := main.Pipeline(ctx, "Authenticate", token)  // push only
data, _ := Call[string](ctx, auth, "GetData")          // push + pull + await
defer auth.Release(ctx)

package main

import (
	"context"
	"fmt"
	"net/http"
	"net/http/httptest"
	"strings"
	"time"

	capnweb "github.com/flaticols/capnweb-go"
)

// Calculator is an RpcTarget returned by reference from MathService.
type Calculator struct {
	capnweb.RpcTargetBase
}

func (c *Calculator) Multiply(_ context.Context, a, b float64) (float64, error) {
	return a * b, nil
}

type MathService struct {
	capnweb.RpcTargetBase
}

func (s *MathService) GetCalculator(_ context.Context) (*Calculator, error) {
	return &Calculator{}, nil
}

func main() {
	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		tr, _ := capnweb.WSAccept(w, r, &capnweb.WSAcceptOptions{Origins: []string{"*"}})
		sess := capnweb.NewSession(tr, &MathService{})
		sess.Run(r.Context())
	}))
	defer srv.Close()

	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()

	tr, _ := capnweb.WSDial(ctx, "ws"+strings.TrimPrefix(srv.URL, "http"), nil)
	client := capnweb.NewSession(tr, nil)
	go client.Run(ctx)
	defer client.Close()

	main := client.Main()

	// Pipeline: GetCalculator returns a stub, then Multiply is called
	// on it — both calls are sent before waiting for any result.
	calc, _ := main.Pipeline(ctx, "GetCalculator")
	result, _ := capnweb.Call[float64](ctx, calc, "Multiply", 6.0, 7.0)
	calc.Release(ctx)

	fmt.Println(result)
}

Output:
42

func (s *Stub) Release(ctx context.Context) error

func (s *Stub) String() string

type Transport ¶

type Transport interface {
	// Send sends a message to the remote endpoint.
	// Must be safe for concurrent use from multiple goroutines.
	Send(ctx context.Context, msg Message) error

	// Recv receives the next message from the remote endpoint.
	// Blocks until a message is available, the context is cancelled, or the
	// transport is closed.
	Recv(ctx context.Context) (Message, error)

	// Close closes the transport. Any blocked Recv call returns an error.
	Close() error
}

Transport is a bidirectional message stream. Implementations handle framing and JSON serialization for a specific underlying protocol (WebSocket, HTTP batch, etc.).

type UndefinedExpr ¶

type UndefinedExpr struct{}

UndefinedExpr represents JavaScript undefined — ["undefined"].

type WSAcceptOptions ¶

type WSAcceptOptions struct {
	Origins []string // allowed origins; nil allows any
}

WSAcceptOptions configures server-side WebSocket upgrade.

type WSDialOptions ¶

type WSDialOptions struct {
	HTTPClient *http.Client
}

WSDialOptions configures a client WebSocket connection.

type WSTransport ¶

type WSTransport struct {
	// contains filtered or unexported fields
}

WSTransport implements Transport over a WebSocket connection. One WebSocket text message = one capnweb message.

func NewWSTransport(conn *websocket.Conn) *WSTransport

func WSAccept(w http.ResponseWriter, r *http.Request, opts *WSAcceptOptions) (*WSTransport, error)

func WSDial(ctx context.Context, url string, opts *WSDialOptions) (*WSTransport, error)

func (t *WSTransport) Close() error

func (t *WSTransport) Recv(ctx context.Context) (Message, error)

func (t *WSTransport) Send(ctx context.Context, msg Message) error

type WritableExpr ¶

type WritableExpr struct{ ExportID int64 }

WritableExpr references a writable stream — ["writable", exportId].