gorpc

package module
v0.5.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jul 6, 2026 License: MIT Imports: 16 Imported by: 0

README

GoRPC

Go Reference Go Report Card CI

Small Go-to-Go RPC for internal service calls.

This is meant to keep the useful shape of net/rpc without inheriting gob as the default wire choice or dragging in protobuf, schema files, generated stubs, duplicate DTO models, service discovery, load balancing, or cross-language ceremony.

Current Scope

  • TCP listener/client
  • Long-lived full-duplex peer connection after the client dials and the server accepts
  • Automatic client reconnect with exponential backoff
  • Ping/pong connection monitoring
  • Length-prefixed MessagePack frames
  • Shared Go request/response structs
  • Synchronous and asynchronous unary request/response calls from either side
  • One-way notifications/push messages from either side
  • Server streaming, client streaming, and bidirectional streaming from either side
  • Request IDs in every request, response, notification, and stream frame
  • Context deadline propagation and best-effort cancel frames
  • Client-side correlation IDs for asynchronous callbacks
  • Message-scoped *gorpc.Context with client name, request/notification ID, function, and connection addresses
  • Structured remote errors
  • Max frame size enforcement
  • Basic protocol/version/codec handshake with optional client name metadata
  • Optional HMAC-SHA256 shared-secret handshake auth
  • Optional slog debug hooks
  • Graceful server shutdown

Server.ServeTCP, Server.ServeUnix, and Server.ServeUnixPacket cover the common listener cases. Server.ServeListener accepts any existing net.Listener.

The words server and client only describe who accepts the connection and who initiates it. Once connected, both sides can register functions, send requests, receive responses, send one-way notifications, open streams, and handle incoming messages over the same full-duplex connection.

TCPDial, UnixDial, and UnixPacketDial establish the first connection, then the returned client keeps monitoring and reconnecting until Close is called. Reconnect attempts are intentionally aggressive: quick retry, exponential backoff capped at seconds, jitter, explicit dial timeouts, write deadlines, and ping/pong stale-connection detection. The lower-level Dial accepts a context, network, address, and full ClientOptions when you need explicit startup control. Use NewTCPClient, NewUnixClient, or NewUnixPacketClient when the dialing side needs to register functions before connecting. Client.Call, Client.CallWithTimeout, and Client.CallContext cover synchronous calls. Client.AsyncCall sends the request and invokes a typed callback when the response arrives. Client.Notify sends a one-way message and returns after the frame is written locally. Calls made while disconnected wait for the next connection; timeout/context variants bound that wait. Calls and streams already in flight when a connection drops fail with ErrUnavailable; GoRPC does not silently replay them because the remote side may already have processed the request or some stream items.

Service discovery, pub/sub, load balancing, and generated code are intentionally out of v1.

Shared-secret auth is optional. The secret is not sent over the wire; GoRPC uses a handshake challenge and HMAC-SHA256 proof.

auth := gorpc.SharedSecret("change-me")

server := gorpc.NewServer(gorpc.ServerOptions{
	Auth: auth,
})

client, err := gorpc.TCPDial("127.0.0.1:9070", "inventory-example-client", gorpc.ClientOptions{
	Auth: auth,
})

Install

go get github.com/dan-sherwin/gorpc

Example

The types are shown inline here to keep the example self-contained. In a real app, put them in a shared Go package imported by both sides. The function string is just the wire dispatch name; it does not have to match the local Go function name.

Server app:

package main

import (
	"log"

	"github.com/dan-sherwin/gorpc"
)

type GetItemRequest struct {
	ID string
}

type GetItemResponse struct {
	ID   string
	Name string
}

type ClientNote struct {
	ItemID string
}

func getItem(ctx *gorpc.Context, req GetItemRequest) (GetItemResponse, error) {
	log.Printf("handling %s request_id=%d client=%q remote=%s",
		ctx.Function(),
		ctx.RequestID(),
		ctx.ClientName(),
		ctx.RemoteAddr(),
	)

	if err := ctx.Notify("client_note", ClientNote{ItemID: req.ID}); err != nil {
		log.Printf("client notification failed: %v", err)
	}

	return GetItemResponse{
		ID:   req.ID,
		Name: "Widget Pack",
	}, nil
}

func main() {
	server := gorpc.NewServer(gorpc.ServerOptions{})

	gorpc.MustRegister(server, "get_an_item", getItem)

	log.Println("listening on 127.0.0.1:9070")
	if err := server.ServeTCP("127.0.0.1:9070"); err != nil {
		log.Fatal(err)
	}
}

Client app:

package main

import (
	"context"
	"fmt"
	"log"
	"time"

	"github.com/dan-sherwin/gorpc"
)

type GetItemRequest struct {
	ID string
}

type GetItemResponse struct {
	ID   string
	Name string
}

type ClientNote struct {
	ItemID string
}

func main() {
	client := gorpc.NewTCPClient("127.0.0.1:9070", "inventory-example-client")
	gorpc.MustRegisterNotify(client, "client_note", clientNote)

	if err := client.Connect(context.Background()); err != nil {
		log.Fatal(err)
	}
	defer func() {
		_ = client.Close()
	}()

	var item GetItemResponse
	if err := client.CallWithTimeout("get_an_item", GetItemRequest{ID: "widget-001"}, &item, 5*time.Second); err != nil {
		log.Fatal(err)
	}

	fmt.Printf("%s: %s\n", item.ID, item.Name)
}

func clientNote(_ *gorpc.Context, note ClientNote) error {
	fmt.Println("server push: client saw request for", note.ItemID)
	return nil
}

Async calls use the same request/response structs and add a callback plus a caller-owned correlation ID:

func handleGetItem(ctx gorpc.ClientContext, resp *GetItemResponse) {
	if ctx.Error() != nil {
		log.Fatal(ctx.Error())
	}

	fmt.Printf("async %s: %s: %s\n", ctx.CorrelationID(), resp.ID, resp.Name)
}

if err := client.AsyncCall("get_an_item", GetItemRequest{ID: "widget-async"}, handleGetItem, "example-async-1"); err != nil {
	log.Fatal(err)
}

One-way notifications use RegisterNotify and Notify. A notify sender only learns whether the frame was written locally; it does not receive a remote success/error response.

type ItemChanged struct {
	ID string
}

gorpc.MustRegisterNotify(server, "item_changed", func(ctx *gorpc.Context, msg ItemChanged) error {
	log.Println("item changed", msg.ID)
	return nil
})

if err := client.Notify("item_changed", ItemChanged{ID: "widget-001"}); err != nil {
	log.Fatal(err)
}

For server-initiated calls outside an existing request handler, use ServerOptions.OnConnect or server.Connections() to get a *gorpc.Conn, then call conn.Call, conn.CallWithTimeout, or conn.AsyncCall.

Streaming

GoRPC supports the three normal stream shapes without generated stubs:

  • RegisterServerStream + ServerStream: one request in, many items out.
  • RegisterClientStream + ClientStream: many items in, one response out.
  • RegisterBidiStream + BidiStream: both sides send and receive items.

The names describe the stream shape, not which process accepted the socket. Either side can register stream handlers, and either side can open streams. Use a *gorpc.Client when the dialing side opens a stream. Use a *gorpc.Conn when the accepted side opens a stream back to the dialing side.

Server streaming:

type ListItemsRequest struct {
	Prefix string
	Count  int
}

type ItemEvent struct {
	Value string
}

gorpc.MustRegisterServerStream(server, "list_items", func(ctx *gorpc.Context, req ListItemsRequest, stream *gorpc.StreamWriter[ItemEvent]) error {
	for i := 1; i <= req.Count; i++ {
		if err := stream.Send(ItemEvent{Value: fmt.Sprintf("%s-%d", req.Prefix, i)}); err != nil {
			return err
		}
	}
	return nil
})

reader, err := gorpc.ServerStream[ListItemsRequest, ItemEvent](context.Background(), client, "list_items", ListItemsRequest{
	Prefix: "widget",
	Count:  3,
})
if err != nil {
	log.Fatal(err)
}

for {
	item, err := reader.Recv()
	if errors.Is(err, io.EOF) {
		break
	}
	if err != nil {
		log.Fatal(err)
	}
	fmt.Println(item.Value)
}

Client streaming:

type UploadSummary struct {
	Count int
}

gorpc.MustRegisterClientStream(server, "upload_items", func(ctx *gorpc.Context, reader *gorpc.StreamReader[ItemEvent]) (UploadSummary, error) {
	count := 0
	for {
		item, err := reader.Recv()
		if errors.Is(err, io.EOF) {
			return UploadSummary{Count: count}, nil
		}
		if err != nil {
			return UploadSummary{}, err
		}
		_ = item
		count++
	}
})

stream, err := gorpc.ClientStream[ItemEvent, UploadSummary](context.Background(), client, "upload_items")
if err != nil {
	log.Fatal(err)
}

_ = stream.Send(ItemEvent{Value: "alpha"})
_ = stream.Send(ItemEvent{Value: "bravo"})

summary, err := stream.CloseAndRecv()
if err != nil {
	log.Fatal(err)
}
fmt.Println(summary.Count)

Bidirectional streaming:

gorpc.MustRegisterBidiStream(server, "echo_items", func(ctx *gorpc.Context, stream *gorpc.BidiStreamHandle[ItemEvent, ItemEvent]) error {
	for {
		item, err := stream.Recv()
		if errors.Is(err, io.EOF) {
			return nil
		}
		if err != nil {
			return err
		}
		if err := stream.Send(ItemEvent{Value: strings.ToUpper(item.Value)}); err != nil {
			return err
		}
	}
})

stream, err := gorpc.BidiStream[ItemEvent, ItemEvent](context.Background(), client, "echo_items")
if err != nil {
	log.Fatal(err)
}

_ = stream.Send(ItemEvent{Value: "alpha"})
reply, err := stream.Recv()
if err != nil {
	log.Fatal(err)
}
fmt.Println(reply.Value)

_ = stream.CloseSend()

Server-initiated streaming uses the same helpers with a *gorpc.Conn:

server := gorpc.NewServer(gorpc.ServerOptions{
	OnConnect: func(conn *gorpc.Conn) {
		reader, err := gorpc.ServerStream[ListItemsRequest, ItemEvent](context.Background(), conn, "client_list_items", ListItemsRequest{
			Prefix: "client",
			Count:  2,
		})
		if err != nil {
			log.Println(err)
			return
		}
		for {
			item, err := reader.Recv()
			if errors.Is(err, io.EOF) {
				return
			}
			if err != nil {
				log.Println(err)
				return
			}
			log.Println(item.Value)
		}
	},
})

Streaming rules:

  • Recv returns io.EOF after the remote side closes cleanly.
  • CloseSend half-closes the local send side. It does not stop receiving.
  • Cancel sends a best-effort cancel frame and ends the whole stream locally.
  • Each stream item is one GoRPC frame and must fit MaxFrameSize; the default is 64 MiB per frame.
  • Streaming avoids building one huge response, but it does not bypass the per-frame limit.
  • If the connection breaks mid-stream, the active stream fails with ErrUnavailable.
  • The client keeps reconnecting after a break. New calls and new streams can use the new connection.
  • GoRPC does not replay active streams after reconnect. Retrying is application logic because the other side may already have processed some items.

For the deeper implementation guide, see docs/streaming.md.

Runnable Example

A working server/client example with sync calls, async callbacks, and structured remote errors lives under examples/inventory.

Build both commands:

go build ./examples/inventory/server
go build ./examples/inventory/client

Run the server:

go run ./examples/inventory/server

Run the client in another terminal:

go run ./examples/inventory/client

License

MIT. See LICENSE.

Versioning

Semantic Versioning. First public tag: v0.1.0.

CI expectations

  • go mod tidy check
  • go build ./...
  • go vet ./...
  • go test ./... -race
  • golangci-lint run
  • govulncheck ./...

Supported Go version: 1.26.3+.

Documentation

Overview

Package gorpc provides a small Go-to-Go RPC transport for internal services.

It is intentionally not a protobuf, gRPC, Connect, or IDL replacement. Both sides share normal Go request and response types, and the wire protocol uses length-prefixed MessagePack frames over a single full-duplex connection. Once connected, either side can send unary requests, receive responses, send one-way notifications, and open server-streaming, client-streaming, or bidirectional-streaming calls.

The dialing Client reconnects aggressively after network loss. Calls and streams already in flight fail with ErrUnavailable instead of being replayed, because the remote peer may already have processed the request or some stream items. New calls and new streams can use the re-established connection.

Index

Constants

View Source 
const (
	DefaultDialTimeout       = 5 * time.Second
	DefaultWriteTimeout      = 10 * time.Second
	DefaultReconnectMinDelay = 100 * time.Millisecond
	DefaultReconnectMaxDelay = 5 * time.Second
	DefaultReconnectJitter   = 0.2
	DefaultPingInterval      = 10 * time.Second
	DefaultPingTimeout       = 3 * time.Second
)

Reconnect defaults used by Client when options are unset.

View Source 
const (
	ErrorCodeCanceled         = "canceled"
	ErrorCodeDeadlineExceeded = "deadline_exceeded"
	ErrorCodeInternal         = "internal"
	ErrorCodeInvalidRequest   = "invalid_request"
	ErrorCodeNotFound         = "not_found"
	ErrorCodeUnauthorized     = "unauthorized"
	ErrorCodeUnavailable      = "unavailable"
)

Remote error codes used by the built-in server and helpers.

View Source 
const CodecMessagePack = "msgpack"

CodecMessagePack is the v1 MessagePack codec name used during handshake.

View Source 
const DefaultHandshakeTimeout = 5 * time.Second

DefaultHandshakeTimeout is the default timeout for the initial protocol handshake.

View Source 
const DefaultMaxFrameSize int64 = 64 * 1024 * 1024

DefaultMaxFrameSize is the default maximum encoded frame size.

View Source 
const ProtocolVersion uint16 = 1

ProtocolVersion is the current GoRPC wire protocol version.

Variables

View Source 
var (
	ErrClosed            = errors.New("gorpc: closed")
	ErrAuthentication    = errors.New("gorpc: authentication failed")
	ErrDuplicateFunction = errors.New("gorpc: duplicate function")
	ErrInvalidFunction   = errors.New("gorpc: invalid function")
	ErrInvalidHandler    = errors.New("gorpc: invalid handler")
	ErrInvalidResponse   = errors.New("gorpc: invalid response")
	ErrUnavailable       = errors.New("gorpc: unavailable")
)

Common GoRPC errors.

View Source 
var (
	ErrFrameTooLarge = errors.New("gorpc: frame too large")
	ErrProtocol      = errors.New("gorpc: protocol error")
)

Frame read/write errors.

Functions

func Call 

func Call[Req, Resp any](ctx context.Context, client *Client, function string, req Req) (Resp, error)

Call performs a typed unary request/response call.

func MustRegister 

func MustRegister[Req, Resp any](target any, function string, fn HandlerFunc[Req, Resp])

MustRegister is Register that panics on error.

func MustRegisterBidiStream added in v0.5.0 

func MustRegisterBidiStream[Recv, Send any](target any, function string, fn BidiStreamHandlerFunc[Recv, Send])

MustRegisterBidiStream is RegisterBidiStream that panics on error.

func MustRegisterClientStream added in v0.5.0 

func MustRegisterClientStream[Item, Resp any](target any, function string, fn ClientStreamHandlerFunc[Item, Resp])

MustRegisterClientStream is RegisterClientStream that panics on error.

func MustRegisterNotify added in v0.4.0 

func MustRegisterNotify[Req any](target any, function string, fn NotifyHandlerFunc[Req])

MustRegisterNotify is RegisterNotify that panics on error.

func MustRegisterServerStream added in v0.5.0 

func MustRegisterServerStream[Req, Item any](target any, function string, fn ServerStreamHandlerFunc[Req, Item])

MustRegisterServerStream is RegisterServerStream that panics on error.

func Notify added in v0.4.0 

func Notify[Req any](ctx context.Context, client *Client, function string, req Req) error

Notify sends a typed one-way notification.

func Register 

func Register[Req, Resp any](target any, function string, fn HandlerFunc[Req, Resp]) error

Register binds a typed unary handler to a function name. The target can be a *Server, for functions the accepted side handles, or a *Client, for functions the dialing side handles after the connection is established.

func RegisterBidiStream added in v0.5.0 

func RegisterBidiStream[Recv, Send any](target any, function string, fn BidiStreamHandlerFunc[Recv, Send]) error

RegisterBidiStream binds a typed bidirectional-streaming handler to a function name. Both sides can send and receive stream items.

func RegisterClientStream added in v0.5.0 

func RegisterClientStream[Item, Resp any](target any, function string, fn ClientStreamHandlerFunc[Item, Resp]) error

RegisterClientStream binds a typed client-streaming handler to a function name. The caller sends zero or more request items and receives one response.

func RegisterNotify added in v0.4.0 

func RegisterNotify[Req any](target any, function string, fn NotifyHandlerFunc[Req]) error

RegisterNotify binds a typed one-way notification handler to a function name. The sender gets write success/failure only; handler errors are local to the receiver.

func RegisterServerStream added in v0.5.0 

func RegisterServerStream[Req, Item any](target any, function string, fn ServerStreamHandlerFunc[Req, Item]) error

RegisterServerStream binds a typed server-streaming handler to a function name. The caller sends one request and receives zero or more response items.

Types

type Auth added in v0.2.0 

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

Auth configures optional connection authentication.

func SharedSecret added in v0.2.0 

func SharedSecret(secret string) Auth

SharedSecret enables HMAC-SHA256 challenge/response authentication.

type BidiStreamHandle added in v0.5.0 

type BidiStreamHandle[Send, Recv any] struct {
	// contains filtered or unexported fields
}

BidiStreamHandle is a typed bidirectional stream wrapper. Send and Recv can be used concurrently by different goroutines.

func BidiStream added in v0.5.0 

func BidiStream[Send, Recv any](ctx context.Context, target any, function string) (*BidiStreamHandle[Send, Recv], error)

BidiStream opens a bidirectional stream. Send and Recv can be used concurrently by different goroutines. Recv returns io.EOF when the remote send side closes cleanly.

The target can be either *Client or an accepted *Conn, so either connected side can open a stream to the other side.

func (*BidiStreamHandle[Send, Recv]) Cancel added in v0.5.0 

func (s *BidiStreamHandle[Send, Recv]) Cancel() error

Cancel cancels the stream and sends a best-effort cancel frame.

func (*BidiStreamHandle[Send, Recv]) CloseSend added in v0.5.0 

func (s *BidiStreamHandle[Send, Recv]) CloseSend() error

CloseSend closes the local sending side of the stream.

func (*BidiStreamHandle[Send, Recv]) Recv added in v0.5.0 

func (s *BidiStreamHandle[Send, Recv]) Recv() (Recv, error)

Recv receives one typed stream item. It returns io.EOF when the remote side cleanly closes its send side.

func (*BidiStreamHandle[Send, Recv]) Send added in v0.5.0 

func (s *BidiStreamHandle[Send, Recv]) Send(item Send) error

Send sends one typed stream item.

func (*BidiStreamHandle[Send, Recv]) Stream added in v0.5.0 

func (s *BidiStreamHandle[Send, Recv]) Stream() *Stream

Stream returns the raw stream.

type BidiStreamHandlerFunc added in v0.5.0 

type BidiStreamHandlerFunc[Recv, Send any] func(*Context, *BidiStreamHandle[Send, Recv]) error

BidiStreamHandlerFunc receives and sends stream items until either side closes its sending direction.

type Client 

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

Client is the dialing side of a long-lived full-duplex GoRPC connection. It can send requests, register functions for the accepted side to call, and reconnects automatically after connection loss until Close is called.

func Dial 

func Dial(ctx context.Context, network, address string, opts ClientOptions) (*Client, error)

Dial connects to a GoRPC server, completes the protocol handshake, and starts background reconnect monitoring.

func NewClient added in v0.3.0 

func NewClient(network, address string, opts ClientOptions) *Client

NewClient creates a client without connecting it. Use this when the dialing side needs to register functions before the accepted side can call them.

func NewTCPClient added in v0.3.0 

func NewTCPClient(address, clientName string, opts ...ClientOptions) *Client

NewTCPClient creates a TCP client without connecting it.

func NewUnixClient added in v0.3.0 

func NewUnixClient(path, clientName string, opts ...ClientOptions) *Client

NewUnixClient creates a Unix socket client without connecting it.

func NewUnixPacketClient added in v0.3.0 

func NewUnixPacketClient(path, clientName string, opts ...ClientOptions) *Client

NewUnixPacketClient creates a Unix packet socket client without connecting it.

func TCPDial added in v0.2.0 

func TCPDial(address, clientName string, opts ...ClientOptions) (*Client, error)

TCPDial connects to address using TCP and reconnects automatically until Close is called.

func UnixDial added in v0.2.0 

func UnixDial(path, clientName string, opts ...ClientOptions) (*Client, error)

UnixDial connects to path using a Unix socket and reconnects automatically until Close is called.

func UnixPacketDial added in v0.2.0 

func UnixPacketDial(path, clientName string, opts ...ClientOptions) (*Client, error)

UnixPacketDial connects to path using a Unix packet socket and reconnects automatically until Close is called.

func (*Client) AsyncCall added in v0.2.0 

func (c *Client) AsyncCall(function string, req any, handler any, correlationID string) error

AsyncCall sends a unary request and invokes handler when the response arrives.

func (*Client) AsyncCallContext added in v0.2.0 

func (c *Client) AsyncCallContext(ctx context.Context, function string, req any, handler any, correlationID string) error

AsyncCallContext sends a unary request and invokes handler when the response arrives. The context only controls waiting for a connection and writing the request frame.

func (*Client) AsyncCallWithTimeout added in v0.2.0 

func (c *Client) AsyncCallWithTimeout(function string, req any, handler any, correlationID string, timeout time.Duration) error

AsyncCallWithTimeout sends a unary request using a timeout while waiting for a connection and writing the request frame. The response handler runs later.

func (*Client) Call added in v0.2.0 

func (c *Client) Call(function string, req any, resp any) error

Call performs a unary request/response call using context.Background.

func (*Client) CallContext added in v0.2.0 

func (c *Client) CallContext(ctx context.Context, function string, req any, resp any) error

CallContext performs a unary request/response call. If the client is reconnecting, CallContext waits for the next connection until ctx is canceled.

func (*Client) CallWithTimeout added in v0.2.0 

func (c *Client) CallWithTimeout(function string, req any, resp any, timeout time.Duration) error

CallWithTimeout performs a unary request/response call with a timeout.

func (*Client) Close 

func (c *Client) Close() error

Close closes the client and stops reconnect attempts.

func (*Client) Connect added in v0.3.0 

func (c *Client) Connect(ctx context.Context) error

Connect establishes the first connection and starts background reconnect monitoring. It is called automatically by Dial and the TCPDial helpers.

func (*Client) Notify added in v0.4.0 

func (c *Client) Notify(function string, req any) error

Notify sends a one-way typed notification using context.Background.

func (*Client) NotifyContext added in v0.4.0 

func (c *Client) NotifyContext(ctx context.Context, function string, req any) error

NotifyContext sends a one-way typed notification. Success means the frame was written locally; GoRPC does not wait for remote handler completion or remote errors.

func (*Client) NotifyWithTimeout added in v0.4.0 

func (c *Client) NotifyWithTimeout(function string, req any, timeout time.Duration) error

NotifyWithTimeout sends a one-way typed notification with a timeout while waiting for a connection and writing the notification frame.

func (*Client) WaitReady added in v0.2.0 

func (c *Client) WaitReady(ctx context.Context) error

WaitReady blocks until the client has an active connection or ctx is canceled.

type ClientContext added in v0.2.0 

type ClientContext interface {
	CorrelationID() string
	RequestID() uint64
	Function() string
	Error() error
}

ClientContext is passed to asynchronous response handlers for requests made by either a Client or an accepted Conn.

type ClientFunc added in v0.2.0 

type ClientFunc[Req, Resp any] func(context.Context, Req) (Resp, error)

ClientFunc is the typed function shape returned by Function.

func Function added in v0.2.0 

func Function[Req, Resp any](client *Client, function string) ClientFunc[Req, Resp]

Function returns a typed client function bound to a remote function name.

type ClientOptions 

type ClientOptions struct {
	ClientName       string
	Codec            Codec
	MaxFrameSize     int64
	HandshakeTimeout time.Duration
	Auth             Auth
	DialTimeout      time.Duration
	WriteTimeout     time.Duration
	Logger           *slog.Logger
	Dialer           *net.Dialer

	ReconnectMinDelay time.Duration
	ReconnectMaxDelay time.Duration
	ReconnectJitter   float64
	PingInterval      time.Duration
	PingTimeout       time.Duration
}

ClientOptions configures Dial and the network-specific dial helpers.

type ClientStreamHandle added in v0.5.0 

type ClientStreamHandle[Item, Resp any] struct {
	// contains filtered or unexported fields
}

ClientStreamHandle is returned by ClientStream. It lets the caller send many request items and then receive one final response.

func ClientStream added in v0.5.0 

func ClientStream[Item, Resp any](ctx context.Context, target any, function string) (*ClientStreamHandle[Item, Resp], error)

ClientStream opens a client-streaming call. The caller sends zero or more typed items and then calls CloseAndRecv for the final typed response.

The target can be either *Client or an accepted *Conn, so either connected side can open a stream to the other side.

func (*ClientStreamHandle[Item, Resp]) Cancel added in v0.5.0 

func (s *ClientStreamHandle[Item, Resp]) Cancel() error

Cancel cancels the stream and sends a best-effort cancel frame.

func (*ClientStreamHandle[Item, Resp]) CloseAndRecv added in v0.5.0 

func (s *ClientStreamHandle[Item, Resp]) CloseAndRecv() (Resp, error)

CloseAndRecv closes the local sending side and waits for the final typed response.

func (*ClientStreamHandle[Item, Resp]) Send added in v0.5.0 

func (s *ClientStreamHandle[Item, Resp]) Send(item Item) error

Send sends one typed request item.

func (*ClientStreamHandle[Item, Resp]) Stream added in v0.5.0 

func (s *ClientStreamHandle[Item, Resp]) Stream() *Stream

Stream returns the raw stream.

type ClientStreamHandlerFunc added in v0.5.0 

type ClientStreamHandlerFunc[Item, Resp any] func(*Context, *StreamReader[Item]) (Resp, error)

ClientStreamHandlerFunc receives zero or more request items and returns one final response.

type Codec 

type Codec interface {
	Name() string
	Marshal(v any) ([]byte, error)
	Unmarshal(data []byte, v any) error
}

Codec marshals frame envelopes and function payloads.

type Conn added in v0.3.0 

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

Conn is one accepted GoRPC connection. A Conn can receive requests through server-registered functions and can also initiate requests back to the client over the same full-duplex connection.

func (*Conn) AsyncCall added in v0.3.0 

func (c *Conn) AsyncCall(function string, req any, handler any, correlationID string) error

AsyncCall sends a unary request to the connected client and invokes handler when the response arrives.

func (*Conn) AsyncCallContext added in v0.3.0 

func (c *Conn) AsyncCallContext(ctx context.Context, function string, req any, handler any, correlationID string) error

AsyncCallContext sends a unary request to the connected client and invokes handler when the response arrives.

func (*Conn) AsyncCallWithTimeout added in v0.3.0 

func (c *Conn) AsyncCallWithTimeout(function string, req any, handler any, correlationID string, timeout time.Duration) error

AsyncCallWithTimeout sends a unary request to the connected client using a timeout while writing the request frame. The response handler runs later.

func (*Conn) Call added in v0.3.0 

func (c *Conn) Call(function string, req any, resp any) error

Call performs a unary request/response call to the connected client.

func (*Conn) CallContext added in v0.3.0 

func (c *Conn) CallContext(ctx context.Context, function string, req any, resp any) error

CallContext performs a unary request/response call to the connected client.

func (*Conn) CallWithTimeout added in v0.3.0 

func (c *Conn) CallWithTimeout(function string, req any, resp any, timeout time.Duration) error

CallWithTimeout performs a unary request/response call to the connected client with a timeout.

func (*Conn) ClientName added in v0.3.0 

func (c *Conn) ClientName() string

ClientName returns the self-reported client name from the connection handshake. It is useful for logs and metrics, but is not authenticated.

func (*Conn) Close added in v0.3.0 

func (c *Conn) Close() error

Close closes the accepted connection, cancels active inbound handlers, and fails pending outbound calls.

func (*Conn) LocalAddr added in v0.3.0 

func (c *Conn) LocalAddr() net.Addr

LocalAddr returns the local address for the connection.

func (*Conn) Notify added in v0.4.0 

func (c *Conn) Notify(function string, req any) error

Notify sends a one-way typed notification to the connected client.

func (*Conn) NotifyContext added in v0.4.0 

func (c *Conn) NotifyContext(ctx context.Context, function string, req any) error

NotifyContext sends a one-way typed notification to the connected client. Success means the frame was written locally; GoRPC does not wait for remote handler completion or remote errors.

func (*Conn) NotifyWithTimeout added in v0.4.0 

func (c *Conn) NotifyWithTimeout(function string, req any, timeout time.Duration) error

NotifyWithTimeout sends a one-way typed notification to the connected client with a timeout while writing the notification frame.

func (*Conn) RemoteAddr added in v0.3.0 

func (c *Conn) RemoteAddr() net.Addr

RemoteAddr returns the peer address for the connection.

type Context added in v0.2.0 

type Context struct {
	context.Context
	// contains filtered or unexported fields
}

Context is the message-scoped context passed to request and notification handlers.

func (*Context) Call added in v0.3.0 

func (c *Context) Call(function string, req any, resp any) error

Call performs a unary request/response call back over the same accepted connection that delivered this request.

func (*Context) CallContext added in v0.3.0 

func (c *Context) CallContext(ctx context.Context, function string, req any, resp any) error

CallContext performs a unary request/response call back over the same accepted connection.

func (*Context) CallWithTimeout added in v0.3.0 

func (c *Context) CallWithTimeout(function string, req any, resp any, timeout time.Duration) error

CallWithTimeout performs a unary request/response call back over the same accepted connection with a timeout.

func (*Context) ClientName added in v0.2.0 

func (c *Context) ClientName() string

ClientName returns the self-reported client name from the connection handshake. It is useful for logs and metrics, but is not authenticated.

func (*Context) Conn added in v0.3.0 

func (c *Context) Conn() *Conn

Conn returns the accepted connection that delivered this request when the handler is running on a Server. Client-side handlers return nil here because they can already call back through their Client.

func (*Context) Function added in v0.2.0 

func (c *Context) Function() string

Function returns the remote function name for the request.

func (*Context) IsNotify added in v0.4.0 

func (c *Context) IsNotify() bool

IsNotify reports whether the inbound message is a one-way notification.

func (*Context) IsStream added in v0.5.0 

func (c *Context) IsStream() bool

IsStream reports whether the inbound message opened a stream.

func (*Context) LocalAddr added in v0.2.0 

func (c *Context) LocalAddr() net.Addr

LocalAddr returns the local address for the connection.

func (*Context) Notify added in v0.4.0 

func (c *Context) Notify(function string, req any) error

Notify sends a one-way typed notification back over the same accepted connection that delivered this request.

func (*Context) NotifyContext added in v0.4.0 

func (c *Context) NotifyContext(ctx context.Context, function string, req any) error

NotifyContext sends a one-way typed notification back over the same accepted connection.

func (*Context) NotifyWithTimeout added in v0.4.0 

func (c *Context) NotifyWithTimeout(function string, req any, timeout time.Duration) error

NotifyWithTimeout sends a one-way typed notification back over the same accepted connection with a timeout while writing the notification frame.

func (*Context) RemoteAddr added in v0.2.0 

func (c *Context) RemoteAddr() net.Addr

RemoteAddr returns the peer address for the connection.

func (*Context) RequestID added in v0.2.0 

func (c *Context) RequestID() uint64

RequestID returns the request or notification ID from the GoRPC frame.

func (*Context) StreamKind added in v0.5.0 

func (c *Context) StreamKind() StreamKind

StreamKind returns the stream shape for streaming handlers. For non-stream handlers it returns zero.

type Frame 

type Frame struct {
	Version          uint16     `msgpack:"version"`
	Type             FrameType  `msgpack:"type"`
	RequestID        uint64     `msgpack:"request_id,omitempty"`
	Function         string     `msgpack:"function,omitempty"`
	StreamKind       StreamKind `msgpack:"stream_kind,omitempty"`
	DeadlineUnixNano int64      `msgpack:"deadline_unix_nano,omitempty"`
	Payload          []byte     `msgpack:"payload,omitempty"`
}

Frame is the v1 wire envelope. It is MessagePack-encoded and written with a 4-byte big-endian length prefix.

type FrameType 

type FrameType uint8

FrameType identifies the kind of message carried by a frame. 

const (
	FrameHello FrameType = iota + 1
	FrameHelloAck
	FrameRequest
	FrameResponse
	FrameError
	FrameCancel
	FramePing
	FramePong
	FrameStreamItem
	FrameStreamEnd
	FrameAuth
	FrameAuthAck
	FrameNotify
	FrameStreamStart
)

Frame types used by the v1 protocol.

func (FrameType) String 

func (t FrameType) String() string

type HandlerFunc 

type HandlerFunc[Req, Resp any] func(*Context, Req) (Resp, error)

HandlerFunc is the typed function shape used by registered unary functions.

type MessagePackCodec 

type MessagePackCodec struct{}

MessagePackCodec is the default v1 codec.

func (MessagePackCodec) Marshal 

func (MessagePackCodec) Marshal(v any) ([]byte, error)

Marshal encodes v as MessagePack.

func (MessagePackCodec) Name 

func (MessagePackCodec) Name() string

Name returns the handshake name for MessagePackCodec.

func (MessagePackCodec) Unmarshal 

func (MessagePackCodec) Unmarshal(data []byte, v any) error

Unmarshal decodes MessagePack data into v.

type NotifyFunc added in v0.4.0 

type NotifyFunc[Req any] func(context.Context, Req) error

NotifyFunc is the typed function shape returned by Notification.

func Notification added in v0.4.0 

func Notification[Req any](client *Client, function string) NotifyFunc[Req]

Notification returns a typed client notification function bound to a remote function name.

type NotifyHandlerFunc added in v0.4.0 

type NotifyHandlerFunc[Req any] func(*Context, Req) error

NotifyHandlerFunc is the typed function shape used by registered one-way notification handlers.

type RemoteError 

type RemoteError struct {
	Code    string         `msgpack:"code" json:"code"`
	Message string         `msgpack:"message" json:"message"`
	Details map[string]any `msgpack:"details,omitempty" json:"details,omitempty"`
}

RemoteError is sent in FrameError payloads and returned by callers when the server handled the request but rejected or failed it.

func NewRemoteError 

func NewRemoteError(code, message string, details map[string]any) *RemoteError

NewRemoteError creates a structured error suitable for returning from a handler.

func (*RemoteError) Error 

func (e *RemoteError) Error() string

type Server 

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

Server accepts GoRPC connections, dispatches registered functions, and exposes accepted connections that can initiate requests back to the dialing side.

func NewServer 

func NewServer(opts ServerOptions) *Server

NewServer creates a Server with default codec and limits where options are unset.

func (*Server) Connections added in v0.3.0 

func (s *Server) Connections() []*Conn

Connections returns a snapshot of currently accepted connections.

func (*Server) ServeListener added in v0.2.0 

func (s *Server) ServeListener(ln net.Listener) error

ServeListener accepts GoRPC connections from ln until Shutdown is called or the listener returns an unrecoverable error.

func (*Server) ServeTCP added in v0.2.0 

func (s *Server) ServeTCP(address string) error

ServeTCP listens on address with the "tcp" network and serves GoRPC connections.

func (*Server) ServeUnix added in v0.2.0 

func (s *Server) ServeUnix(path string) error

ServeUnix listens on path with the "unix" network and serves GoRPC connections.

func (*Server) ServeUnixPacket added in v0.2.0 

func (s *Server) ServeUnixPacket(path string) error

ServeUnixPacket listens on path with the "unixpacket" network and serves GoRPC connections.

func (*Server) Shutdown 

func (s *Server) Shutdown(ctx context.Context) error

Shutdown closes the listener, closes active connections, and waits for handlers to exit.

type ServerOptions 

type ServerOptions struct {
	Codec            Codec
	MaxFrameSize     int64
	HandshakeTimeout time.Duration
	Auth             Auth
	WriteTimeout     time.Duration
	Logger           *slog.Logger
	OnConnect        func(*Conn)
	OnDisconnect     func(*Conn)
}

ServerOptions configures a GoRPC server.

type ServerStreamHandlerFunc added in v0.5.0 

type ServerStreamHandlerFunc[Req, Item any] func(*Context, Req, *StreamWriter[Item]) error

ServerStreamHandlerFunc handles one request and sends zero or more response items before returning.

type Stream added in v0.5.0 

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

Stream is the raw bidirectional item stream used by the typed streaming helpers. Most callers should prefer ServerStream, ClientStream, BidiStream, and the typed handler registration functions.

func (*Stream) Cancel added in v0.5.0 

func (s *Stream) Cancel() error

Cancel cancels the whole stream and sends a best-effort cancel frame to the remote side.

func (*Stream) CloseSend added in v0.5.0 

func (s *Stream) CloseSend() error

CloseSend closes the local sending side of the stream with a stream_end frame. It does not cancel receiving items from the remote side.

func (*Stream) Context added in v0.5.0 

func (s *Stream) Context() context.Context

Context returns the stream context. It is canceled when the stream is locally canceled, the connection closes, or a remote stream error is received.

func (*Stream) Function added in v0.5.0 

func (s *Stream) Function() string

Function returns the remote function name for the stream.

func (*Stream) Recv added in v0.5.0 

func (s *Stream) Recv(item any) error

Recv reads one stream item into item. It returns io.EOF after the remote side closes its send side with a stream_end frame.

func (*Stream) RequestID added in v0.5.0 

func (s *Stream) RequestID() uint64

RequestID returns the stream request ID.

func (*Stream) Send added in v0.5.0 

func (s *Stream) Send(item any) error

Send writes one stream item. The item is MessagePack-encoded into a FrameStreamItem payload.

type StreamKind added in v0.5.0 

type StreamKind uint8

StreamKind identifies the shape of a streaming function. 

const (
	// StreamKindServer means the caller sends one request and the handler sends zero or more items.
	StreamKindServer StreamKind = iota + 1
	// StreamKindClient means the caller sends zero or more items and the handler sends one response.
	StreamKindClient
	// StreamKindBidi means both sides can send and receive stream items.
	StreamKindBidi
)

func (StreamKind) String added in v0.5.0 

func (k StreamKind) String() string

type StreamReader added in v0.5.0 

type StreamReader[T any] struct {
	// contains filtered or unexported fields
}

StreamReader is a typed receive-only stream wrapper.

func ServerStream added in v0.5.0 

func ServerStream[Req, Item any](ctx context.Context, target any, function string, req Req) (*StreamReader[Item], error)

ServerStream opens a server-streaming call. The caller sends one request and receives zero or more typed items until Recv returns io.EOF or an error.

The target can be either *Client or an accepted *Conn, so either connected side can open a stream to the other side.

func (*StreamReader[T]) Cancel added in v0.5.0 

func (r *StreamReader[T]) Cancel() error

Cancel cancels the stream and sends a best-effort cancel frame.

func (*StreamReader[T]) Recv added in v0.5.0 

func (r *StreamReader[T]) Recv() (T, error)

Recv receives one typed stream item. It returns io.EOF when the remote side cleanly closes its send side.

func (*StreamReader[T]) Stream added in v0.5.0 

func (r *StreamReader[T]) Stream() *Stream

Stream returns the raw stream.

type StreamWriter added in v0.5.0 

type StreamWriter[T any] struct {
	// contains filtered or unexported fields
}

StreamWriter is a typed send-only stream wrapper.

func (*StreamWriter[T]) Close added in v0.5.0 

func (w *StreamWriter[T]) Close() error

Close closes the local sending side of the stream.

func (*StreamWriter[T]) Send added in v0.5.0 

func (w *StreamWriter[T]) Send(item T) error

Send sends one typed stream item.

func (*StreamWriter[T]) Stream added in v0.5.0 

func (w *StreamWriter[T]) Stream() *Stream

Stream returns the raw stream.

Source Files

View all Source files

Directories

Path Synopsis
examples
inventory/client command
Package main runs the GoRPC Inventory example client.
 Package main runs the GoRPC Inventory example client.
inventory/server command
Package main runs the GoRPC Inventory example server.
 Package main runs the GoRPC Inventory example server.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.