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
68 lines
2.2 KiB
Go
68 lines
2.2 KiB
Go
package state
|
|
|
|
import (
|
|
"github.com/michaelquigley/pfxlog"
|
|
"github.com/hanzozt/channel/v4"
|
|
"github.com/hanzozt/zt/v2/common"
|
|
"github.com/hanzozt/zt/v2/common/pb/edge_ctrl_pb"
|
|
controllerEnv "github.com/hanzozt/zt/v2/controller/env"
|
|
"google.golang.org/protobuf/proto"
|
|
)
|
|
|
|
type DataStateHandler struct {
|
|
state Manager
|
|
}
|
|
|
|
func NewDataStateHandler(state Manager) *DataStateHandler {
|
|
return &DataStateHandler{
|
|
state: state,
|
|
}
|
|
}
|
|
|
|
func (*DataStateHandler) ContentType() int32 {
|
|
return controllerEnv.DataStateType
|
|
}
|
|
|
|
func (self *DataStateHandler) HandleReceive(msg *channel.Message, ch channel.Channel) {
|
|
currentSubscription := self.state.GetCurrentDataModelSubscription()
|
|
|
|
logger := pfxlog.Logger().
|
|
WithField("ctrlId", ch.Id()).
|
|
WithField("subscribedCtrlId", currentSubscription.CtrlId).
|
|
WithField("subscriptionId", currentSubscription.SubscriptionId)
|
|
|
|
// ignore state from controllers we are not currently subscribed to
|
|
if !currentSubscription.IsCurrentController(ch.Id()) {
|
|
logger.Info("data state 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 received from inactive or invalid subscription")
|
|
return
|
|
}
|
|
|
|
err := self.state.GetRouterDataModelPool().QueueOrError(func() {
|
|
newState := &edge_ctrl_pb.DataState{}
|
|
if err := proto.Unmarshal(msg.Body, newState); err != nil {
|
|
logger.WithError(err).Errorf("could not unmarshal data state event message")
|
|
return
|
|
}
|
|
|
|
logger.WithField("index", newState.EndIndex).Info("received full router data model state")
|
|
|
|
model := common.NewReceiverRouterDataModelFromDataState(newState, self.state.GetEnv().GetCloseNotify())
|
|
self.state.SetRouterDataModel(model, false)
|
|
|
|
logger.WithField("index", newState.EndIndex).Info("finished processing full router data model state")
|
|
})
|
|
|
|
if err != nil {
|
|
pfxlog.Logger().WithError(err).Error("could not queue processing of full router data model state, requesting re-sync")
|
|
// because we were unable to accept events, we're out of sync and will have to completely resync
|
|
self.state.ResyncRouterDataModel()
|
|
}
|
|
}
|