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
83 lines
2.3 KiB
Go
83 lines
2.3 KiB
Go
package state
|
|
|
|
import (
|
|
"github.com/michaelquigley/pfxlog"
|
|
"github.com/hanzozt/channel/v4"
|
|
"github.com/hanzozt/channel/v4/protobufs"
|
|
"github.com/hanzozt/zt/v2/common"
|
|
"github.com/hanzozt/zt/v2/common/pb/edge_ctrl_pb"
|
|
controllerEnv "github.com/hanzozt/zt/v2/controller/env"
|
|
"github.com/hanzozt/zt/v2/router/env"
|
|
log "github.com/sirupsen/logrus"
|
|
"google.golang.org/protobuf/proto"
|
|
)
|
|
|
|
type ValidateDataStateRequestHandler struct {
|
|
state Manager
|
|
env env.RouterEnv
|
|
}
|
|
|
|
func NewValidateDataStateRequestHandler(state Manager, env env.RouterEnv) *ValidateDataStateRequestHandler {
|
|
return &ValidateDataStateRequestHandler{
|
|
state: state,
|
|
env: env,
|
|
}
|
|
}
|
|
|
|
func (*ValidateDataStateRequestHandler) ContentType() int32 {
|
|
return controllerEnv.ValidateDataStateType
|
|
}
|
|
|
|
func (self *ValidateDataStateRequestHandler) HandleReceive(msg *channel.Message, ch channel.Channel) {
|
|
request := &edge_ctrl_pb.RouterDataModelValidateRequest{}
|
|
|
|
if err := proto.Unmarshal(msg.Body, request); err != nil {
|
|
pfxlog.Logger().WithError(err).Errorf("could not unmarshal validate data state request")
|
|
return
|
|
}
|
|
|
|
newState := request.State
|
|
model := common.NewBareRouterDataModel()
|
|
model.WhileLocked(func(u uint64) {
|
|
for _, event := range newState.Events {
|
|
model.Handle(newState.EndIndex, event)
|
|
}
|
|
model.SetCurrentIndex(newState.EndIndex)
|
|
})
|
|
|
|
current := self.state.RouterDataModel()
|
|
model.InheritLocalData(current)
|
|
|
|
response := &edge_ctrl_pb.RouterDataModelValidateResponse{
|
|
OrigEntityCounts: model.GetEntityCounts(),
|
|
CopyEntityCounts: current.GetEntityCounts(),
|
|
}
|
|
|
|
reportedF := func(entityType string, id string, diffType common.DiffType, detail string) {
|
|
response.Diffs = append(response.Diffs, &edge_ctrl_pb.RouterDataModelDiff{
|
|
EntityType: entityType,
|
|
EntityId: id,
|
|
DiffType: string(diffType),
|
|
Detail: detail,
|
|
})
|
|
}
|
|
|
|
current.Validate(model, reportedF)
|
|
|
|
if len(response.Diffs) > 0 && request.Fix {
|
|
model = common.NewReceiverRouterDataModelFromExisting(model, self.state.GetEnv().GetCloseNotify())
|
|
self.state.SetRouterDataModel(model, true)
|
|
}
|
|
|
|
go func() {
|
|
err := protobufs.MarshalTyped(response).
|
|
ReplyTo(msg).
|
|
WithTimeout(self.env.DefaultRequestTimeout()).
|
|
SendAndWaitForWire(ch)
|
|
|
|
if err != nil {
|
|
log.WithError(err).Error("failed to send validate router data model response")
|
|
}
|
|
}()
|
|
}
|