mirror of
https://github.com/hanzozt/zt.git
synced 2026-10-30 15:04:55 +00:00
zeekay
241625e7d9
- Repo names: ziti-* → zt-* - Go modules: github.com/hanzozt/ziti → github.com/hanzozt/zt - Docker images: hanzozt/ziti-* → hanzozt/zt-* - C/Java/JS symbols: ziti_* → zt_* - File and directory names updated
66 lines
2.2 KiB
Go
66 lines
2.2 KiB
Go
package state
|
|
|
|
import (
|
|
"github.com/michaelquigley/pfxlog"
|
|
"github.com/hanzozt/channel/v4"
|
|
"github.com/hanzozt/zt/v2/common/pb/edge_ctrl_pb"
|
|
controllerEnv "github.com/hanzozt/zt/v2/controller/env"
|
|
"google.golang.org/protobuf/proto"
|
|
)
|
|
|
|
type dataStateChangeSetHandler struct {
|
|
state Manager
|
|
}
|
|
|
|
func NewDataStateEventHandler(state Manager) channel.TypedReceiveHandler {
|
|
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")
|
|
model.ApplyChangeSet(newEvent)
|
|
})
|
|
|
|
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
|
|
}
|