Documentation
¶
Index ¶
- Variables
- func BoundedReceive[Req any](ctx context.Context, stream grpc.ServerStream, timeout time.Duration) (*Req, error)
- func Heartbeats[Req any](stream grpc.ServerStream, validFn func(*Req) bool) <-chan struct{}
- func ProtoEqualError(a, b proto.Message) error
- func SetEqualError[T proto.Message](a, b []T) error
- func SliceClone[T proto.Message](src []T) []T
- func SliceEqualError[T proto.Message](a, b []T) error
- type Ptr
- type Sender
Constants ¶
This section is empty.
Variables ¶
var ErrSenderAbandoned = errors.New("stream sender abandoned after a timed-out send")
ErrSenderAbandoned is returned by Sender.Send once an earlier send timed out or was cancelled: that send may still be in flight on the goroutine, so the stream can no longer be written to in order. The handler should return.
Functions ¶
func BoundedReceive ¶
func BoundedReceive[Req any]( ctx context.Context, stream grpc.ServerStream, timeout time.Duration, ) (*Req, error)
BoundedReceive waits up to timeout for one message. It spawns a goroutine for the receive, so it suits a one-off (a stream's opening request) rather than a receive loop — see Heartbeats for the latter.
func Heartbeats ¶ added in v1.30.0
func Heartbeats[Req any](stream grpc.ServerStream, validFn func(*Req) bool) <-chan struct{}
Heartbeats receives on the stream for its whole life on one goroutine, signalling on the returned channel for each message validFn accepts and closing it on a receive error or a rejected message. Signals coalesce (a buffer of one, never blocking the receive): a heartbeat is a liveness fact, not a count, so a handler busy elsewhere misses nothing it needs. Enforcing a heartbeat deadline is the handler's, with a timer it resets per signal — a blocked RecvMsg cannot be interrupted, so the timeout has to be observed where the handler can act on it.
func ProtoEqualError ¶
func SetEqualError ¶
func SliceClone ¶
func SliceEqualError ¶
Types ¶
type Sender ¶ added in v1.30.0
type Sender[Resp any] struct { // contains filtered or unexported fields }
Sender serializes a server stream's sends on one goroutine that lives as long as the stream, so bounding a send costs nothing per message: a stuck SendMsg cannot be interrupted from the handler (only the client going away or the handler returning ends it), so a goroutine is what lets the handler walk away from it — but one per stream, not one per send, and the timeout comes from a single timer reused across sends.
Send is not safe for concurrent use: it is the handler loop's, which sends one message at a time.
func (*Sender[Resp]) Close ¶ added in v1.30.0
func (s *Sender[Resp]) Close()
Close lets the send goroutine exit. Call it when the handler returns; a send still in flight finishes (or is cut off by the stream ending) first.
Source Files