Documentation
¶
Index ¶
Constants ¶
This section is empty.
Variables ¶
Functions ¶
func StartReconnectingStream ¶
func StartReconnectingStream[S grpcClientStream, R any, E any]( ctx context.Context, cfg ReconnectingStreamConfig[S, R, E], ) (<-chan E, func(), error)
StartReconnectingStream opens a stream, continuously receives messages and emits mapped events to a channel until context cancelation or a terminal error.
Retryable receive/open errors are handled with backoff and reopen attempts. When ConnectionEvent is provided, DISCONNECTED/RECONNECTED lifecycle events are emitted around reconnect windows.
It returns the event channel, a close function, and an initialization error if the initial stream open fails.
Type parameters:
- S: concrete stream type used by Open/Recv.
- R: raw response type read from the stream.
- E: output event type written to the returned channel.
Example:
eventsCh, closeFn, err := StartReconnectingStream(
ctx,
ReconnectingStreamConfig[
arkv1.ArkService_GetTransactionsStreamClient,
*arkv1.GetTransactionsStreamResponse,
client.TransactionEvent,
]{
Open: func(ctx context.Context) (arkv1.ArkService_GetTransactionsStreamClient, error) {
return svc.GetTransactionsStream(ctx, &arkv1.GetTransactionsStreamRequest{})
},
Recv: func(s arkv1.ArkService_GetTransactionsStreamClient) (*arkv1.GetTransactionsStreamResponse, error) {
return s.Recv()
},
HandleResp: func(
ctx context.Context,
out chan<- client.TransactionEvent,
resp *arkv1.GetTransactionsStreamResponse,
) error {
// map response and send out one or more domain events
return nil
},
ErrorEvent: func(err error) client.TransactionEvent {
return client.TransactionEvent{Err: err}
},
},
)
if err != nil {
return err
}
defer closeFn()
for ev := range eventsCh {
_ = ev
}
Types ¶
type Broadcaster ¶
type Broadcaster[T any] struct { // contains filtered or unexported fields }
func NewBroadcaster ¶
func NewBroadcaster[T any]() *Broadcaster[T]
func (*Broadcaster[T]) Close ¶
func (l *Broadcaster[T]) Close()
func (*Broadcaster[T]) Publish ¶
func (l *Broadcaster[T]) Publish(v T) int
func (*Broadcaster[T]) Subscribe ¶
func (l *Broadcaster[T]) Subscribe(buf int) <-chan T
func (*Broadcaster[T]) Unsubscribe ¶
func (l *Broadcaster[T]) Unsubscribe(ch <-chan T)
type ReconnectingStreamConfig ¶
type ReconnectingStreamConfig[S grpcClientStream, R any, E any] struct { // Connect creates a new stream instance. Called once at startup Connect func(context.Context) (S, error) // Reconnect creates a new stream instance after retryable failures while reconnecting. Reconnect func(context.Context) (string, S, error) // Recv reads one response from the current stream instance. Recv func(S) (*R, error) // HandleResp maps one response into domain events and writes them to eventsCh. // Returning an error terminates the stream and emits ErrorEvent. HandleResp func(context.Context, chan<- E, R) error // ErrorEvent maps terminal errors into the stream event type. ErrorEvent func(error) E // ConnectionEvent maps DISCONNECTED/RECONNECTED lifecycle transitions into // the stream event type. Optional. ConnectionEvent func(ReconnectingStreamStateEvent) E // OnServerClosed runs when Recv returns io.EOF. OnServerClosed func() // OnDisconnect runs for every retryable receive error before sleep. OnDisconnect func(error) // OnReconnectSuccess runs after a stream reopen succeeds and the first // message has been received from the new stream. OnReconnectSuccess func(R) }
ReconnectingStreamConfig defines how to open, read and map a server stream into domain events while handling reconnects in a shared generic loop.
Required fields: Open, Recv, HandleResp, ErrorEvent. Optional callbacks are used for connection lifecycle events, hooks and logs.
Type parameters:
- S is the concrete gRPC client stream type (must implement CloseSend), for example arkv1.ArkService_GetEventStreamClient.
- R is the raw message type returned by stream.Recv(), for example *arkv1.GetEventStreamResponse.
- E is the final event type emitted on the output channel, for example client.BatchEventChannel.
type ReconnectingStreamState ¶
type ReconnectingStreamState string
const ( ReconnectingStreamStateDisconnected ReconnectingStreamState = "DISCONNECTED" ReconnectingStreamStateReconnected ReconnectingStreamState = "RECONNECTED" ReconnectingStreamStateReady ReconnectingStreamState = "READY" )
Source Files
Click to show internal directories.
Click to hide internal directories.