Documentation
¶
Overview ¶
Package stream provides
Index ¶
Constants ¶
const ( BufferSize = 128 ReconnectTimeout = 5 * time.Minute )
Variables ¶
Functions ¶
This section is empty.
Types ¶
type ConnectionObserver ¶
type ConnectionObserver interface {
ConnState() connectivity.State
WaitForReconnect(ctx context.Context) bool
}
type PublishClient ¶
type PublishClient interface {
ConnectionObserver
PublishStream(context.Context, ...grpc.CallOption) (api.Ensign_PublishClient, error)
}
type Publisher ¶
type Publisher struct {
// contains filtered or unexported fields
}
Publisher wraps an stream.PublishClient to maintain an open publish stream to an Ensign node. When the publisher is started it kicks off two go routines: one go routine ensures that the publish stream is re-opened when the connection becomes available and manages the recv go routine. The second go routine (the recv routine) listens for messages incoming from the server and handles them. If the recv routine cannot receive a message, it marks the stream as down and stops running, allowing the start go routine to re-establish the connection.
Publishing messages happens synchronously in the user thread, and an error is returned if the message cannot be published.
func NewPublisher ¶
func NewPublisher(client PublishClient, opts ...grpc.CallOption) (*Publisher, error)
Create a new low-level publisher stream manager that maintains the open publish stream and allows users to publish events and receive acks/nacks from the Ensign node. This function opens a publish stream and returns an error if the user is not authenticated or the stream cannot be opened. If the stream is opened successfully, the start go routine is kicked off, which ensures the stream stays open even if the remote node temporarily goes down. The start go routine also kicks of the receive routine to get acks/nacks back from the server as well as other streaming messages.
func (*Publisher) Close ¶
Close the publisher gracefully, once closed, the publisher cannot be restarted.
func (*Publisher) Err ¶
Err returns any fatal errors that are set on the publisher. If a non-nil error is returned then the publisher is not running and all events published will fail.
func (*Publisher) Publish ¶
func (p *Publisher) Publish(topic string, event *api.Event) (_ *api.EventWrapper, _ <-chan *api.PublisherReply, err error)
Publish an event to the publish stream. This method blocks until a stream is available to send on and synchronously sends the event.
Publish wraps the api.Event in an event wrapper by looking up the topic in the local topic map. Users can supply either a string ULID for the topicID or the name of the topic, which must be in the topic map returned by the server at the start of the publish stream. This method also assigns the topic a localID and returns a channel for the user to consume an ack/nack on to check that the event has been published.
type SubscribeClient ¶
type SubscribeClient interface {
ConnectionObserver
SubscribeStream(context.Context, ...grpc.CallOption) (api.Ensign_SubscribeClient, error)
}
type Subscriber ¶
type Subscriber struct {
// contains filtered or unexported fields
}
Subscriber wraps a stream.SubscribeClient to maintain an open subscribe stream to an Ensign node. When the subscriber is started it kicks off a go routine that watches for when the stream goes down and attempts to reconnect it gracefully. This go routine also spins off go routines for receiving messages from the stream. The received events are passed to a channel that must be consumed by the caller; if the channel fills up, the event will be dropped and a nack sent back to the server.
Sending acks/nacks back to the server happens synchronously in the user thread, an error is returned if the message cannot be sent.
func NewSubscriber ¶
func NewSubscriber(client SubscribeClient, topics []string, opts ...grpc.CallOption) (_ <-chan *api.EventWrapper, _ *Subscriber, err error)
Create a new low-level subscribe stream manager that maintains an open subscribe stream and allows user to send acks/nack back to the Ensign node. This function opens a subscribe stream and returns an error if the user is not authenticated or the stream cannot be opened. If the stream is opened successfully, the start go routine is kicked off, which ensures the stream stays open even if the remote node temporarily goes down. The start go routine also kicks off the receive routine to get events from the server, which are sent down the returned event channel.
NOTE: it is the caller's responsibility to consume the returned event channel; if the buffer gets filled up events may be dropped and nacked back to the server.
func (*Subscriber) Ack ¶
func (c *Subscriber) Ack(ack *api.Ack) error
Ack sends an acknowledgement to the server via the subscribe stream. This method blocks until a stream is available to send on and synchronously sends the ack.
func (*Subscriber) Close ¶
func (c *Subscriber) Close() error
Close the subscriber gracefully, once closed, the subscriber cannot be restarted.
func (*Subscriber) Err ¶
func (c *Subscriber) Err() error
Err returns any fatal errors that are set on the subscriber. If a non-nil error is returned then the subscriber is not running so no events will be received and no messages can be sent to the server.
Source Files