submatview

Documentation

Index

• type Deps
• type Materializer
  • func NewMaterializer(deps Deps) *Materializer
  • func (m *Materializer) Run(ctx context.Context)
• type Request
• type Result
• type Store
  • func NewStore(logger hclog.Logger) *Store
  • func (s *Store) Get(ctx context.Context, req Request) (Result, error)
  • func (s *Store) Notify(ctx context.Context, req Request, correlationID string, ...) error
  • func (s *Store) Run(ctx context.Context)
• type StreamClient
• type View

Types

type Deps

type Deps struct {
	View    View
	Client  StreamClient
	Logger  hclog.Logger
	Waiter  *retry.Waiter
	Request func(index uint64) pbsubscribe.SubscribeRequest
}

type Materializer

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

Materializer consumes the event stream, handling any framing events, and sends the events to View as they are received.

Materializer is used as the cache.Result.State for a streaming cache type and manages the actual streaming RPC call to the servers behind the scenes until the cache result is discarded when TTL expires.

func NewMaterializer

func NewMaterializer(deps Deps) *Materializer

NewMaterializer returns a new Materializer. Run must be called to start it.

func (*Materializer) Run

func (m *Materializer) Run(ctx context.Context)

Run receives events from the StreamClient and sends them to the View. It runs until ctx is cancelled, so it is expected to be run in a goroutine.

type Request

type Request interface {
	cache.Request
	// NewMaterializer will be called if there is no active materializer to fulfil
	// the request. It should return a Materializer appropriate for streaming
	// data to fulfil this request.
	NewMaterializer() (*Materializer, error)
	// Type should return a string which uniquely identifies this type of request.
	// The returned value is used as the prefix of the key used to index
	// entries in the Store.
	Type() string
}

Request is used to request data from the Store. Note that cache.Request is required, but some of the fields cache.RequestInfo fields are ignored (ex: MaxAge, and MustRevalidate).

type Result

type Result struct {
	Index uint64
	Value interface{}
}

type Store

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

Store of Materializers. Store implements an interface similar to agent/cache.Cache, and allows a single Materializer to fulfil multiple requests as long as the requests are identical. Store is used in place of agent/cache.Cache because with the streaming backend there is no longer any need to run a background goroutine to refresh stored values.

func NewStore

func NewStore(logger hclog.Logger) *Store

NewStore creates and returns a Store that is ready for use. The caller must call Store.Run (likely in a separate goroutine) to start the expiration loop.

func (*Store) Get

func (s *Store) Get(ctx context.Context, req Request) (Result, error)

Get a value from the store, blocking if the store has not yet seen the req.Index value. See agent/cache.Cache.Get for complete documentation.

func (*Store) Notify

func (s *Store) Notify(
	ctx context.Context,
	req Request,
	correlationID string,
	updateCh chan<- cache.UpdateEvent,
) error

Notify the updateCh when there are updates to the entry identified by req. See agent/cache.Cache.Notify for complete documentation.

Request.CacheInfo().Timeout is ignored because it is not really relevant in this case. Instead set a deadline on the context.

func (*Store) Run

func (s *Store) Run(ctx context.Context)

Run the expiration loop until the context is cancelled.

type StreamClient

type StreamClient interface {
	Subscribe(ctx context.Context, in *pbsubscribe.SubscribeRequest, opts ...grpc.CallOption) (pbsubscribe.StateChangeSubscription_SubscribeClient, error)
}

StreamClient provides a subscription to state change events.

type View

type View interface {
	// Update is called when one or more events are received. The first call will
	// include _all_ events in the initial snapshot which may be an empty set.
	// Subsequent calls will contain one or more update events in the order they
	// are received.
	Update(events []*pbsubscribe.Event) error

	// Result returns the type-specific cache result based on the state. When no
	// events have been delivered yet the result should be an empty value type
	// suitable to return to clients in case there is an empty result on the
	// servers. The index the materialized view represents is maintained
	// separately and passed in in case the return type needs an Index field
	// populating. This allows implementations to not worry about maintaining
	// indexes seen during Update.
	Result(index uint64) interface{}

	// Reset the view to the zero state, done in preparation for receiving a new
	// snapshot.
	Reset()
}

View receives events from, and return results to, Materializer. A view is responsible for converting the pbsubscribe.Event.Payload into the local type, and storing it so that it can be returned by Result().