diff --git a/controller/env/broker.go b/controller/env/broker.go index 3c5b84b12..5e198511e 100644 --- a/controller/env/broker.go +++ b/controller/env/broker.go @@ -19,6 +19,7 @@ package env import ( "github.com/kataras/go-events" "github.com/michaelquigley/pfxlog" + "github.com/openziti/channel" "github.com/openziti/edge/controller/persistence" "github.com/openziti/edge/pb/edge_ctrl_pb" "github.com/openziti/fabric/controller/network" @@ -74,6 +75,10 @@ func NewBroker(ae *AppEnv, synchronizer RouterSyncStrategy) *Broker { return broker } +func (broker *Broker) GetReceiveHandlers() []channel.TypedReceiveHandler { + return broker.routerSyncStrategy.GetReceiveHandlers() +} + func (broker *Broker) RouterConnected(router *network.Router) { go func() { log := pfxlog.Logger().WithField("routerId", router.Id).WithField("routerName", router.Name).WithField("routerFingerprint", router.Fingerprint) diff --git a/controller/env/sync.go b/controller/env/sync.go index 0e80495fb..1f6e030da 100644 --- a/controller/env/sync.go +++ b/controller/env/sync.go @@ -17,6 +17,7 @@ package env import ( + "github.com/openziti/channel" "github.com/openziti/edge/controller/model" "github.com/openziti/edge/controller/persistence" "github.com/openziti/fabric/controller/network" @@ -66,6 +67,7 @@ type RouterSyncStrategy interface { type RouterConnectionHandler interface { RouterConnected(edgeRouter *model.EdgeRouter, router *network.Router) RouterDisconnected(router *network.Router) + GetReceiveHandlers() []channel.TypedReceiveHandler } // RouterSynchronizerEventHandler is responsible for keeping Edge Routers up to date on API Sessions @@ -132,7 +134,7 @@ type LockingRouterState struct { func NewLockingRouterStatus() *LockingRouterState { return &LockingRouterState{ internal: NewRouterStatusValues(), - lock: sync.Mutex{}, + lock: sync.Mutex{}, } } diff --git a/controller/handler_edge_ctrl/common.go b/controller/handler_edge_ctrl/common.go index acdda1dd7..bbdfc8c90 100644 --- a/controller/handler_edge_ctrl/common.go +++ b/controller/handler_edge_ctrl/common.go @@ -12,7 +12,7 @@ import ( "github.com/openziti/fabric/controller/network" "github.com/openziti/fabric/controller/xt" "github.com/openziti/fabric/logcontext" - "github.com/openziti/foundation/channel" + "github.com/openziti/channel" "github.com/openziti/foundation/identity/identity" "github.com/openziti/foundation/storage/boltz" "github.com/openziti/foundation/util/stringz" diff --git a/controller/handler_edge_ctrl/create_circuit.go b/controller/handler_edge_ctrl/create_circuit.go index 2a2bf5495..05cbe341e 100644 --- a/controller/handler_edge_ctrl/create_circuit.go +++ b/controller/handler_edge_ctrl/create_circuit.go @@ -19,17 +19,17 @@ package handler_edge_ctrl import ( "github.com/golang/protobuf/proto" "github.com/michaelquigley/pfxlog" + "github.com/openziti/channel" "github.com/openziti/edge/controller/env" "github.com/openziti/edge/controller/persistence" "github.com/openziti/edge/pb/edge_ctrl_pb" - "github.com/openziti/foundation/channel" ) type createCircuitHandler struct { baseRequestHandler } -func NewCreateCircuitHandler(appEnv *env.AppEnv, ch channel.Channel) channel.ReceiveHandler { +func NewCreateCircuitHandler(appEnv *env.AppEnv, ch channel.Channel) channel.TypedReceiveHandler { return &createCircuitHandler{ baseRequestHandler: baseRequestHandler{ ch: ch, diff --git a/controller/handler_edge_ctrl/create_terminator.go b/controller/handler_edge_ctrl/create_terminator.go index f436785c1..cdb581e29 100644 --- a/controller/handler_edge_ctrl/create_terminator.go +++ b/controller/handler_edge_ctrl/create_terminator.go @@ -20,12 +20,12 @@ import ( "fmt" "github.com/golang/protobuf/proto" "github.com/michaelquigley/pfxlog" + "github.com/openziti/channel" "github.com/openziti/edge/controller/env" "github.com/openziti/edge/controller/persistence" "github.com/openziti/edge/edge_common" "github.com/openziti/edge/pb/edge_ctrl_pb" "github.com/openziti/fabric/controller/network" - "github.com/openziti/foundation/channel" "github.com/openziti/foundation/storage/boltz" "go.etcd.io/bbolt" "math" @@ -35,7 +35,7 @@ type createTerminatorHandler struct { baseRequestHandler } -func NewCreateTerminatorHandler(appEnv *env.AppEnv, ch channel.Channel) channel.ReceiveHandler { +func NewCreateTerminatorHandler(appEnv *env.AppEnv, ch channel.Channel) channel.TypedReceiveHandler { return &createTerminatorHandler{ baseRequestHandler{ ch: ch, diff --git a/controller/handler_edge_ctrl/create_tunnel_api_session.go b/controller/handler_edge_ctrl/create_tunnel_api_session.go index ae8ec25f4..de1617e5d 100644 --- a/controller/handler_edge_ctrl/create_tunnel_api_session.go +++ b/controller/handler_edge_ctrl/create_tunnel_api_session.go @@ -2,9 +2,9 @@ package handler_edge_ctrl import ( "github.com/golang/protobuf/proto" + "github.com/openziti/channel" "github.com/openziti/edge/controller/env" "github.com/openziti/edge/pb/edge_ctrl_pb" - "github.com/openziti/foundation/channel" "github.com/sirupsen/logrus" ) @@ -13,7 +13,7 @@ type createApiSessionHandler struct { *TunnelState } -func NewCreateApiSessionHandler(appEnv *env.AppEnv, ch channel.Channel, tunnelState *TunnelState) channel.ReceiveHandler { +func NewCreateApiSessionHandler(appEnv *env.AppEnv, ch channel.Channel, tunnelState *TunnelState) channel.TypedReceiveHandler { return &createApiSessionHandler{ baseRequestHandler: baseRequestHandler{ch: ch, appEnv: appEnv}, TunnelState: tunnelState, diff --git a/controller/handler_edge_ctrl/create_tunnel_circuit.go b/controller/handler_edge_ctrl/create_tunnel_circuit.go index dc14f524d..084911d94 100644 --- a/controller/handler_edge_ctrl/create_tunnel_circuit.go +++ b/controller/handler_edge_ctrl/create_tunnel_circuit.go @@ -19,10 +19,10 @@ package handler_edge_ctrl import ( "github.com/golang/protobuf/proto" "github.com/michaelquigley/pfxlog" + "github.com/openziti/channel" "github.com/openziti/edge/controller/env" "github.com/openziti/edge/controller/persistence" "github.com/openziti/edge/pb/edge_ctrl_pb" - "github.com/openziti/foundation/channel" ) type createCircuitForServiceHandler struct { @@ -30,7 +30,7 @@ type createCircuitForServiceHandler struct { *TunnelState } -func NewCreateCircuitForTunnelHandler(appEnv *env.AppEnv, ch channel.Channel, tunnelState *TunnelState) channel.ReceiveHandler { +func NewCreateCircuitForTunnelHandler(appEnv *env.AppEnv, ch channel.Channel, tunnelState *TunnelState) channel.TypedReceiveHandler { return &createCircuitForServiceHandler{ baseRequestHandler: baseRequestHandler{ch: ch, appEnv: appEnv}, TunnelState: tunnelState, diff --git a/controller/handler_edge_ctrl/create_tunnel_terminator.go b/controller/handler_edge_ctrl/create_tunnel_terminator.go index bf3adf397..b6497b34c 100644 --- a/controller/handler_edge_ctrl/create_tunnel_terminator.go +++ b/controller/handler_edge_ctrl/create_tunnel_terminator.go @@ -20,12 +20,12 @@ import ( "fmt" "github.com/golang/protobuf/proto" "github.com/michaelquigley/pfxlog" + "github.com/openziti/channel" "github.com/openziti/edge/controller/env" "github.com/openziti/edge/controller/persistence" "github.com/openziti/edge/edge_common" "github.com/openziti/edge/pb/edge_ctrl_pb" "github.com/openziti/fabric/controller/network" - "github.com/openziti/foundation/channel" "github.com/sirupsen/logrus" "math" ) @@ -35,7 +35,7 @@ type createTunnelTerminatorHandler struct { *TunnelState } -func NewCreateTunnelTerminatorHandler(appEnv *env.AppEnv, ch channel.Channel, tunnelState *TunnelState) channel.ReceiveHandler { +func NewCreateTunnelTerminatorHandler(appEnv *env.AppEnv, ch channel.Channel, tunnelState *TunnelState) channel.TypedReceiveHandler { return &createTunnelTerminatorHandler{ baseRequestHandler: baseRequestHandler{ch: ch, appEnv: appEnv}, TunnelState: tunnelState, diff --git a/controller/handler_edge_ctrl/extend_enrollment.go b/controller/handler_edge_ctrl/extend_enrollment.go index 1cf955b04..633397ecb 100644 --- a/controller/handler_edge_ctrl/extend_enrollment.go +++ b/controller/handler_edge_ctrl/extend_enrollment.go @@ -19,11 +19,11 @@ package handler_edge_ctrl import ( "github.com/golang/protobuf/proto" "github.com/michaelquigley/pfxlog" + "github.com/openziti/channel" "github.com/openziti/edge/controller/env" "github.com/openziti/edge/controller/model" "github.com/openziti/edge/internal/cert" "github.com/openziti/edge/pb/edge_ctrl_pb" - "github.com/openziti/foundation/channel" ) type extendEnrollmentHandler struct { diff --git a/controller/handler_edge_ctrl/extend_enrollment_verify.go b/controller/handler_edge_ctrl/extend_enrollment_verify.go index 5a27ccd30..c7f4b5800 100644 --- a/controller/handler_edge_ctrl/extend_enrollment_verify.go +++ b/controller/handler_edge_ctrl/extend_enrollment_verify.go @@ -22,7 +22,7 @@ import ( "github.com/michaelquigley/pfxlog" "github.com/openziti/edge/controller/env" "github.com/openziti/edge/pb/edge_ctrl_pb" - "github.com/openziti/foundation/channel" + "github.com/openziti/channel" ) type extendEnrollmentVerifyHandler struct { diff --git a/controller/handler_edge_ctrl/health_event.go b/controller/handler_edge_ctrl/health_event.go index 4eb2fb6b0..61ac56dce 100644 --- a/controller/handler_edge_ctrl/health_event.go +++ b/controller/handler_edge_ctrl/health_event.go @@ -19,10 +19,10 @@ package handler_edge_ctrl import ( "github.com/golang/protobuf/proto" "github.com/michaelquigley/pfxlog" + "github.com/openziti/channel" "github.com/openziti/edge/controller/env" "github.com/openziti/edge/controller/persistence" "github.com/openziti/edge/pb/edge_ctrl_pb" - "github.com/openziti/foundation/channel" "github.com/openziti/foundation/metrics" "time" ) @@ -33,7 +33,7 @@ type healthEventHandler struct { serviceHealthCheckFailedCounter metrics.IntervalCounter } -func NewHealthEventHandler(appEnv *env.AppEnv, ch channel.Channel) channel.ReceiveHandler { +func NewHealthEventHandler(appEnv *env.AppEnv, ch channel.Channel) channel.TypedReceiveHandler { serviceEventMetrics := appEnv.GetHostController().GetNetwork().GetServiceEventsMetricsRegistry() return &healthEventHandler{ baseRequestHandler: baseRequestHandler{ diff --git a/controller/handler_edge_ctrl/heartbeat.go b/controller/handler_edge_ctrl/heartbeat.go index 28425e15c..d0894c167 100644 --- a/controller/handler_edge_ctrl/heartbeat.go +++ b/controller/handler_edge_ctrl/heartbeat.go @@ -21,7 +21,7 @@ import ( "github.com/michaelquigley/pfxlog" "github.com/openziti/edge/controller/env" "github.com/openziti/edge/pb/edge_ctrl_pb" - "github.com/openziti/foundation/channel" + "github.com/openziti/channel" ) type sessionHeartbeatHandler struct { diff --git a/controller/handler_edge_ctrl/hello.go b/controller/handler_edge_ctrl/hello.go index 3b6a3b635..f1b3b064c 100644 --- a/controller/handler_edge_ctrl/hello.go +++ b/controller/handler_edge_ctrl/hello.go @@ -19,10 +19,10 @@ package handler_edge_ctrl import ( "github.com/golang/protobuf/proto" "github.com/michaelquigley/pfxlog" + "github.com/openziti/channel" "github.com/openziti/edge/controller/env" "github.com/openziti/edge/pb/edge_ctrl_pb" "github.com/openziti/fabric/controller/network" - "github.com/openziti/foundation/channel" ) type helloHandler struct { diff --git a/controller/handler_edge_ctrl/list_tunnel_services.go b/controller/handler_edge_ctrl/list_tunnel_services.go index 48c2f4f1c..547e44b3c 100644 --- a/controller/handler_edge_ctrl/list_tunnel_services.go +++ b/controller/handler_edge_ctrl/list_tunnel_services.go @@ -3,9 +3,9 @@ package handler_edge_ctrl import ( "encoding/json" "github.com/golang/protobuf/proto" + "github.com/openziti/channel" "github.com/openziti/edge/controller/env" "github.com/openziti/edge/pb/edge_ctrl_pb" - "github.com/openziti/foundation/channel" "github.com/openziti/foundation/storage/ast" "github.com/sirupsen/logrus" "time" @@ -16,7 +16,7 @@ type listTunnelServicesHandler struct { *TunnelState } -func NewListTunnelServicesHandler(appEnv *env.AppEnv, ch channel.Channel, tunnelState *TunnelState) channel.ReceiveHandler { +func NewListTunnelServicesHandler(appEnv *env.AppEnv, ch channel.Channel, tunnelState *TunnelState) channel.TypedReceiveHandler { return &listTunnelServicesHandler{ baseRequestHandler: baseRequestHandler{ch: ch, appEnv: appEnv}, TunnelState: tunnelState, diff --git a/controller/handler_edge_ctrl/remove_terminator.go b/controller/handler_edge_ctrl/remove_terminator.go index 3f14fd15f..a1dbef9df 100644 --- a/controller/handler_edge_ctrl/remove_terminator.go +++ b/controller/handler_edge_ctrl/remove_terminator.go @@ -19,11 +19,11 @@ package handler_edge_ctrl import ( "github.com/golang/protobuf/proto" "github.com/michaelquigley/pfxlog" + "github.com/openziti/channel" "github.com/openziti/edge/controller/env" "github.com/openziti/edge/controller/persistence" "github.com/openziti/edge/edge_common" "github.com/openziti/edge/pb/edge_ctrl_pb" - "github.com/openziti/foundation/channel" "github.com/sirupsen/logrus" ) @@ -31,7 +31,7 @@ type removeTerminatorHandler struct { baseRequestHandler } -func NewRemoveTerminatorHandler(appEnv *env.AppEnv, ch channel.Channel) channel.ReceiveHandler { +func NewRemoveTerminatorHandler(appEnv *env.AppEnv, ch channel.Channel) channel.TypedReceiveHandler { return &removeTerminatorHandler{ baseRequestHandler{ ch: ch, diff --git a/controller/handler_edge_ctrl/remove_tunnel_terminator.go b/controller/handler_edge_ctrl/remove_tunnel_terminator.go index 531d50052..4386c33d0 100644 --- a/controller/handler_edge_ctrl/remove_tunnel_terminator.go +++ b/controller/handler_edge_ctrl/remove_tunnel_terminator.go @@ -17,10 +17,10 @@ package handler_edge_ctrl import ( + "github.com/openziti/channel" "github.com/openziti/edge/controller/env" "github.com/openziti/edge/edge_common" "github.com/openziti/edge/pb/edge_ctrl_pb" - "github.com/openziti/foundation/channel" "github.com/sirupsen/logrus" ) @@ -28,7 +28,7 @@ type removeTunnelTerminatorHandler struct { baseRequestHandler } -func NewRemoveTunnelTerminatorHandler(appEnv *env.AppEnv, ch channel.Channel) channel.ReceiveHandler { +func NewRemoveTunnelTerminatorHandler(appEnv *env.AppEnv, ch channel.Channel) channel.TypedReceiveHandler { return &removeTunnelTerminatorHandler{ baseRequestHandler{ ch: ch, diff --git a/controller/handler_edge_ctrl/resync.go b/controller/handler_edge_ctrl/resync.go index 53af17d93..9e7870f1f 100644 --- a/controller/handler_edge_ctrl/resync.go +++ b/controller/handler_edge_ctrl/resync.go @@ -19,10 +19,10 @@ package handler_edge_ctrl import ( "github.com/golang/protobuf/proto" "github.com/michaelquigley/pfxlog" + "github.com/openziti/channel" "github.com/openziti/edge/controller/env" "github.com/openziti/edge/pb/edge_ctrl_pb" "github.com/openziti/fabric/controller/network" - "github.com/openziti/foundation/channel" ) type resyncHandler struct { diff --git a/controller/handler_edge_ctrl/tunnel_health_event.go b/controller/handler_edge_ctrl/tunnel_health_event.go index 7679f4e12..97ad4ee63 100644 --- a/controller/handler_edge_ctrl/tunnel_health_event.go +++ b/controller/handler_edge_ctrl/tunnel_health_event.go @@ -17,10 +17,10 @@ package handler_edge_ctrl import ( + "github.com/openziti/channel" "github.com/openziti/edge/controller/env" "github.com/openziti/edge/edge_common" "github.com/openziti/edge/pb/edge_ctrl_pb" - "github.com/openziti/foundation/channel" "github.com/openziti/foundation/metrics" "time" ) @@ -31,7 +31,7 @@ type tunnelHealthEventHandler struct { serviceHealthCheckFailedCounter metrics.IntervalCounter } -func NewTunnelHealthEventHandler(appEnv *env.AppEnv, ch channel.Channel) channel.ReceiveHandler { +func NewTunnelHealthEventHandler(appEnv *env.AppEnv, ch channel.Channel) channel.TypedReceiveHandler { serviceEventMetrics := appEnv.GetHostController().GetNetwork().GetServiceEventsMetricsRegistry() return &tunnelHealthEventHandler{ baseRequestHandler: baseRequestHandler{ diff --git a/controller/handler_edge_ctrl/update_terminator.go b/controller/handler_edge_ctrl/update_terminator.go index d784a4183..37d4c38c5 100644 --- a/controller/handler_edge_ctrl/update_terminator.go +++ b/controller/handler_edge_ctrl/update_terminator.go @@ -19,11 +19,11 @@ package handler_edge_ctrl import ( "github.com/golang/protobuf/proto" "github.com/michaelquigley/pfxlog" + "github.com/openziti/channel" "github.com/openziti/edge/controller/env" "github.com/openziti/edge/controller/persistence" "github.com/openziti/edge/edge_common" "github.com/openziti/edge/pb/edge_ctrl_pb" - "github.com/openziti/foundation/channel" "github.com/sirupsen/logrus" ) @@ -31,7 +31,7 @@ type updateTerminatorHandler struct { baseRequestHandler } -func NewUpdateTerminatorHandler(appEnv *env.AppEnv, ch channel.Channel) channel.ReceiveHandler { +func NewUpdateTerminatorHandler(appEnv *env.AppEnv, ch channel.Channel) channel.TypedReceiveHandler { return &updateTerminatorHandler{ baseRequestHandler{ ch: ch, diff --git a/controller/handler_edge_ctrl/update_tunnel_terminator.go b/controller/handler_edge_ctrl/update_tunnel_terminator.go index 3e8e08b2a..aec21c39c 100644 --- a/controller/handler_edge_ctrl/update_tunnel_terminator.go +++ b/controller/handler_edge_ctrl/update_tunnel_terminator.go @@ -19,10 +19,10 @@ package handler_edge_ctrl import ( "github.com/golang/protobuf/proto" "github.com/michaelquigley/pfxlog" + "github.com/openziti/channel" "github.com/openziti/edge/controller/env" "github.com/openziti/edge/edge_common" "github.com/openziti/edge/pb/edge_ctrl_pb" - "github.com/openziti/foundation/channel" "github.com/sirupsen/logrus" ) @@ -30,7 +30,7 @@ type updateTunnelTerminatorHandler struct { baseRequestHandler } -func NewUpdateTunnelTerminatorHandler(appEnv *env.AppEnv, ch channel.Channel) channel.ReceiveHandler { +func NewUpdateTunnelTerminatorHandler(appEnv *env.AppEnv, ch channel.Channel) channel.TypedReceiveHandler { return &updateTunnelTerminatorHandler{ baseRequestHandler{ ch: ch, diff --git a/controller/handler_edge_ctrl/validate_sessions.go b/controller/handler_edge_ctrl/validate_sessions.go index eb6202cc6..2b958f687 100644 --- a/controller/handler_edge_ctrl/validate_sessions.go +++ b/controller/handler_edge_ctrl/validate_sessions.go @@ -19,9 +19,9 @@ package handler_edge_ctrl import ( "github.com/golang/protobuf/proto" "github.com/michaelquigley/pfxlog" + "github.com/openziti/channel" "github.com/openziti/edge/controller/env" "github.com/openziti/edge/pb/edge_ctrl_pb" - "github.com/openziti/foundation/channel" "go.etcd.io/bbolt" ) @@ -29,7 +29,7 @@ type validateSessionsHandler struct { baseRequestHandler } -func NewValidateSessionsHandler(appEnv *env.AppEnv, ch channel.Channel) channel.ReceiveHandler { +func NewValidateSessionsHandler(appEnv *env.AppEnv, ch channel.Channel) channel.TypedReceiveHandler { return &validateSessionsHandler{ baseRequestHandler{ ch: ch, diff --git a/controller/server/controller.go b/controller/server/controller.go index e165d18f7..f5ab41894 100644 --- a/controller/server/controller.go +++ b/controller/server/controller.go @@ -19,10 +19,10 @@ package server import "C" import ( "fmt" + "github.com/openziti/channel" sync2 "github.com/openziti/edge/controller/sync_strats" "github.com/openziti/edge/pb/edge_ctrl_pb" "github.com/openziti/fabric/controller/api_impl" - "github.com/openziti/foundation/channel" "io/ioutil" "sync" "time" @@ -36,7 +36,6 @@ import ( _ "github.com/openziti/edge/controller/internal/routes" "github.com/openziti/edge/controller/model" "github.com/openziti/edge/runner" - "github.com/openziti/foundation/channel2" "github.com/openziti/foundation/config" "github.com/openziti/foundation/storage/boltz" ) @@ -142,10 +141,11 @@ func (c *Controller) SetHostController(h env.HostController) { } } -func (c *Controller) GetCtrlHandlers(ch channel.Channel) []channel.ReceiveHandler { +func (c *Controller) GetCtrlHandlers(binding channel.Binding) []channel.TypedReceiveHandler { + ch := binding.GetChannel() tunnelState := handler_edge_ctrl.NewTunnelState() - return []channel.ReceiveHandler{ + result := []channel.TypedReceiveHandler{ handler_edge_ctrl.NewSessionHeartbeatHandler(c.AppEnv), handler_edge_ctrl.NewCreateCircuitHandler(c.AppEnv, ch), handler_edge_ctrl.NewCreateTerminatorHandler(c.AppEnv, ch), @@ -164,10 +164,14 @@ func (c *Controller) GetCtrlHandlers(ch channel.Channel) []channel.ReceiveHandle handler_edge_ctrl.NewExtendEnrollmentHandler(c.AppEnv), handler_edge_ctrl.NewExtendEnrollmentVerifyHandler(c.AppEnv), } + + result = append(result, c.AppEnv.Broker.GetReceiveHandlers()...) + + return result } -func (c *Controller) GetMgmtHandlers() []channel2.ReceiveHandler { - return []channel2.ReceiveHandler{} +func (c *Controller) GetMgmtHandlers() []channel.TypedReceiveHandler { + return []channel.TypedReceiveHandler{} } func (c *Controller) LoadConfig(cfgmap map[interface{}]interface{}) error { @@ -321,8 +325,7 @@ func (c *Controller) Shutdown() { } type subctrl struct { - parent *Controller - channel channel.Channel + parent *Controller } func (c *subctrl) GetTraceDecoders() []channel.TraceMessageDecoder { @@ -342,11 +345,10 @@ func (c *subctrl) Enabled() bool { return c.parent.Enabled() } -func (c *subctrl) BindChannel(ch channel.Channel) error { - for _, h := range c.parent.GetCtrlHandlers(ch) { - ch.AddReceiveHandler(h) +func (c *subctrl) BindChannel(binding channel.Binding) error { + for _, h := range c.parent.GetCtrlHandlers(binding) { + binding.AddTypedReceiveHandler(h) } - c.channel = ch return nil } @@ -355,8 +357,7 @@ func (c *subctrl) Run(channel.Channel, boltz.Db, chan struct{}) error { } type submgmt struct { - parent *Controller - channel channel2.Channel + parent *Controller } func (m *submgmt) LoadConfig(map[interface{}]interface{}) error { @@ -367,14 +368,13 @@ func (m *submgmt) Enabled() bool { return m.parent.Enabled() } -func (m *submgmt) BindChannel(ch channel2.Channel) error { +func (m *submgmt) BindChannel(binding channel.Binding) error { for _, h := range m.parent.GetMgmtHandlers() { - ch.AddReceiveHandler(h) + binding.AddTypedReceiveHandler(h) } - m.channel = ch return nil } -func (m *submgmt) Run(channel2.Channel, chan struct{}) error { +func (m *submgmt) Run(channel.Channel, chan struct{}) error { return nil } diff --git a/controller/sync_strats/rtx.go b/controller/sync_strats/rtx.go index fd4960367..59642b425 100644 --- a/controller/sync_strats/rtx.go +++ b/controller/sync_strats/rtx.go @@ -18,11 +18,11 @@ package sync_strats import ( "github.com/michaelquigley/pfxlog" + "github.com/openziti/channel" "github.com/openziti/edge/controller/env" "github.com/openziti/edge/controller/model" "github.com/openziti/edge/eid" "github.com/openziti/fabric/controller/network" - "github.com/openziti/foundation/channel" "github.com/openziti/foundation/channel2" "github.com/openziti/foundation/util/concurrenz" "github.com/pkg/errors" diff --git a/controller/sync_strats/sync_instant.go b/controller/sync_strats/sync_instant.go index 28d87de68..b0ed75a81 100644 --- a/controller/sync_strats/sync_instant.go +++ b/controller/sync_strats/sync_instant.go @@ -22,6 +22,7 @@ import ( "github.com/golang/protobuf/proto" "github.com/lucsky/cuid" "github.com/michaelquigley/pfxlog" + "github.com/openziti/channel" "github.com/openziti/edge/controller/env" "github.com/openziti/edge/controller/handler_edge_ctrl" "github.com/openziti/edge/controller/model" @@ -29,7 +30,6 @@ import ( "github.com/openziti/edge/pb/edge_ctrl_pb" "github.com/openziti/fabric/build" "github.com/openziti/fabric/controller/network" - "github.com/openziti/foundation/channel" "github.com/openziti/foundation/storage/ast" "github.com/openziti/foundation/util/debugz" "go.etcd.io/bbolt" @@ -80,8 +80,8 @@ type InstantStrategy struct { rtxMap *routerTxMap - helloHandler channel.ReceiveHandler - resyncHandler channel.ReceiveHandler + helloHandler channel.TypedReceiveHandler + resyncHandler channel.TypedReceiveHandler ae *env.AppEnv routerConnectedQueue chan *RouterSender @@ -182,6 +182,17 @@ func (strategy *InstantStrategy) RouterDisconnected(router *network.Router) { strategy.rtxMap.Remove(router.Id) } +func (strategy *InstantStrategy) GetReceiveHandlers() []channel.TypedReceiveHandler { + var result []channel.TypedReceiveHandler + if strategy.helloHandler != nil { + result = append(result, strategy.helloHandler) + } + if strategy.resyncHandler != nil { + result = append(result, strategy.resyncHandler) + } + return result +} + func (strategy *InstantStrategy) ApiSessionAdded(apiSession *persistence.ApiSession) { logger := pfxlog.Logger().WithField("strategy", strategy.Type()) @@ -294,7 +305,6 @@ func (strategy *InstantStrategy) startSynchronizeWorker() { } func (strategy *InstantStrategy) hello(rtx *RouterSender) { - logger := rtx.logger().WithField("strategy", strategy.Type()) logger.Info("edge router sync starting") @@ -305,7 +315,6 @@ func (strategy *InstantStrategy) hello(rtx *RouterSender) { return } - rtx.Router.Control.AddReceiveHandler(strategy.helloHandler) rtx.SetSyncStatus(env.RouterSyncHello) logger.WithField("syncStatus", rtx.SyncStatus()).Info("sending edge router hello") strategy.sendHello(rtx) @@ -324,7 +333,7 @@ func (strategy *InstantStrategy) sendHello(rtx *RouterSender) { return } - if err = rtx.Router.Control.Send(channel.NewMessage(env.ServerHelloType, buf).WithTimeout(strategy.HelloSendTimeout)); err != nil { + if err = channel.NewMessage(env.ServerHelloType, buf).WithTimeout(strategy.HelloSendTimeout).Send(rtx.Router.Control); err != nil { if rtx.Router.Control.IsClosed() { rtx.SetSyncStatus(env.RouterSyncDisconnected) rtx.logger().WithError(err).Error("timed out sending serverHello message for edge router, connection closed, giving up") diff --git a/go.mod b/go.mod index 3cc7f9d1f..b0947690a 100644 --- a/go.mod +++ b/go.mod @@ -42,7 +42,7 @@ require ( github.com/miekg/dns v1.1.45 github.com/mitchellh/mapstructure v1.4.3 github.com/netfoundry/secretstream v0.1.2 - github.com/openziti/fabric v0.17.26 + github.com/openziti/fabric v0.17.27 github.com/openziti/foundation v0.16.17 github.com/openziti/sdk-golang v0.15.135 github.com/orcaman/concurrent-map v0.0.0-20210106121528-16402b402231 @@ -56,10 +56,10 @@ require ( github.com/teris-io/shortid v0.0.0-20201117134242-e59966efd125 github.com/xeipuuv/gojsonschema v1.2.0 go.etcd.io/bbolt v1.3.5-0.20200615073812-232d8fc87f50 - golang.org/x/crypto v0.0.0-20210817164053-32db794688a5 + golang.org/x/crypto v0.0.0-20220126234351-aa10faf2a1f8 golang.org/x/net v0.0.0-20211209124913-491a49abca63 golang.org/x/sync v0.0.0-20210220032951-036812b2e83c - golang.org/x/sys v0.0.0-20211210111614-af8b64212486 + golang.org/x/sys v0.0.0-20220114195835-da31bd327af9 gopkg.in/resty.v1 v1.12.0 gopkg.in/yaml.v2 v2.4.0 gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b @@ -91,7 +91,6 @@ require ( github.com/josharian/intern v1.0.0 // indirect github.com/josharian/native v0.0.0-20200817173448-b6b71def0850 // indirect github.com/lucas-clemente/quic-go v0.23.0 // indirect - github.com/mailru/easyjson v0.7.7 // indirect github.com/marten-seemann/qtls-go1-16 v0.1.4 // indirect github.com/marten-seemann/qtls-go1-17 v0.1.0 // indirect github.com/mattn/go-colorable v0.1.12 // indirect @@ -105,7 +104,7 @@ require ( github.com/oklog/ulid v1.3.1 // indirect github.com/onsi/ginkgo v1.16.4 // indirect github.com/opentracing/opentracing-go v1.2.0 // indirect - github.com/parallaxsecond/parsec-client-go v0.0.0-20210416104105-e2d188152601 // indirect + github.com/parallaxsecond/parsec-client-go v0.0.0-20220111122524-cb78842db373 // indirect github.com/pmezard/go-difflib v1.0.0 // indirect github.com/shirou/gopsutil v2.21.11+incompatible // indirect github.com/speps/go-hashids v2.0.0+incompatible // indirect @@ -117,7 +116,7 @@ require ( go.mongodb.org/mongo-driver v1.7.5 // indirect go.mozilla.org/pkcs7 v0.0.0-20200128120323-432b2356ecb1 // indirect golang.org/x/mod v0.5.1 // indirect - golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1 // indirect + golang.org/x/term v0.0.0-20210927222741-03fcf44c2211 // indirect golang.org/x/text v0.3.7 // indirect golang.org/x/tools v0.1.7 // indirect golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1 // indirect @@ -125,3 +124,8 @@ require ( gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7 // indirect honnef.co/go/tools v0.2.2 // indirect ) + +require ( + github.com/mailru/easyjson v0.7.7 // indirect + github.com/openziti/channel v0.18.0 +) diff --git a/go.sum b/go.sum index 6071a235f..52383c12d 100644 --- a/go.sum +++ b/go.sum @@ -647,6 +647,8 @@ github.com/mgutz/ansi v0.0.0-20200706080929-d51e80ef957d/go.mod h1:01TrycV0kFyex github.com/michaelquigley/pfxlog v0.3.1/go.mod h1:pwJmJ9nN787xk9U+0z8/5nP2GMhU6GGCTfgBWGpiXaQ= github.com/michaelquigley/pfxlog v0.6.3 h1:/od363CtyX4z03vhzyJnfzOevXEYihGHVnLVI64a/e0= github.com/michaelquigley/pfxlog v0.6.3/go.mod h1:D2vg1tPyPdSXWWkSnGk6Fomwh5b3clwVJDUh71tq8Sk= +github.com/michaelquigley/pfxlog v0.6.9 h1:K/weH6ARu58aEDQi0ccinItvV958CeO6Ri4jqeFwd/w= +github.com/michaelquigley/pfxlog v0.6.9/go.mod h1:D2vg1tPyPdSXWWkSnGk6Fomwh5b3clwVJDUh71tq8Sk= github.com/microcosm-cc/bluemonday v1.0.1/go.mod h1:hsXNsILzKxV+sX77C5b8FSuKF00vh2OMYv+xgHpAMF4= github.com/miekg/dns v1.0.14/go.mod h1:W1PPwlIAgtquWBMBEV9nkV9Cazfe8ScdGz/Lj7v3Nrg= github.com/miekg/dns v1.1.26/go.mod h1:bPDLeHnStXmXAq1m/Ch/hvfNHr14JKNPMBo3VZKjuso= @@ -707,9 +709,12 @@ github.com/onsi/gomega v1.13.0/go.mod h1:lRk9szgn8TxENtWd0Tp4c3wjlRfMTMH27I+3Je4 github.com/opentracing/opentracing-go v1.2.0 h1:uEJPy/1a5RIPAJ0Ov+OIO8OxWu77jEv+1B0VhjKrZUs= github.com/opentracing/opentracing-go v1.2.0/go.mod h1:GxEUsuufX4nBwe+T+Wl9TAgYrxe9dPLANfrWvHYVTgc= github.com/openzipkin/zipkin-go v0.1.1/go.mod h1:NtoC/o8u3JlF1lSlyPNswIbeQH9bJTmOf0Erfk+hxe8= +github.com/openziti/channel v0.17.1-0.20220127192458-59ffac7bb5d0/go.mod h1:wmpGSNQranDE7jsiT+Cv31UzOO3z+m69JxE2tzYwkyY= +github.com/openziti/channel v0.18.0 h1:31Sh7/lBAx78jeYNyhm9zrxVjBxR0pun+XHJ4C0qy90= +github.com/openziti/channel v0.18.0/go.mod h1:wmpGSNQranDE7jsiT+Cv31UzOO3z+m69JxE2tzYwkyY= github.com/openziti/dilithium v0.3.3/go.mod h1:vsCjI2AU/hon9e+dLhUFbCNGesJDj2ASgkySOcpmvjo= -github.com/openziti/fabric v0.17.26 h1:JzZuqxm1Z7paxzh4/1vxH0433nHuaWJ2c70kn8Al7yc= -github.com/openziti/fabric v0.17.26/go.mod h1:Hudso2g47yUREm7V1i6CPVAfChSrKR8I3wyd94dhZE0= +github.com/openziti/fabric v0.17.27 h1:7RRvxUVYeLDVMrAXtiDesDws6WUPmY9x4r6j8BXa0+Q= +github.com/openziti/fabric v0.17.27/go.mod h1:XjZ6x5eEoW2OE3allkoWJ/IxkiVtSBhxibXHCnrmLB4= github.com/openziti/foundation v0.16.17 h1:EusJpSV+STGJSbfnQAcZVz8jbVZomSmc00UhTzx3dTc= github.com/openziti/foundation v0.16.17/go.mod h1:D8IchItaO2SvK/fc/JpX9yF0vNymmOw2gzeNqwHz4+g= github.com/openziti/sdk-golang v0.15.135 h1:4sLhEvJ9YY9v/vRq5eqXjqAuVcvcKe8+nOqIA0zoQqI= @@ -722,8 +727,9 @@ github.com/otiai10/curr v0.0.0-20150429015615-9b4961190c95/go.mod h1:9qAhocn7zKJ github.com/otiai10/curr v1.0.0/go.mod h1:LskTG5wDwr8Rs+nNQ+1LlxRjAtTZZjtJW4rMXl6j4vs= github.com/otiai10/mint v1.3.0/go.mod h1:F5AjcsTsWUqX+Na9fpHb52P8pcRX2CI6A3ctIT91xUo= github.com/otiai10/mint v1.3.1/go.mod h1:/yxELlJQ0ufhjUwhshSj+wFjZ78CnZ48/1wtmBH1OTc= -github.com/parallaxsecond/parsec-client-go v0.0.0-20210416104105-e2d188152601 h1:1YdmkuR6lJ+mDvdkI8cBDKOljRWL4kdPFSu3qRAEf6g= github.com/parallaxsecond/parsec-client-go v0.0.0-20210416104105-e2d188152601/go.mod h1:iFrEMRwLidGESW7mwIhRWDD7m5ktzKeCX2XNCyAFtNk= +github.com/parallaxsecond/parsec-client-go v0.0.0-20220111122524-cb78842db373 h1:CUvH4JL/8OVy023LMER3dB/MerNQ6OIz4QV3E/JQ3UY= +github.com/parallaxsecond/parsec-client-go v0.0.0-20220111122524-cb78842db373/go.mod h1:gLH27qo/dvMhLTVVyMELpe3Tut7sOfkiDg7ZpeqKwsw= github.com/pascaldekloe/goe v0.0.0-20180627143212-57f6aae5913c/go.mod h1:lzWF7FIEvWOWxwDKqyGYQf6ZUaNfKdP144TG7ZOy1lc= github.com/pascaldekloe/goe v0.1.0/go.mod h1:lzWF7FIEvWOWxwDKqyGYQf6ZUaNfKdP144TG7ZOy1lc= github.com/pborman/uuid v1.2.0/go.mod h1:X/NO0urCmaxf9VXbdlT7C2Yzkj2IKimNn4k+gtPdI/k= @@ -963,8 +969,9 @@ golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPh golang.org/x/crypto v0.0.0-20200820211705-5c72a883971a/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto= golang.org/x/crypto v0.0.0-20201221181555-eec23a3978ad/go.mod h1:jdWPYTVW3xRLrWPugEBEK3UY2ZEsg3UU495nc5E+M+I= golang.org/x/crypto v0.0.0-20210616213533-5ff15b29337e/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc= -golang.org/x/crypto v0.0.0-20210817164053-32db794688a5 h1:HWj/xjIHfjYU5nVXpTM0s39J9CbLn7Cc5a7IC5rwsMQ= golang.org/x/crypto v0.0.0-20210817164053-32db794688a5/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc= +golang.org/x/crypto v0.0.0-20220126234351-aa10faf2a1f8 h1:kACShD3qhmr/3rLmg1yXyt+N4HcwutKyPRB93s54TIU= +golang.org/x/crypto v0.0.0-20220126234351-aa10faf2a1f8/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4= golang.org/x/exp v0.0.0-20190121172915-509febef88a4/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA= golang.org/x/exp v0.0.0-20190306152737-a1d7652674e8/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA= golang.org/x/exp v0.0.0-20190510132918-efd6b22b2522/go.mod h1:ZjyILWgesfNpC6sMxTJOJm9Kp84zZh5NQWvqDGG3Qr8= @@ -1071,6 +1078,7 @@ golang.org/x/net v0.0.0-20210813160813-60bc85c4be6d/go.mod h1:9nx3DQGgdP8bBQD5qx golang.org/x/net v0.0.0-20210928044308-7d9f5e0b762b/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y= golang.org/x/net v0.0.0-20211020060615-d418f374d309/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y= golang.org/x/net v0.0.0-20211029224645-99673261e6eb/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y= +golang.org/x/net v0.0.0-20211112202133-69e39bad7dc2/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y= golang.org/x/net v0.0.0-20211209124913-491a49abca63 h1:iocB37TsdFuN6IBRZ+ry36wrkoV51/tl5vOWqkcPGvY= golang.org/x/net v0.0.0-20211209124913-491a49abca63/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y= golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U= @@ -1209,11 +1217,13 @@ golang.org/x/sys v0.0.0-20211007075335-d3039528d8ac/go.mod h1:oPkhp1MJrh7nUepCBc golang.org/x/sys v0.0.0-20211025201205-69cdffdb9359/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.0.0-20211124211545-fe61309f8881/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.0.0-20211205182925-97ca703d548d/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= -golang.org/x/sys v0.0.0-20211210111614-af8b64212486 h1:5hpz5aRr+W1erYCL5JRhSUBJRph7l9XkNveoExlrKYk= golang.org/x/sys v0.0.0-20211210111614-af8b64212486/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.0.0-20220114195835-da31bd327af9 h1:XfKQ4OlFl8okEOr5UvAqFRVj8pY/4yfcXrddB8qAbU0= +golang.org/x/sys v0.0.0-20220114195835-da31bd327af9/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/term v0.0.0-20201117132131-f5c789dd3221/go.mod h1:Nr5EML6q2oocZ2LXRh80K7BxOlk5/8JxuGnuhpl+muw= -golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1 h1:v+OssWQX+hTHEmOBgwxdZxK4zHq3yOs8F9J7mk0PY8E= golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo= +golang.org/x/term v0.0.0-20210927222741-03fcf44c2211 h1:JGgROgKl9N8DuW20oFS5gxc+lE67/N3FcwmBPMe7ArY= +golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8= golang.org/x/text v0.0.0-20170915032832-14c0d48ead0c/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= golang.org/x/text v0.3.1-0.20180807135948-17ff2d5776d2/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= diff --git a/pb/edge_ctrl_pb/decoder.go b/pb/edge_ctrl_pb/decoder.go index b81dc2b48..a1e223a32 100644 --- a/pb/edge_ctrl_pb/decoder.go +++ b/pb/edge_ctrl_pb/decoder.go @@ -20,7 +20,7 @@ import ( "fmt" "github.com/golang/protobuf/proto" "github.com/michaelquigley/pfxlog" - "github.com/openziti/foundation/channel" + "github.com/openziti/channel" "github.com/openziti/foundation/channel2" "strings" ) diff --git a/router/xgress_edge/certchecker_test.go b/router/xgress_edge/certchecker_test.go index d33185dae..e64cb0784 100644 --- a/router/xgress_edge/certchecker_test.go +++ b/router/xgress_edge/certchecker_test.go @@ -259,8 +259,8 @@ func Test_CertExpirationChecker(t *testing.T) { extender := &stubExtender{ done: func() error { invoked = true - certChecker.id.Cert().Leaf.NotAfter = time.Now().AddDate(1,0,0) - certChecker.id.ServerCert().Leaf.NotAfter = time.Now().AddDate(1,0,0) + certChecker.id.Cert().Leaf.NotAfter = time.Now().AddDate(1, 0, 0) + certChecker.id.ServerCert().Leaf.NotAfter = time.Now().AddDate(1, 0, 0) return errors.New("test") }, } @@ -576,7 +576,7 @@ func (ch *simpleTestChannel) AddTransformHandler(channel2.TransformHandler) { } func (ch *simpleTestChannel) AddReceiveHandler(channel2.ReceiveHandler) { - panic("implement AddReceiveHandler") + panic("implement AddTypedReceiveHandler") } func (ch *simpleTestChannel) AddErrorHandler(channel2.ErrorHandler) { @@ -663,7 +663,7 @@ func (ch *simpleTestChannel) GetTimeSinceLastRead() time.Duration { type stubExtender struct { isRequesting concurrenz.AtomicBoolean - done func() error + done func() error } func (s stubExtender) IsRequestingCompareAndSwap(expected bool, value bool) bool { @@ -686,4 +686,4 @@ func (s stubExtender) ExtendEnrollment() error { func (s stubExtender) IsRequesting() bool { return s.isRequesting.Get() -} \ No newline at end of file +}