transport

type Header struct{ Name, Value string }

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

type Message struct {
	Direction string // "sent" or "received"
	At        time.Time
	Headers   []Header
	Body      []byte
	// Channel, on a message sent over a Session, is where to send it; empty
	// means the channel the connection was opened on.
	Channel   string
	Key       string // Kafka key
	Partition int32  // Kafka partition
	Offset    int64  // Kafka offset
	Sequence  uint64 // JetStream stream sequence
}

type Opener interface {
	Open(ctx context.Context, op Operation) (Session, error)
}

type Operation struct {
	Server  string
	Channel string
	Send    *Message
	Receive *Receive
	// Headers are connection headers (the WebSocket handshake, the SSE
	// request); message headers are Send.Headers.
	Headers []Header
	// Binding holds the protocol's options by name.
	Binding map[string]string
	TLS     *tls.Config
	Clock   func() time.Time // nil → time.Now
}

type Receive struct {
	Count   int
	Timeout time.Duration
	// Match, when set, decides whether a message counts; a message that does
	// not match is dropped. An error stops the receive.
	Match func(Message) (bool, error)
	// MaxBytes, when positive, stops the receive once the bodies it kept add
	// up to more than this.
	MaxBytes int64
	// OnMessage, when set, is called with each message kept, as it is kept
	// (so a caller can show progress while Count is still being waited for).
	// It runs on the goroutine that called Wait and must return quickly.
	OnMessage func(Message)
	// OnKept, when set, is called after OnMessage with each message kept. An
	// error stops the receive: Wait returns the messages kept so far, this one
	// included, and the error. It is how a hook can stop a run at the first
	// message that fails its check. It runs on the goroutine that called Wait.
	OnKept func(Message) error
}

type Registry map[string]Transport

type Result struct {
	Sent     *Message
	Received []Message
	Connect  time.Duration
	Total    time.Duration
}

type Session interface {
	// Send puts one message on the connection and returns it as sent.
	Send(ctx context.Context, m Message) (Message, error)
	// Receive collects messages as Receive says, from those already held and
	// those still to come. A timeout returns what was kept with ErrTimeout and
	// leaves the connection open.
	Receive(ctx context.Context, r Receive) ([]Message, error)
	// Close ends the connection and every goroutine it started. Closing twice
	// is harmless.
	Close() error
}

type Transport interface {
	Run(ctx context.Context, op Operation) (Result, error)
}