iprotoconn

Documentation

Overview

Package iprotoconn is a client/server library for the iproto wire protocol: request/response and fire-and-forget messaging over TCP or Unix stream sockets, framed by a fixed 12-byte header with application-defined payloads (github.com/adventures-team/go-iproto is the preferred serialization).

The client routes calls across a sharded cluster of masters and replicas configured from an onlineconf Module or Subtree, with retries, hedging, a retry budget, per-shard circuit breakers, and request cancellation propagated to servers. The server dispatches to typed handlers registered on a Mux with compile-time checking. Messaging is symmetric: servers push notifications or issue calls to connected clients through retainable Peer handles, and clients receive them by attaching their own Mux via Client.Handle.

Protocol

Every packet is a 12-byte little-endian header — message type, payload length, sequence number — followed by an opaque payload. Message types below 0xFF00 belong to the application; above them sit the control messages (ping, shutdown, cancel) and a reserved range. Sequence number 0 marks notifications and control messages; every request attempt draws a fresh nonzero number from a process-wide counter, and a response echoes the request's type and sequence number back on the same connection. There is no version negotiation, no checksums and no compression; recovery from a desynchronized stream is closing the connection.

Security model

The protocol assumes a trusted internal network: there is no authentication, authorization, or encryption. Never expose an iproto endpoint to untrusted traffic — front it with a gateway that authenticates and admission-controls. Within that trust model the library hardens itself against misbehaving peers and accidental overload: connection and per-connection packet-rate limits, staged read deadlines against trickle attacks, a payload size cap checked before allocation, throttled logging, per-connection isolation of request tracking and cancellation, and panic containment — a panic in a handler or a plugged-in dependency costs at most a request or a connection, never the process.

Delivery guarantees

Request/response is at-least-once when retries are enabled (the default): a retried or hedged request may execute more than once, and cancellation is best-effort — make handlers idempotent, or disable retries by setting WithMaxTries to 1 for at-most-once. Notifications are best-effort: no acknowledgement, and loss on queue overflow or connection failure is silent. There is no exactly-once and no cross-request ordering.

Calls, retries and timeouts

A call runs under a total hard deadline (/request-timeout, 500 ms by default, combined with the caller's context) that is never retried past. Attempts failing with hard or network outcomes are retried on other servers with exponential backoff, up to /max-tries and within a sliding-window retry budget that damps retry storms; the optional per-attempt timeout (/attempt-timeout) turns hung attempts into retryable failures, and the RetryEarly flag adds hedged duplicate requests against tail latency. A shard whose recent calls mostly failed is fast-failed with ErrBreakerOpen until its circuit breaker's probes succeed. See the Outcome constants for how responses are classified.

Replicas

Masters accept reads and writes, replicas reads only. Calls go to masters unless replica preference is selected per call (WithUseReplica, WithFrom) or per request type (ReplicaUser). Replica preference falls back to masters when replicas fail — during selection and across retries — unless /master-fallback is false.

Strict unmarshal

Decoding a response ignores a non-empty tail left by UnmarshalIProto by default. Strict mode — per call with WithStrictUnmarshal, global with /strict-unmarshal — fails such calls with ErrStrictUnmarshal instead, catching drifting message definitions early.

Examples of the client, server, and server-push flows accompany NewClient, Handle, and Client.Handle.

Index

• Constants
• Variables
• func Handle[Req any, Resp any, PReq interface{ ... }, PResp interface{ ... }](mux *Mux, fn func(ctx context.Context, req *Req) *Resp) error
• func HandleNotify[Req any, PReq interface{ ... }](mux *Mux, fn func(ctx context.Context, req *Req)) error
• func MustHandle[Req any, Resp any, PReq interface{ ... }, PResp interface{ ... }](mux *Mux, fn func(ctx context.Context, req *Req) *Resp)
• func MustHandleNotify[Req any, PReq interface{ ... }](mux *Mux, fn func(ctx context.Context, req *Req))
• type BatchCall
• type CallOption
  • func WithAttemptTimeout(d time.Duration) CallOption
  • func WithClassifier[Resp any, PResp interface{ ... }](fn func(resp *Resp) Outcome) CallOption
  • func WithClassifierRaw(fn func(cmd uint32, payload []byte) Outcome) CallOption
  • func WithEarlyTimeout(d time.Duration) CallOption
  • func WithFrom(f RolePref) CallOption
  • func WithMaxTries(n int) CallOption
  • func WithRetryFlags(f RetryFlags) CallOption
  • func WithShard(n int) CallOption
  • func WithSoftRetry[Resp any, PResp interface{ ... }](fn func(resp *Resp) bool) CallOption
  • func WithSoftRetryRaw(fn func(cmd uint32, payload []byte) bool) CallOption
  • func WithStrictUnmarshal(strict bool) CallOption
  • func WithTimeout(d time.Duration) CallOption
  • func WithUseReplica() CallOption
• type Client
  • func NewClient(ctx context.Context, cfg ConfigSource, opts ...ClientOption) (*Client, error)
  • func (c *Client) Call(ctx context.Context, req Request, resp Response, opts ...CallOption) error
  • func (c *Client) CallBatch(ctx context.Context, batch []BatchCall, opts ...CallOption) error
  • func (c *Client) Close(ctx context.Context) error
  • func (c *Client) Handle(h Handler) error
  • func (c *Client) Notify(ctx context.Context, req Request, opts ...CallOption) error
• type ClientOption
  • func WithDialer(d Dialer) ClientOption
  • func WithLogger(l *slog.Logger) ClientOption
  • func WithMetrics(r metrics.Registry) ClientOption
  • func WithPool(p wire.Pool) ClientOption
• type Cmder
• type ConfigSource
• type Dialer
• type Error
• type Handler
• type Incoming
• type Mux
  • func NewMux() *Mux
• type Outcome
• type Peer
  • func PeerFrom(ctx context.Context) (Peer, bool)
• type ReplicaUser
• type Request
• type Response
• type RetryFlags
• type RolePref
• type Server
  • func NewServer(ctx context.Context, cfg ConfigSource, h Handler, opts ...ClientOption) (*Server, error)
• type Sharder

Examples

• Client.Handle
• Handle
• NewClient

Constants

const (
	Master            = cluster.Master
	Replica           = cluster.Replica
	MasterThenReplica = cluster.MasterThenReplica
	ReplicaThenMaster = cluster.ReplicaThenMaster
)

Role preferences select which servers of a shard a call may be routed to. Master is the default: with no WithFrom or WithUseReplica option and no ReplicaUser implementation on the request type, calls go to masters only. Override it per call with WithFrom or WithUseReplica, or per request type by implementing UseReplica.

Replica preference falls back to masters — both during selection and across retries — unless /master-fallback is set to false; ReplicaThenMaster requests that fallback explicitly, regardless of the setting, and MasterThenReplica is its mirror image.

const (
	OutcomeOK      = breaker.OutcomeOK
	OutcomeSoft    = breaker.OutcomeSoft
	OutcomeHard    = breaker.OutcomeHard
	OutcomeNetwork = breaker.OutcomeNetwork
)

Outcomes classify the result of every attempt and drive retries, server freezing and the circuit breaker.

OutcomeOK completes the call. OutcomeSoft is an application-level failure carried by an otherwise healthy response (say, "record not found"): the call completes immediately with the decoded response and a soft Error — the server is not frozen, the breaker records a success, nothing is retried. OutcomeHard means the responding server itself is unhealthy (say, a storage failure reported in the payload): the server is frozen and deprioritized, the breaker records an error, and the attempt is retried on another server — exactly like OutcomeNetwork, which the transport assigns on dial, I/O and timeout failures.

A typed WithClassifier callback (or its raw-payload form, WithClassifierRaw) assigns outcomes to responses; without one, every response is OutcomeOK, and a response the typed classifier cannot decode is OutcomeHard — an undecodable response marks the server unusable. Retries and hedges appear in the iproto_retries_total metric (retry_type = regular, early or soft) and breaker reactions in iproto_circuit_breaker_state and iproto_circuit_breaker_transitions_total.

const (
	RetryEarly = dispatch.RetryEarly
	RetrySafe  = dispatch.RetrySafe
	RetrySame  = dispatch.RetrySame
)

Retry flags tune the retry strategy, per call with WithRetryFlags or globally via /retry-flags; the configured default is RetrySafe alone.

RetryEarly enables hedging: when no response has arrived within the early timeout (/early-timeout, 50 ms by default, overridable with WithEarlyTimeout), up to /hedge-fanout (default 2) duplicate copies of the request are additionally sent to other servers of the shard, each as a fresh attempt. The original stays in flight; the first OK or soft response wins, the losing attempts are cancelled, and their late responses are discarded silently (counted in iproto_speculative_waste_total). Hedges consume the retry budget and appear as retry_type="early" in iproto_retries_total, and hedging suspends itself while the shard's circuit breaker is not closed or the retry budget runs hot. Because hedged copies execute concurrently on several servers, enable hedging for idempotent operations only.

RetrySafe — the default — stops further retries once a message is unsafe: one of its attempts has already produced a response while other copies are still in flight (a state hedging creates), so a retry could duplicate work that may yet succeed.

RetrySame lets a retry select the same server again instead of excluding every server already tried.

Variables

var (
	ErrTimeout            = dispatch.ErrTimeout
	ErrConnClosed         = conn.ErrConnClosed
	ErrQueueFull          = conn.ErrQueueFull
	ErrMaxInFlight        = conn.ErrMaxInFlight
	ErrBreakerOpen        = dispatch.ErrBreakerOpen
	ErrBudgetExhausted    = dispatch.ErrBudgetExhausted
	ErrNoServers          = cluster.ErrNoServers
	ErrShuttingDown       = conn.ErrShuttingDown
	ErrTypeHandledLocally = server.ErrTypeHandledLocally
	ErrNoShard            = dispatch.ErrNoShard
	ErrStrictUnmarshal    = server.ErrStrictUnmarshal
	ErrDuplicateType      = server.ErrDuplicateType
	ErrMuxFrozen          = server.ErrMuxFrozen
)

Sentinel errors; match with errors.Is.

Functions

func Handle

func Handle[
	Req any, Resp any,
	PReq interface {
		*Req
		iproto.Unmarshaler
		Cmder
	},
	PResp interface {
		*Resp
		iproto.Marshaler
	},
](mux *Mux, fn func(ctx context.Context, req *Req) *Resp) error

Handle registers a typed request-response handler. All four type parameters are inferred at the call site:

iprotoconn.Handle(mux, func(ctx context.Context, req *MyRequest) *MyResponse { ... })

Req must carry Cmd() uint32 (value or pointer receiver — the constraint sits on *Req); *Req must implement iproto.Unmarshaler and *Resp iproto.Marshaler, all checked at compile time. Cmd() is invoked once, on a zero value, at registration. A nil return sends no response; a response for a notification (sequence number 0) is discarded. Registration failures (ErrDuplicateType, ErrMuxFrozen) are returned; use MustHandle for the panicking startup idiom.

Example

ExampleHandle registers a typed handler — the request/response contracts are checked at compile time — and serves it.

package main

import (
	"context"
	"strconv"
	"strings"
	"time"

	iprotoconn "github.com/adventures-team/go-iprotoconn"
)

const exampleAddr = "127.0.0.1:3300"

// exampleSource is a minimal in-memory ConfigSource. Real applications
// pass an *onlineconf.Module or *onlineconf.Subtree, which satisfy the
// same interface.
type exampleSource map[string]string

func (s exampleSource) GetString(p, d string) string {
	if v, ok := s[p]; ok {
		return v
	}

	return d
}

func (s exampleSource) GetStringIfExists(p string) (string, bool) {
	v, ok := s[p]
	return v, ok
}

func (s exampleSource) GetInt(p string, d int) int {
	if v, ok := s[p]; ok {
		if n, err := strconv.Atoi(v); err == nil {
			return n
		}
	}

	return d
}

func (s exampleSource) GetBool(p string, d bool) bool {
	if v, ok := s[p]; ok {
		return v == "true"
	}

	return d
}

func (s exampleSource) GetDuration(p string, d time.Duration) time.Duration {
	if v, ok := s[p]; ok {
		if dur, err := time.ParseDuration(v); err == nil {
			return dur
		}
	}

	return d
}

func (s exampleSource) GetFloat(p string, d float64) float64 {
	if v, ok := s[p]; ok {
		if f, err := strconv.ParseFloat(v, 64); err == nil {
			return f
		}
	}

	return d
}

func (s exampleSource) GetStrings(p string, d []string) []string {
	if v, ok := s[p]; ok {
		return strings.Split(v, ",")
	}

	return d
}

// GetUserReq is a request message: Cmd identifies its iproto message
// type, and the two codec methods are what iprotogen generates for
// types marked //adv:iproto:.
type GetUserReq struct{ ID uint32 }

func (GetUserReq) Cmd() uint32 { return 0x11 }

func (r GetUserReq) MarshalIProto(b []byte) ([]byte, error) {
	return append(b, byte(r.ID), byte(r.ID>>8), byte(r.ID>>16), byte(r.ID>>24)), nil
}

func (r *GetUserReq) UnmarshalIProto(b []byte) ([]byte, error) {
	r.ID = uint32(b[0]) | uint32(b[1])<<8 | uint32(b[2])<<16 | uint32(b[3])<<24
	return b[4:], nil
}

// GetUserResp is the matching response message.
type GetUserResp struct{ Name string }

func (GetUserResp) Cmd() uint32 { return 0x11 }

func (r GetUserResp) MarshalIProto(b []byte) ([]byte, error) {
	return append(b, r.Name...), nil
}

func (r *GetUserResp) UnmarshalIProto(b []byte) ([]byte, error) {
	r.Name = string(b)
	return nil, nil
}

func main() {
	ctx := context.Background()

	mux := iprotoconn.NewMux()
	iprotoconn.MustHandle(mux, func(_ context.Context, req *GetUserReq) *GetUserResp {
		return &GetUserResp{Name: "user-" + strconv.Itoa(int(req.ID))} // nil = no response
	})

	srv, err := iprotoconn.NewServer(ctx, exampleSource{"/listen": exampleAddr}, mux)
	if err != nil {
		return
	}

	_ = srv.ListenAndServe(ctx)
}

func HandleNotify

func HandleNotify[
	Req any,
	PReq interface {
		*Req
		iproto.Unmarshaler
		Cmder
	},
](mux *Mux, fn func(ctx context.Context, req *Req)) error

HandleNotify registers a typed handler for a notification-only message type: no response type and no response sent — even if the inbound packet carries a nonzero sequence number.

func MustHandle

func MustHandle[
	Req any, Resp any,
	PReq interface {
		*Req
		iproto.Unmarshaler
		Cmder
	},
	PResp interface {
		*Resp
		iproto.Marshaler
	},
](mux *Mux, fn func(ctx context.Context, req *Req) *Resp)

MustHandle is Handle panicking on registration errors.

func MustHandleNotify

func MustHandleNotify[
	Req any,
	PReq interface {
		*Req
		iproto.Unmarshaler
		Cmder
	},
](mux *Mux, fn func(ctx context.Context, req *Req))

MustHandleNotify is HandleNotify panicking on registration errors.

Types

type BatchCall

type BatchCall struct {
	// Req is the request to send. It may implement [Sharder] and
	// [ReplicaUser] to carry its own routing.
	Req Request
	// Resp receives the decoded response; nil discards the payload.
	Resp Response
	// Opts apply to this message only, overriding the batch-level
	// options passed to [Client.CallBatch], which in turn override
	// configuration defaults.
	Opts []CallOption
	// Err is this message's outcome, set by [Client.CallBatch]: nil,
	// a classified [Error], or a sentinel such as [ErrTimeout].
	Err error
}

BatchCall is one message of a batch.

Every message routes independently: through request-carried routing (Sharder, ReplicaUser) or per-message Opts, the requests of one batch may target different shards and different servers. Messages run in parallel within a chunk of at most /max-bulk-size (100 by default); chunks run sequentially, so no more than one chunk of the batch is in flight at a time.

type CallOption

type CallOption func(*callOptions)

CallOption tunes one message. Per-call options override request-carried routing (Sharder/ReplicaUser), which overrides configuration defaults.

func WithAttemptTimeout

func WithAttemptTimeout(d time.Duration) CallOption

WithAttemptTimeout overrides the per-attempt timeout; 0 disables it explicitly (attempts bounded only by the total timeout).

func WithClassifier

func WithClassifier[Resp any, PResp interface {
	*Resp
	iproto.Unmarshaler
}](fn func(resp *Resp) Outcome) CallOption

WithClassifier sets a typed response classifier: after every attempt whose network round-trip succeeded, the response payload is decoded into a fresh *Resp — exactly like a handler's request — and fn classifies the decoded value. Without any classifier every response is OutcomeOK. OutcomeSoft completes the call with the response and a soft Error; OutcomeHard freezes the responding server, feeds the circuit breaker and retries on another server, like a network error (see the Outcome constants).

A payload that fails to decode classifies as OutcomeHard: a server answering with an undecodable response — schema drift included — is literally unusable from this client's perspective, so it is frozen and the attempt is retried elsewhere. The classifying decode runs once per attempt (in addition to Client.Call's final decode of the winning payload) and ignores non-empty tails regardless of strict mode. A panicking fn is contained and treated as OutcomeOK.

Because tails are ignored, Resp need not be the response type passed to Client.Call: a prefix type decoding only the leading fields classifies at minimal cost. Protocols that encode a status in the first bytes of every response can classify through a status-only view,

type StatusPrefix struct { // decodes 4 bytes of any response
	Err uint32
}

type AppResp struct { // the full type, decoded once by Call
	Err      uint32
	UserID   uint32
	UserName string
	// many more fields
}

making the per-attempt decode a few bytes regardless of response size. This requires the discriminating fields to be encoded first.

Use WithClassifierRaw when the response type is not nameable at the call site or to skip the per-attempt decode.

func WithClassifierRaw

func WithClassifierRaw(fn func(cmd uint32, payload []byte) Outcome) CallOption

WithClassifierRaw sets the response classifier in its raw-payload form: fn inspects the undecoded payload of every network-successful attempt and returns an Outcome, with the semantics of WithClassifier.

func WithEarlyTimeout

func WithEarlyTimeout(d time.Duration) CallOption

WithEarlyTimeout overrides the hedging trigger for this call: with RetryEarly set, a call still unanswered after this delay sends hedged duplicate requests to other servers (see the retry flag constants for the full hedging semantics).

func WithFrom

func WithFrom(f RolePref) CallOption

WithFrom sets the server role preference; without it (and without a ReplicaUser implementation on the request) the default is Master.

func WithMaxTries

func WithMaxTries(n int) CallOption

WithMaxTries overrides the maximum send attempts.

func WithRetryFlags

func WithRetryFlags(f RetryFlags) CallOption

WithRetryFlags overrides the retry strategy for this call, replacing the configured /retry-flags entirely — combine flags with the bitwise OR, e.g. [RetrySafe]|[RetryEarly] to keep safe-retry semantics while enabling hedging (see the retry flag constants).

func WithShard

func WithShard(n int) CallOption

WithShard routes the message to a shard (1-based; 0 is valid only on single-shard clusters). When present, a Sharder implementation on the request is not consulted.

func WithSoftRetry

func WithSoftRetry[Resp any, PResp interface {
	*Resp
	iproto.Unmarshaler
}](fn func(resp *Resp) bool) CallOption

WithSoftRetry sets a typed soft-retry callback: invoked on each OutcomeOK response, decoded into a fresh *Resp like a handler's request; returning true re-dispatches the call after a linearly growing delay (counted against the try limit, shown as retry_type="soft" in iproto_retries_total). Useful for "accepted, try again later" replies. A payload that fails to decode is not soft-retried: the call completes and the decode error surfaces from Client.Call.

func WithSoftRetryRaw

func WithSoftRetryRaw(fn func(cmd uint32, payload []byte) bool) CallOption

WithSoftRetryRaw sets the soft-retry callback in its raw-payload form, with the semantics of WithSoftRetry.

func WithStrictUnmarshal

func WithStrictUnmarshal(strict bool) CallOption

WithStrictUnmarshal overrides /strict-unmarshal for this call: in strict mode a non-empty tail left by the response's UnmarshalIProto fails the call with ErrStrictUnmarshal; by default the tail is ignored.

func WithTimeout

func WithTimeout(d time.Duration) CallOption

WithTimeout overrides the total request timeout (hard deadline).

func WithUseReplica

func WithUseReplica() CallOption

WithUseReplica selects replica preference for this call: replicas are tried first, falling back to masters during selection and retries unless /master-fallback is false. Equivalent to WithFrom with Replica.

type Client

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

Client is the sharded cluster client. Shards, masters and replicas come entirely from the ConfigSource; parameters are read on the fly and topology changes are applied live when the source is Subscribable.

func NewClient

func NewClient(ctx context.Context, cfg ConfigSource, opts ...ClientOption) (*Client, error)

NewClient builds a client over cfg (an onlineconf Module or Subtree, or any ConfigSource). ctx bounds the background machinery (topology watcher); cancelling it is equivalent to Close with immediate drop.

Example

ExampleNewClient issues a typed request-response call against a cluster configured by the source (shards, masters and replicas all come from configuration).

package main

import (
	"context"
	"log"
	"strconv"
	"strings"
	"time"

	iprotoconn "github.com/adventures-team/go-iprotoconn"
)

const exampleAddr = "127.0.0.1:3300"

// exampleSource is a minimal in-memory ConfigSource. Real applications
// pass an *onlineconf.Module or *onlineconf.Subtree, which satisfy the
// same interface.
type exampleSource map[string]string

func (s exampleSource) GetString(p, d string) string {
	if v, ok := s[p]; ok {
		return v
	}

	return d
}

func (s exampleSource) GetStringIfExists(p string) (string, bool) {
	v, ok := s[p]
	return v, ok
}

func (s exampleSource) GetInt(p string, d int) int {
	if v, ok := s[p]; ok {
		if n, err := strconv.Atoi(v); err == nil {
			return n
		}
	}

	return d
}

func (s exampleSource) GetBool(p string, d bool) bool {
	if v, ok := s[p]; ok {
		return v == "true"
	}

	return d
}

func (s exampleSource) GetDuration(p string, d time.Duration) time.Duration {
	if v, ok := s[p]; ok {
		if dur, err := time.ParseDuration(v); err == nil {
			return dur
		}
	}

	return d
}

func (s exampleSource) GetFloat(p string, d float64) float64 {
	if v, ok := s[p]; ok {
		if f, err := strconv.ParseFloat(v, 64); err == nil {
			return f
		}
	}

	return d
}

func (s exampleSource) GetStrings(p string, d []string) []string {
	if v, ok := s[p]; ok {
		return strings.Split(v, ",")
	}

	return d
}

// GetUserReq is a request message: Cmd identifies its iproto message
// type, and the two codec methods are what iprotogen generates for
// types marked //adv:iproto:.
type GetUserReq struct{ ID uint32 }

func (GetUserReq) Cmd() uint32 { return 0x11 }

func (r GetUserReq) MarshalIProto(b []byte) ([]byte, error) {
	return append(b, byte(r.ID), byte(r.ID>>8), byte(r.ID>>16), byte(r.ID>>24)), nil
}

func (r *GetUserReq) UnmarshalIProto(b []byte) ([]byte, error) {
	r.ID = uint32(b[0]) | uint32(b[1])<<8 | uint32(b[2])<<16 | uint32(b[3])<<24
	return b[4:], nil
}

// GetUserResp is the matching response message.
type GetUserResp struct{ Name string }

func (GetUserResp) Cmd() uint32 { return 0x11 }

func (r GetUserResp) MarshalIProto(b []byte) ([]byte, error) {
	return append(b, r.Name...), nil
}

func (r *GetUserResp) UnmarshalIProto(b []byte) ([]byte, error) {
	r.Name = string(b)
	return nil, nil
}

func main() {
	ctx := context.Background()

	client, err := iprotoconn.NewClient(ctx, exampleSource{examplePathKey: exampleAddr})
	if err != nil {
		return
	}

	defer func() { _ = client.Close(context.Background()) }()

	var resp GetUserResp

	err = client.Call(ctx, GetUserReq{ID: 123}, &resp, iprotoconn.WithUseReplica())
	if err != nil {
		return
	}

	log.Printf("user 123 is %q", resp.Name)
}

func (*Client) Call

func (c *Client) Call(ctx context.Context, req Request, resp Response, opts ...CallOption) error

Call issues one request-response message and decodes the winning payload into resp; a nil resp discards the payload. Routing follows per-call options, then request-carried routing (Sharder, ReplicaUser), then configuration defaults. A response classified OutcomeSoft still decodes into resp and returns an Error with Class OutcomeSoft; hard and network failures are retried per the retry configuration before an Error is returned. A non-empty decode tail is ignored unless strict mode is on (WithStrictUnmarshal or /strict-unmarshal).

func (*Client) CallBatch

func (c *Client) CallBatch(ctx context.Context, batch []BatchCall, opts ...CallOption) error

CallBatch dispatches the batch in sequential chunks of at most /max-bulk-size messages; within a chunk every message runs concurrently, each routed independently (see BatchCall). Per-message outcomes land in BatchCall.Err; CallBatch itself returns a context error only when the batch deadline or cancellation interrupts processing.

func (*Client) Close

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

Close gracefully drains: topology watcher stopped, every server released with a shutdown handshake on its connection.

func (*Client) Handle

func (c *Client) Handle(h Handler) error

Handle attaches the handler for peer-initiated requests and notifications — server pushes — on all of this client's connections. Attach once, during initialization; a second attach returns an error. Attaching freezes the handler's TypeSet, and the message types it serves become unsendable through this client (ErrTypeHandledLocally).

Example

ExampleClient_Handle receives server-pushed notifications: the client attaches its own mux, and the server delivers UserEvent messages to it through the Peer handle it obtained from a handler context (PeerFrom).

package main

import (
	"context"
	"log"
	"strconv"
	"strings"
	"time"

	iprotoconn "github.com/adventures-team/go-iprotoconn"
)

const exampleAddr = "127.0.0.1:3300"

// exampleSource is a minimal in-memory ConfigSource. Real applications
// pass an *onlineconf.Module or *onlineconf.Subtree, which satisfy the
// same interface.
type exampleSource map[string]string

func (s exampleSource) GetString(p, d string) string {
	if v, ok := s[p]; ok {
		return v
	}

	return d
}

func (s exampleSource) GetStringIfExists(p string) (string, bool) {
	v, ok := s[p]
	return v, ok
}

func (s exampleSource) GetInt(p string, d int) int {
	if v, ok := s[p]; ok {
		if n, err := strconv.Atoi(v); err == nil {
			return n
		}
	}

	return d
}

func (s exampleSource) GetBool(p string, d bool) bool {
	if v, ok := s[p]; ok {
		return v == "true"
	}

	return d
}

func (s exampleSource) GetDuration(p string, d time.Duration) time.Duration {
	if v, ok := s[p]; ok {
		if dur, err := time.ParseDuration(v); err == nil {
			return dur
		}
	}

	return d
}

func (s exampleSource) GetFloat(p string, d float64) float64 {
	if v, ok := s[p]; ok {
		if f, err := strconv.ParseFloat(v, 64); err == nil {
			return f
		}
	}

	return d
}

func (s exampleSource) GetStrings(p string, d []string) []string {
	if v, ok := s[p]; ok {
		return strings.Split(v, ",")
	}

	return d
}

// UserEvent is a notification pushed by the server to subscribed
// clients.
type UserEvent struct{ Kind uint32 }

func (UserEvent) Cmd() uint32 { return 0x21 }

func (e UserEvent) MarshalIProto(b []byte) ([]byte, error) {
	return append(b, byte(e.Kind), byte(e.Kind>>8), byte(e.Kind>>16), byte(e.Kind>>24)), nil
}

func (e *UserEvent) UnmarshalIProto(b []byte) ([]byte, error) {
	e.Kind = uint32(b[0]) | uint32(b[1])<<8 | uint32(b[2])<<16 | uint32(b[3])<<24
	return b[4:], nil
}

func main() {
	ctx := context.Background()

	client, err := iprotoconn.NewClient(ctx, exampleSource{examplePathKey: exampleAddr})
	if err != nil {
		return
	}

	defer func() { _ = client.Close(context.Background()) }()

	events := iprotoconn.NewMux()
	iprotoconn.MustHandleNotify(events, func(_ context.Context, ev *UserEvent) {
		log.Printf("event %d", ev.Kind)
	})

	if err := client.Handle(events); err != nil {
		return
	}
}

func (*Client) Notify

func (c *Client) Notify(ctx context.Context, req Request, opts ...CallOption) error

Notify sends a fire-and-forget message.

type ClientOption

type ClientOption func(*clientOptions)

ClientOption tunes client construction.

func WithDialer

func WithDialer(d Dialer) ClientOption

WithDialer supplies the transport dialer (default: &net.Dialer{}).

func WithLogger

func WithLogger(l *slog.Logger) ClientOption

WithLogger supplies the client's logger (used as-is, without the default throttling wrapper).

func WithMetrics

func WithMetrics(r metrics.Registry) ClientOption

WithMetrics supplies the metrics registry (default: metrics.Nop).

func WithPool

func WithPool(p wire.Pool) ClientOption

WithPool supplies the payload buffer pool (default: wire.NewPool()).

type Cmder

type Cmder = server.Cmder

type ConfigSource

type ConfigSource = config.Source

ConfigSource is satisfied by *onlineconf.Module and *onlineconf.Subtree (structurally — no onlineconf import here).

type Dialer

type Dialer = cluster.Dialer

type Error

type Error = dispatch.Error

Error is the classified call error.

type Handler

type Handler = server.Handler

type Incoming

type Incoming = server.Incoming

type Mux

type Mux = server.Mux

func NewMux

func NewMux() *Mux

NewMux returns an empty typed-registration multiplexer.

type Outcome

type Outcome = breaker.Outcome

type Peer

type Peer = server.Peer

func PeerFrom

func PeerFrom(ctx context.Context) (Peer, bool)

PeerFrom extracts the Peer injected into every handler invocation's context.

type ReplicaUser

type ReplicaUser interface {
	UseReplica() bool
}

ReplicaUser supplies request-carried role preference: UseReplica() returning true selects replica preference (with master fallback, unless disabled by /master-fallback) when WithFrom and WithUseReplica are absent.

type Request

type Request = server.Request // = iproto.Marshaler + Cmd() uint32

type Response

type Response = server.Response // = iproto.Unmarshaler

type RetryFlags

type RetryFlags = dispatch.RetryFlags

RetryFlags is the retry strategy bitmask.

type RolePref

type RolePref = cluster.RolePref

type Server

type Server = server.Server

Server re-exports the acceptor.

func NewServer

func NewServer(ctx context.Context, cfg ConfigSource, h Handler, opts ...ClientOption) (*Server, error)

NewServer builds a server over cfg (an onlineconf Module or Subtree, or any ConfigSource); the listen address comes from /listen. The handler's TypeSet is frozen on attach. SIGTERM handling stays in the application:

srv, err := iprotoconn.NewServer(ctx, oc.Subtree("/my-service/server"), mux)
err = srv.ListenAndServe(ctx)

type Sharder

type Sharder interface {
	Shard(totalShards int) int
}

Sharder supplies request-carried shard routing: when the request type implements it and WithShard is absent, Client.Call, Client.Notify and Client.CallBatch invoke Shard(totalShards) — 1-based, validated exactly like WithShard's argument. totalShards is passed in because topology is live configuration the application cannot know statically; ID-based sharding is typically "int(r.ID)%totalShards + 1".