mirror of
https://github.com/openziti/ziti.git
synced 2026-10-26 13:39:58 +00:00
Paul Lorenz
ae806045b5
- channel.TypedReceiveHandler -> channel.ContentTypeReceiver - binding.AddTypedReceiveHandler(h) -> channel.AddReceiveHandlers(binding, h) channel/v5 repurposes TypedReceiveHandler for the senders-typed handler and replaces the self-describing pattern with ContentTypeReceiver plus the AddReceiveHandlers free function (openziti/channel#262). Mechanical conversion; does not build on its own.
75 lines
2.5 KiB
Go
75 lines
2.5 KiB
Go
package state
|
|
|
|
import (
|
|
"errors"
|
|
|
|
"github.com/michaelquigley/pfxlog"
|
|
"github.com/openziti/channel/v5"
|
|
"github.com/openziti/ziti/v2/common"
|
|
"github.com/openziti/ziti/v2/common/pb/edge_ctrl_pb"
|
|
controllerEnv "github.com/openziti/ziti/v2/controller/env"
|
|
"google.golang.org/protobuf/proto"
|
|
)
|
|
|
|
type dataStateChangeSetHandler struct {
|
|
state Manager
|
|
}
|
|
|
|
func NewDataStateEventHandler(state Manager) channel.ContentTypeReceiver {
|
|
return &dataStateChangeSetHandler{
|
|
state: state,
|
|
}
|
|
}
|
|
|
|
func (eventHandler *dataStateChangeSetHandler) HandleReceive(msg *channel.Message, ch channel.Channel) {
|
|
currentSubscription := eventHandler.state.GetCurrentDataModelSubscription()
|
|
|
|
logger := pfxlog.Logger().
|
|
WithField("ctrlId", ch.Id()).
|
|
WithField("subscribedCtrlId", currentSubscription.CtrlId).
|
|
WithField("subscriptionId", currentSubscription.SubscriptionId)
|
|
|
|
newEvent := &edge_ctrl_pb.DataState_ChangeSet{}
|
|
if err := proto.Unmarshal(msg.Body, newEvent); err != nil {
|
|
logger.WithError(err).Errorf("could not unmarshal data state change set message")
|
|
return
|
|
}
|
|
|
|
logger = logger.WithField("index", newEvent.Index).WithField("synthetic", newEvent.IsSynthetic)
|
|
|
|
// ignore state from controllers we are not currently subscribed to
|
|
if !currentSubscription.IsCurrentController(ch.Id()) && !newEvent.IsSynthetic {
|
|
logger.Info("data state change received from ctrl other than the one currently subscribed to")
|
|
return
|
|
}
|
|
|
|
subscriptionId, ok := msg.GetStringHeader(int32(edge_ctrl_pb.Header_RouterDataModelSubscriptionId))
|
|
if ok && subscriptionId != currentSubscription.SubscriptionId {
|
|
logger.WithField("eventSubscriptionId", subscriptionId).
|
|
Info("data state change received from inactive or invalid subscription")
|
|
return
|
|
}
|
|
|
|
err := eventHandler.state.GetRouterDataModelPool().QueueOrError(func() {
|
|
model := eventHandler.state.RouterDataModel()
|
|
logger.Debug("received data state change set")
|
|
if applyErr := model.ApplyChangeSet(newEvent); applyErr != nil {
|
|
var gapErr *common.GapDetectedError
|
|
if errors.As(applyErr, &gapErr) {
|
|
logger.WithError(applyErr).Warn("gap detected in data model events, triggering resubscribe")
|
|
eventHandler.state.ResubscribeRouterDataModel()
|
|
}
|
|
}
|
|
})
|
|
|
|
if err != nil {
|
|
logger.WithError(err).Errorf("could not queue processing data state change set message, resyncing full router data model")
|
|
// because we were unable to accept events, we're out of sync and will have to completely resync
|
|
eventHandler.state.ResyncRouterDataModel()
|
|
}
|
|
}
|
|
|
|
func (*dataStateChangeSetHandler) ContentType() int32 {
|
|
return controllerEnv.DataStateChangeSetType
|
|
}
|