streamer

Documentation

Overview

Package streamer provides streamer client Go implementation for golang based gateways

Package streamer provides streamer client Go implementation for golang based gateways

Package streamer provides streamer client Go implementation for golang based gateways

Index

• Constants
• type Client
  • func NewStreamerClient(reg service_registry.GatewayRegistry) Client
• type Listener

Constants

const (
	StreamingInterval = time.Second * 20
)

Types

type Client

type Client interface {
	// AddListener registers a new streaming updates listener for the
	// listener.GetName() stream.
	// The stream name must be unique and AddListener will error out if a listener
	// for the same stream is already registered.
	AddListener(l Listener) error
	// Stream starts streaming loop for a registered by AddListener listener
	// If successful, Stream never return and should be called in it's own go routine or main()
	// If the provided Listener is not registered, Stream will try to register it prior to starting streaming
	Stream(l Listener) error
	// RemoveListener removes currently registered listener. It returns true is the
	// listener with provided l.Name() exists and was unregistered successfully
	// RemoveListener is the only way to terminate streaming loop
	RemoveListener(l Listener) bool
}

Streamer Client Interface The package implememntation provides NewStreamerClient(cr registry.CloudRegistry) Client method to create New streamer clients

func NewStreamerClient

func NewStreamerClient(reg service_registry.GatewayRegistry) Client

NewStreamerClient creates new streamer client with an empty listeners list The created streamer is ready to serve new listeners after they are added via AddListener() call

type Listener

type Listener interface {
	// GetName() returns name of the stream, the listener is getting updates on
	GetName() string
	// ReportError is going to be called by the streamer on every error.
	// If ReportError() will return nil, streamer will try to continue streaming
	// If ReportError() will return error != nil - streaming on the stream will be terminated
	ReportError(e error) error
	// Update will be called for every new update received from the stream
	// u is guaranteed to be of a type returned by New(), so - myUpdate := u.(MyDataType) should never panic
	// Update() returns bool indicating whether to continue streaming:
	//   true - continue streaming; false - stop streaming
	//   If Update() returns false -> ReportError() will be called with io.EOF,
	//   in this case, if ReportError() returns nil, streaming will continue with the new connection & stream
	Update(u *protos.DataUpdateBatch) bool
	// GetExtraArgs will be called prior to each stream request and its returned value will be used to initialize
	// ExtraArgs field in GetUpdates request payload. Most listeners may just return nil
	GetExtraArgs() *any.Any
}

Listener interface defines Stream Listener which will become the receiver of streaming updates for a registered stream Each received update will be unmarshalled into the Listener's update data type determined by the actual type returned by Listener's New() receiver method