rpcstream

type ChannelStream ¶ added in v0.114.1

type ChannelStream struct {
	// Requests takes the caller's messages on a bidi stream; closing it half-closes the
	// stream, the way a client's CloseSend does. It is nil on a server stream, whose one
	// request the handler was served with.
	Requests chan<- proto.Message

	// Responses carries the handler's messages and closes once the handler returned.
	Responses <-chan proto.Message

	// Err is the handler's result: nil when it finished the stream, the error it refused or
	// failed with otherwise. It is written before Responses closes and is only meaningful
	// once it has.
	Err error
}

ChannelStream is a streaming handler served over channels in place of a transport, to a caller in the same process. The handler runs on its own goroutine with the context it was served with, which must carry what the transport's middleware would put there (the tenant, for the engine's handlers). It ends when that context is cancelled, when Requests is closed, or when it finishes on its own; Responses closes once it returned and Err then reports how it ended.

func ServeBidi[Req any, Resp any](ctx context.Context, first proto.Message, handler func(context.Context, func() (*Req, error), *Sender[Resp]) error) (*ChannelStream, error)

func ServeServer[Resp any](ctx context.Context, handler func(context.Context, *Sender[Resp]) error) *ChannelStream

type Sender ¶

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

Sender serializes sends on a server stream and rejects sends once the owning handler has returned.

Handlers hand their stream to other goroutines (the dispatcher sends assigned actions from message queue consumers, subscriptions send from fan-out goroutines). The HTTP/2 server panics on a write that arrives after the handler has returned, so every handler that shares its stream must wrap it in a Sender and defer Close: Close turns every later Send into ErrClosed and does not return while a Send is still writing. A Send to a peer that has stopped reading is interrupted after a short grace period, so a handler can always return.

func NewSender[T any](ctx context.Context, stream Stream[T]) *Sender[T]

func (s *Sender[T]) Close()

func (s *Sender[T]) Send(msg *T) error

type Stream ¶

type Stream[T any] interface {
	Send(*T) error
}

Stream is the sending half of a server stream, satisfied by *connect.ServerStream[T] and *connect.BidiStream[Req, T].