watch

Documentation

Index

• Variables
• func ServerLocalBlockingQuery[ResultType any, StoreType StateStore](ctx context.Context, getStore func() StoreType, minIndex uint64, ...) (uint64, ResultType, error)
• func ServerLocalNotify[ResultType any, StoreType StateStore](ctx context.Context, correlationID string, getStore func() StoreType, ...) error
• type StateStore

Variables

var (
	ErrorNotFound   = errors.New("no data found for query")
	ErrorNotChanged = errors.New("data did not change for query")
)

Functions

func ServerLocalBlockingQuery

func ServerLocalBlockingQuery[ResultType any, StoreType StateStore](
	ctx context.Context,
	getStore func() StoreType,
	minIndex uint64,
	suppressSpuriousWakeup bool,
	query func(memdb.WatchSet, StoreType) (uint64, ResultType, error),
) (uint64, ResultType, error)

ServerLocalBlockingQuery performs a blocking query similar to the pre-existing blockingQuery method on the agent/consul.Server type. There are a few key differences.

1. This function makes use of Go 1.18 generics. The function is parameterized with two types. The first is the ResultType which can be anything. Having this be parameterized instead of using interface{} allows us to simplify the call sites so that no type coercion from interface{} to the real type is necessary. The second parameterized type is something that VERY loosely resembles a agent/consul/state.Store type. The StateStore interface in this package has a single method to get the stores abandon channel so we know when a snapshot restore is occurring and can act accordingly. We could have not parameterized this type and used a real *state.Store instead but then we would have concrete dependencies on the state package and it would make it a little harder to test this function.

   We could have also avoided the need to use a ResultType parameter by taking the route the original blockingQuery method did and to just assume all callers close around a pointer to their results and can modify it as necessary. That way of doing things feels a little gross so I have taken this one a different direction. The old way also gets especially gross with how we have to push concerns of spurious wakeup suppression down into every call site.

2. This method has no internal timeout and can potentially run forever until a state change is observed. If there is a desire to have a timeout, that should be built into the context.Context passed as the first argument.

3. This method bakes in some newer functionality around hashing of results to prevent sending back data when nothing has actually changed. With the old blockingQuery method this has to be done within the closure passed to the method which means the same bit of code is duplicated in many places. As this functionality isn't necessary in many scenarios whether to opt-in to that behavior is a argument to this function.

Similar to the older method:

1. Errors returned from the query will be propagated back to the caller.

The query function must follow these rules:

1. To access data it must use the passed in StoreType (which will be a state.Store when everything gets stiched together outside of unit tests).
2. It must return an index greater than the minIndex if the results returned by the query have changed.
3. Any channels added to the memdb.WatchSet must unblock when the results returned by the query have changed.

To ensure optimal performance of the query, the query function should make a best-effort attempt to follow these guidelines:

1. Only return an index greater than the minIndex.
2. Any channels added to the memdb.WatchSet should only unblock when the results returned by the query have changed. This might be difficult to do when blocking on non-existent data.

func ServerLocalNotify

func ServerLocalNotify[ResultType any, StoreType StateStore](
	ctx context.Context,
	correlationID string,
	getStore func() StoreType,
	query func(memdb.WatchSet, StoreType) (uint64, ResultType, error),
	notify func(ctx context.Context, correlationID string, result ResultType, err error),
) error

ServerLocalNotify will watch for changes in the State Store using the provided query function and invoke the notify callback whenever the results of that query function have changed. This function will return an error if parameter validations fail but otherwise the background go routine to process the notifications will be spawned and nil will be returned. Just like ServerLocalBlockingQuery this makes use of Go Generics and for the same reasons as outlined in the documentation for that function.

Types

type StateStore

type StateStore interface {
	AbandonCh() <-chan struct{}
}