2a4ca71d3f
Previously we had 1 EventPublisher per state.Store. When a state store was closed/abandoned such as during a consul snapshot restore, this had the behavior of force closing subscriptions for that topic and evicting event snapshots from the cache. The intention of this commit is to keep all that behavior. To that end, the shared EventPublisher now supports the ability to refresh a topic. That will perform the force close + eviction. The FSM upon abandoning the previous state.Store will call RefreshTopic for all the topics with events generated by the state.Store.
36 lines
1,023 B
Go
36 lines
1,023 B
Go
package consul
|
|
|
|
import (
|
|
"google.golang.org/grpc"
|
|
|
|
"github.com/hashicorp/consul/acl"
|
|
"github.com/hashicorp/consul/agent/consul/stream"
|
|
"github.com/hashicorp/consul/agent/grpc/private/services/subscribe"
|
|
"github.com/hashicorp/consul/agent/structs"
|
|
)
|
|
|
|
type subscribeBackend struct {
|
|
srv *Server
|
|
connPool GRPCClientConner
|
|
}
|
|
|
|
// TODO: refactor Resolve methods to an ACLBackend that can be used by all
|
|
// the endpoints.
|
|
func (s subscribeBackend) ResolveTokenAndDefaultMeta(
|
|
token string,
|
|
entMeta *acl.EnterpriseMeta,
|
|
authzContext *acl.AuthorizerContext,
|
|
) (acl.Authorizer, error) {
|
|
return s.srv.ResolveTokenAndDefaultMeta(token, entMeta, authzContext)
|
|
}
|
|
|
|
var _ subscribe.Backend = (*subscribeBackend)(nil)
|
|
|
|
func (s subscribeBackend) Forward(info structs.RPCInfo, f func(*grpc.ClientConn) error) (handled bool, err error) {
|
|
return s.srv.ForwardGRPC(s.connPool, info, f)
|
|
}
|
|
|
|
func (s subscribeBackend) Subscribe(req *stream.SubscribeRequest) (*stream.Subscription, error) {
|
|
return s.srv.publisher.Subscribe(req)
|
|
}
|