diff --git a/controller/sync_strats/rtx.go b/controller/sync_strats/rtx.go index 0022c55ad..2b2b5ed97 100644 --- a/controller/sync_strats/rtx.go +++ b/controller/sync_strats/rtx.go @@ -28,9 +28,9 @@ import ( "sync" ) -// routerTx represents a connection from an Edge Router to the controller. Used +// RouterSender represents a connection from an Edge Router to the controller. Used // to asynchronously buffer and send messages to an Edge Router via Start() then Send() -type routerTx struct { +type RouterSender struct { Id string EdgeRouter *model.EdgeRouter Router *network.Router @@ -41,8 +41,8 @@ type routerTx struct { stopping concurrenz.AtomicBoolean } -func newRouterTx(edgeRouter *model.EdgeRouter, router *network.Router, sendBufferSize int) *routerTx { - return &routerTx{ +func newRouterTx(edgeRouter *model.EdgeRouter, router *network.Router, sendBufferSize int) *RouterSender { + return &RouterSender{ Id: eid.New(), EdgeRouter: edgeRouter, Router: router, @@ -54,13 +54,13 @@ func newRouterTx(edgeRouter *model.EdgeRouter, router *network.Router, sendBuffe } } -func (rtx *routerTx) Start() { +func (rtx *RouterSender) Start() { if rtx.running.CompareAndSwap(false, true) { go rtx.run() } } -func (rtx *routerTx) Stop() { +func (rtx *RouterSender) Stop() { if rtx.stopping.CompareAndSwap(false, true) { go func() { rtx.stop <- struct{}{} @@ -68,7 +68,7 @@ func (rtx *routerTx) Stop() { } } -func (rtx *routerTx) run() { +func (rtx *RouterSender) run() { for { select { case <-rtx.stop: @@ -83,7 +83,7 @@ func (rtx *routerTx) run() { } } -func (rtx *routerTx) logger() *logrus.Entry { +func (rtx *RouterSender) logger() *logrus.Entry { return pfxlog.Logger(). WithField("routerTxId", rtx.Id). WithField("routerId", rtx.Router.Id). @@ -92,26 +92,26 @@ func (rtx *routerTx) logger() *logrus.Entry { WithField("routerChannelIsOpen", !rtx.Router.Control.IsClosed()) } -func (rtx *routerTx) Send(msg *channel2.Message) { +func (rtx *RouterSender) Send(msg *channel2.Message) { rtx.send <- msg } -// Map used make working with internal routerTx easier as sync.Map accepts and returns interface{} +// Map used make working with internal RouterSender easier as sync.Map accepts and returns interface{} type routerTxMap struct { - internalMap *sync.Map //id -> routerTx + internalMap *sync.Map //id -> RouterSender } -func (m *routerTxMap) Add(id string, routerMessageTxer *routerTx) { +func (m *routerTxMap) Add(id string, routerMessageTxer *RouterSender) { m.internalMap.Store(id, routerMessageTxer) routerMessageTxer.Start() } -func (m *routerTxMap) Get(id string) *routerTx { +func (m *routerTxMap) Get(id string) *RouterSender { val, found := m.internalMap.Load(id) if !found { return nil } - return val.(*routerTx) + return val.(*RouterSender) } func (m *routerTxMap) Remove(id string) { @@ -122,9 +122,9 @@ func (m *routerTxMap) Remove(id string) { } } -func (m *routerTxMap) Range(f func(entries *routerTx) bool) { +func (m *routerTxMap) Range(f func(entries *RouterSender) bool) { m.internalMap.Range(func(edgeRouterId, value interface{}) bool { - if rtx, ok := value.(*routerTx); ok { + if rtx, ok := value.(*RouterSender); ok { return f(rtx) } pfxlog.Logger().Panic("could not convert edge router entry") diff --git a/controller/sync_strats/sync_instant.go b/controller/sync_strats/sync_instant.go index c20e0d275..c7ecc044d 100644 --- a/controller/sync_strats/sync_instant.go +++ b/controller/sync_strats/sync_instant.go @@ -54,8 +54,8 @@ type InstantStrategy struct { resyncHandler channel2.ReceiveHandler ae *env.AppEnv - helloOutQueue chan *routerTx - helloInQueue chan *routerTx + helloOutQueue chan *RouterSender + helloInQueue chan *RouterSender stopHelloOut chan struct{} stopHelloIn chan struct{} @@ -99,8 +99,8 @@ func NewInstantStrategy(ae *env.AppEnv, options InstantStrategyOptions) *Instant internalMap: &sync.Map{}, }, ae: ae, - helloOutQueue: make(chan *routerTx, 100), - helloInQueue: make(chan *routerTx, 100), + helloOutQueue: make(chan *RouterSender, 100), + helloInQueue: make(chan *RouterSender, 100), stopHelloOut: make(chan struct{}, 0), stopHelloIn: make(chan struct{}, 0), } @@ -166,7 +166,7 @@ func (strategy *InstantStrategy) ApiSessionAdded(apiSession *persistence.ApiSess Sequence: 0, } - strategy.rtxMap.Range(func(rtx *routerTx) bool { + strategy.rtxMap.Range(func(rtx *RouterSender) bool { strategy.sendApiSessionAdded(rtx, false, state, []*edge_ctrl_pb.ApiSession{apiSessionProto}) return true }) @@ -188,7 +188,7 @@ func (strategy *InstantStrategy) ApiSessionUpdated(apiSession *persistence.ApiSe ApiSessions: []*edge_ctrl_pb.ApiSession{apiSessionProto}, } - strategy.rtxMap.Range(func(rtx *routerTx) bool { + strategy.rtxMap.Range(func(rtx *RouterSender) bool { content, _ := proto.Marshal(apiSessionAdded) msg := channel2.NewMessage(env.ApiSessionUpdatedType, content) msg.Headers[env.SyncStrategyTypeHeader] = []byte(strategy.Type()) @@ -203,7 +203,7 @@ func (strategy *InstantStrategy) ApiSessionDeleted(apiSession *persistence.ApiSe Tokens: []string{apiSession.Token}, } - strategy.rtxMap.Range(func(rtx *routerTx) bool { + strategy.rtxMap.Range(func(rtx *RouterSender) bool { content, _ := proto.Marshal(sessionRemoved) msg := channel2.NewMessage(env.ApiSessionRemovedType, content) rtx.Send(msg) @@ -227,7 +227,7 @@ func (strategy *InstantStrategy) SessionAdded(session *persistence.Session) { IsLast: true, Sequence: 0, } - strategy.rtxMap.Range(func(rtx *routerTx) bool { + strategy.rtxMap.Range(func(rtx *RouterSender) bool { strategy.sendSessionAdded(rtx, false, state, []*edge_ctrl_pb.Session{sessionProto}) return true }) @@ -238,7 +238,7 @@ func (strategy *InstantStrategy) SessionDeleted(session *persistence.Session) { Tokens: []string{session.Token}, } - strategy.rtxMap.Range(func(rtx *routerTx) bool { + strategy.rtxMap.Range(func(rtx *RouterSender) bool { content, _ := proto.Marshal(sessionRemoved) msg := channel2.NewMessage(env.SessionRemovedType, content) rtx.Send(msg) @@ -264,7 +264,7 @@ func (strategy *InstantStrategy) startSynchronizeWorker() { } } -func (strategy *InstantStrategy) hello(rtx *routerTx) { +func (strategy *InstantStrategy) hello(rtx *RouterSender) { logger := rtx.logger().WithField("strategy", strategy.Type()) logger.Info("edge router sync starting") @@ -281,7 +281,7 @@ func (strategy *InstantStrategy) hello(rtx *routerTx) { strategy.sendHello(rtx) } -func (strategy *InstantStrategy) sendHello(rtx *routerTx) { +func (strategy *InstantStrategy) sendHello(rtx *RouterSender) { logger := rtx.logger().WithField("strategy", strategy.Type()) serverVersion := build.GetBuildInfo().Version() serverHello := &edge_ctrl_pb.ServerHello{ @@ -373,7 +373,7 @@ func (strategy *InstantStrategy) ReceiveHello(r *network.Router, respHello *edge strategy.helloInQueue <- rtx } -func (strategy *InstantStrategy) synchronize(rtx *routerTx) { +func (strategy *InstantStrategy) synchronize(rtx *RouterSender) { defer func() { rtx.logger().WithField("strategy", strategy.Type()).Infof("exiting synchronization, final status: %s", rtx.Status) }() @@ -479,7 +479,7 @@ func (strategy *InstantStrategy) synchronize(rtx *routerTx) { rtx.Status = env.RouterSyncDone } -func (strategy *InstantStrategy) sendApiSessionAdded(rtx *routerTx, isFullState bool, state *InstantSyncState, apiSessions []*edge_ctrl_pb.ApiSession) { +func (strategy *InstantStrategy) sendApiSessionAdded(rtx *RouterSender, isFullState bool, state *InstantSyncState, apiSessions []*edge_ctrl_pb.ApiSession) { stateBytes, _ := json.Marshal(state) msgContent := &edge_ctrl_pb.ApiSessionAdded{ @@ -497,7 +497,7 @@ func (strategy *InstantStrategy) sendApiSessionAdded(rtx *routerTx, isFullState rtx.Send(msg) } -func (strategy *InstantStrategy) sendSessionAdded(rtx *routerTx, isFullState bool, state *InstantSyncState, sessions []*edge_ctrl_pb.Session) { +func (strategy *InstantStrategy) sendSessionAdded(rtx *RouterSender, isFullState bool, state *InstantSyncState, sessions []*edge_ctrl_pb.Session) { stateBytes, _ := json.Marshal(state) diff --git a/router/handler_edge_ctrl/apiSessionAdded.go b/router/handler_edge_ctrl/apiSessionAdded.go index 0c72b2d8b..551b98c8a 100644 --- a/router/handler_edge_ctrl/apiSessionAdded.go +++ b/router/handler_edge_ctrl/apiSessionAdded.go @@ -274,10 +274,6 @@ type apiSessionAddedWithState struct { *edge_ctrl_pb.ApiSessionAdded } -type apiSessionSyncResult struct { - apiSessions []*apiSessionAddedWithState -} - func parseInstantSyncHeaders(msg *channel2.Message) (string, *sync_strats.InstantSyncState, error) { if syncStrategyType, ok := msg.Headers[env.SyncStrategyTypeHeader]; ok { if syncStrategyState, ok := msg.Headers[env.SyncStrategyStateHeader]; ok {