diff --git a/controller/handler_edge_ctrl/common.go b/controller/handler_edge_ctrl/common.go index bf57d506c..acdda1dd7 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/channel2" + "github.com/openziti/foundation/channel" "github.com/openziti/foundation/identity/identity" "github.com/openziti/foundation/storage/boltz" "github.com/openziti/foundation/util/stringz" @@ -28,7 +28,7 @@ import ( type requestHandler interface { getAppEnv() *env.AppEnv getNetwork() *network.Network - getChannel() channel2.Channel + getChannel() channel.Channel ContentType() int32 Label() string } @@ -41,7 +41,7 @@ type UpdateTerminatorRequest interface { } type baseRequestHandler struct { - ch channel2.Channel + ch channel.Channel appEnv *env.AppEnv } @@ -53,13 +53,13 @@ func (self *baseRequestHandler) getAppEnv() *env.AppEnv { return self.appEnv } -func (self *baseRequestHandler) getChannel() channel2.Channel { +func (self *baseRequestHandler) getChannel() channel.Channel { return self.ch } func (self *baseRequestHandler) returnError(ctx requestContext, err controllerError) { ctx.CleanupOnError() - responseMsg := channel2.NewMessage(int32(edge_ctrl_pb.ContentType_ErrorType), []byte(err.Error())) + responseMsg := channel.NewMessage(int32(edge_ctrl_pb.ContentType_ErrorType), []byte(err.Error())) responseMsg.PutUint32Header(edge.ErrorCodeHeader, err.ErrorCode()) responseMsg.ReplyTo(ctx.GetMessage()) logger := pfxlog. @@ -97,7 +97,7 @@ func (self *baseRequestHandler) logResult(ctx requestContext, err error) { type requestContext interface { GetHandler() requestHandler - GetMessage() *channel2.Message + GetMessage() *channel.Message CleanupOnError() } @@ -108,7 +108,7 @@ type sessionRequestContext interface { type baseSessionRequestContext struct { handler requestHandler - msg *channel2.Message + msg *channel.Message err controllerError sourceRouter *network.Router session *model.Session @@ -129,7 +129,7 @@ func (self *baseSessionRequestContext) CleanupOnError() { } } -func (self *baseSessionRequestContext) GetMessage() *channel2.Message { +func (self *baseSessionRequestContext) GetMessage() *channel.Message { return self.msg } diff --git a/controller/handler_edge_ctrl/create_circuit.go b/controller/handler_edge_ctrl/create_circuit.go index ca4f85178..2a2bf5495 100644 --- a/controller/handler_edge_ctrl/create_circuit.go +++ b/controller/handler_edge_ctrl/create_circuit.go @@ -22,14 +22,14 @@ import ( "github.com/openziti/edge/controller/env" "github.com/openziti/edge/controller/persistence" "github.com/openziti/edge/pb/edge_ctrl_pb" - "github.com/openziti/foundation/channel2" + "github.com/openziti/foundation/channel" ) type createCircuitHandler struct { baseRequestHandler } -func NewCreateCircuitHandler(appEnv *env.AppEnv, ch channel2.Channel) channel2.ReceiveHandler { +func NewCreateCircuitHandler(appEnv *env.AppEnv, ch channel.Channel) channel.ReceiveHandler { return &createCircuitHandler{ baseRequestHandler: baseRequestHandler{ ch: ch, @@ -55,14 +55,14 @@ func (self *createCircuitHandler) sendResponse(ctx *CreateCircuitRequestContext, return } - responseMsg := channel2.NewMessage(response.GetContentType(), body) + responseMsg := channel.NewMessage(response.GetContentType(), body) responseMsg.ReplyTo(ctx.msg) if err = self.ch.Send(responseMsg); err != nil { log.WithError(err).WithField("token", ctx.req.SessionToken).Error("failed to send create circuit response") } } -func (self *createCircuitHandler) HandleReceive(msg *channel2.Message, ch channel2.Channel) { +func (self *createCircuitHandler) HandleReceive(msg *channel.Message, ch channel.Channel) { req := &edge_ctrl_pb.CreateCircuitRequest{} if err := proto.Unmarshal(msg.Body, req); err != nil { pfxlog.ContextLogger(ch.Label()).WithError(err).Error("could not unmarshal CreateCircuitRequest") diff --git a/controller/handler_edge_ctrl/create_terminator.go b/controller/handler_edge_ctrl/create_terminator.go index ec02bde8d..f436785c1 100644 --- a/controller/handler_edge_ctrl/create_terminator.go +++ b/controller/handler_edge_ctrl/create_terminator.go @@ -25,7 +25,7 @@ import ( "github.com/openziti/edge/edge_common" "github.com/openziti/edge/pb/edge_ctrl_pb" "github.com/openziti/fabric/controller/network" - "github.com/openziti/foundation/channel2" + "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 channel2.Channel) channel2.ReceiveHandler { +func NewCreateTerminatorHandler(appEnv *env.AppEnv, ch channel.Channel) channel.ReceiveHandler { return &createTerminatorHandler{ baseRequestHandler{ ch: ch, @@ -52,7 +52,7 @@ func (self *createTerminatorHandler) Label() string { return "create.terminator" } -func (self *createTerminatorHandler) HandleReceive(msg *channel2.Message, ch channel2.Channel) { +func (self *createTerminatorHandler) HandleReceive(msg *channel.Message, ch channel.Channel) { req := &edge_ctrl_pb.CreateTerminatorRequest{} if err := proto.Unmarshal(msg.Body, req); err != nil { pfxlog.ContextLogger(ch.Label()).WithError(err).Error("could not unmarshal CreateTerminatorRequest") @@ -123,7 +123,7 @@ func (self *createTerminatorHandler) CreateTerminator(ctx *CreateTerminatorReque log.WithField("terminator", id).Info("created terminator") - responseMsg := channel2.NewMessage(int32(edge_ctrl_pb.ContentType_CreateTerminatorResponseType), []byte(id)) + responseMsg := channel.NewMessage(int32(edge_ctrl_pb.ContentType_CreateTerminatorResponseType), []byte(id)) responseMsg.ReplyTo(ctx.msg) if err := self.ch.Send(responseMsg); err != nil { log.WithError(err).Error("failed to send create terminator response") diff --git a/controller/handler_edge_ctrl/create_tunnel_api_session.go b/controller/handler_edge_ctrl/create_tunnel_api_session.go index 4ae508063..ae8ec25f4 100644 --- a/controller/handler_edge_ctrl/create_tunnel_api_session.go +++ b/controller/handler_edge_ctrl/create_tunnel_api_session.go @@ -4,7 +4,7 @@ import ( "github.com/golang/protobuf/proto" "github.com/openziti/edge/controller/env" "github.com/openziti/edge/pb/edge_ctrl_pb" - "github.com/openziti/foundation/channel2" + "github.com/openziti/foundation/channel" "github.com/sirupsen/logrus" ) @@ -13,7 +13,7 @@ type createApiSessionHandler struct { *TunnelState } -func NewCreateApiSessionHandler(appEnv *env.AppEnv, ch channel2.Channel, tunnelState *TunnelState) channel2.ReceiveHandler { +func NewCreateApiSessionHandler(appEnv *env.AppEnv, ch channel.Channel, tunnelState *TunnelState) channel.ReceiveHandler { return &createApiSessionHandler{ baseRequestHandler: baseRequestHandler{ch: ch, appEnv: appEnv}, TunnelState: tunnelState, @@ -32,7 +32,7 @@ func (self *createApiSessionHandler) Label() string { return "tunnel.create.api_session" } -func (self *createApiSessionHandler) HandleReceive(msg *channel2.Message, _ channel2.Channel) { +func (self *createApiSessionHandler) HandleReceive(msg *channel.Message, _ channel.Channel) { req := &edge_ctrl_pb.CreateApiSessionRequest{} if err := proto.Unmarshal(msg.Body, req); err != nil { logrus.WithField("router", self.ch.Id().Token).WithError(err).Error("could not unmarshal CreateApiSessionRequest") @@ -77,7 +77,7 @@ func (self *createApiSessionHandler) createApiSession(ctx *createApiSessionReque return } - responseMsg := channel2.NewMessage(int32(edge_ctrl_pb.ContentType_CreateApiSessionResponseType), body) + responseMsg := channel.NewMessage(int32(edge_ctrl_pb.ContentType_CreateApiSessionResponseType), body) responseMsg.ReplyTo(ctx.msg) if err = self.ch.Send(responseMsg); err != nil { logrus.WithError(err).Error("failed to send response") diff --git a/controller/handler_edge_ctrl/create_tunnel_circuit.go b/controller/handler_edge_ctrl/create_tunnel_circuit.go index b6ee943c9..dc14f524d 100644 --- a/controller/handler_edge_ctrl/create_tunnel_circuit.go +++ b/controller/handler_edge_ctrl/create_tunnel_circuit.go @@ -22,7 +22,7 @@ import ( "github.com/openziti/edge/controller/env" "github.com/openziti/edge/controller/persistence" "github.com/openziti/edge/pb/edge_ctrl_pb" - "github.com/openziti/foundation/channel2" + "github.com/openziti/foundation/channel" ) type createCircuitForServiceHandler struct { @@ -30,7 +30,7 @@ type createCircuitForServiceHandler struct { *TunnelState } -func NewCreateCircuitForTunnelHandler(appEnv *env.AppEnv, ch channel2.Channel, tunnelState *TunnelState) channel2.ReceiveHandler { +func NewCreateCircuitForTunnelHandler(appEnv *env.AppEnv, ch channel.Channel, tunnelState *TunnelState) channel.ReceiveHandler { return &createCircuitForServiceHandler{ baseRequestHandler: baseRequestHandler{ch: ch, appEnv: appEnv}, TunnelState: tunnelState, @@ -58,14 +58,14 @@ func (self *createCircuitForServiceHandler) sendResponse(ctx *CreateCircuitForSe return } - responseMsg := channel2.NewMessage(response.GetContentType(), body) + responseMsg := channel.NewMessage(response.GetContentType(), body) responseMsg.ReplyTo(ctx.msg) if err = self.ch.Send(responseMsg); err != nil { log.WithError(err).WithField("service", ctx.req.ServiceName).Error("failed to send CreateCircuitForServiceResponse") } } -func (self *createCircuitForServiceHandler) HandleReceive(msg *channel2.Message, ch channel2.Channel) { +func (self *createCircuitForServiceHandler) HandleReceive(msg *channel.Message, ch channel.Channel) { req := &edge_ctrl_pb.CreateCircuitForServiceRequest{} if err := proto.Unmarshal(msg.Body, req); err != nil { pfxlog.ContextLogger(ch.Label()).WithError(err).Error("could not unmarshal CreateCircuitForServiceRequest") diff --git a/controller/handler_edge_ctrl/create_tunnel_terminator.go b/controller/handler_edge_ctrl/create_tunnel_terminator.go index a3b95e645..bf3adf397 100644 --- a/controller/handler_edge_ctrl/create_tunnel_terminator.go +++ b/controller/handler_edge_ctrl/create_tunnel_terminator.go @@ -25,7 +25,7 @@ import ( "github.com/openziti/edge/edge_common" "github.com/openziti/edge/pb/edge_ctrl_pb" "github.com/openziti/fabric/controller/network" - "github.com/openziti/foundation/channel2" + "github.com/openziti/foundation/channel" "github.com/sirupsen/logrus" "math" ) @@ -35,7 +35,7 @@ type createTunnelTerminatorHandler struct { *TunnelState } -func NewCreateTunnelTerminatorHandler(appEnv *env.AppEnv, ch channel2.Channel, tunnelState *TunnelState) channel2.ReceiveHandler { +func NewCreateTunnelTerminatorHandler(appEnv *env.AppEnv, ch channel.Channel, tunnelState *TunnelState) channel.ReceiveHandler { return &createTunnelTerminatorHandler{ baseRequestHandler: baseRequestHandler{ch: ch, appEnv: appEnv}, TunnelState: tunnelState, @@ -54,7 +54,7 @@ func (self *createTunnelTerminatorHandler) Label() string { return "tunnel.create.terminator" } -func (self *createTunnelTerminatorHandler) HandleReceive(msg *channel2.Message, ch channel2.Channel) { +func (self *createTunnelTerminatorHandler) HandleReceive(msg *channel.Message, ch channel.Channel) { req := &edge_ctrl_pb.CreateTunnelTerminatorRequest{} if err := proto.Unmarshal(msg.Body, req); err != nil { pfxlog.ContextLogger(ch.Label()).WithError(err).Error("could not unmarshal CreateTerminatorRequest") @@ -139,7 +139,7 @@ func (self *createTunnelTerminatorHandler) CreateTerminator(ctx *CreateTunnelTer return } - responseMsg := channel2.NewMessage(response.GetContentType(), body) + responseMsg := channel.NewMessage(response.GetContentType(), body) responseMsg.ReplyTo(ctx.msg) if err = self.ch.Send(responseMsg); err != nil { logger.WithError(err).Error("failed to send CreateTunnelTerminatorResponse") diff --git a/controller/handler_edge_ctrl/extend_enrollment.go b/controller/handler_edge_ctrl/extend_enrollment.go index 49c69a893..1cf955b04 100644 --- a/controller/handler_edge_ctrl/extend_enrollment.go +++ b/controller/handler_edge_ctrl/extend_enrollment.go @@ -23,7 +23,7 @@ import ( "github.com/openziti/edge/controller/model" "github.com/openziti/edge/internal/cert" "github.com/openziti/edge/pb/edge_ctrl_pb" - "github.com/openziti/foundation/channel2" + "github.com/openziti/foundation/channel" ) type extendEnrollmentHandler struct { @@ -40,7 +40,7 @@ func (h *extendEnrollmentHandler) ContentType() int32 { return env.EnrollmentExtendRouterRequestType } -func (h *extendEnrollmentHandler) HandleReceive(msg *channel2.Message, ch channel2.Channel) { +func (h *extendEnrollmentHandler) HandleReceive(msg *channel.Message, ch channel.Channel) { go func() { req := &edge_ctrl_pb.EnrollmentExtendRouterRequest{} certs := ch.Underlay().Certificates() @@ -126,7 +126,7 @@ func (h *extendEnrollmentHandler) HandleReceive(msg *channel2.Message, ch channe return } - msg := channel2.NewMessage(env.EnrollmentCertsResponseType, body) + msg := channel.NewMessage(env.EnrollmentCertsResponseType, body) if err := ch.Send(msg); err != nil { log.WithError(err).Errorf("request to extend router enrollment failed to send enrollment certificate response") diff --git a/controller/handler_edge_ctrl/extend_enrollment_verify.go b/controller/handler_edge_ctrl/extend_enrollment_verify.go index 31a32b164..5a27ccd30 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/channel2" + "github.com/openziti/foundation/channel" ) type extendEnrollmentVerifyHandler struct { @@ -39,16 +39,16 @@ func (h *extendEnrollmentVerifyHandler) ContentType() int32 { return env.EnrollmentExtendRouterVerifyRequestType } -func (h *extendEnrollmentVerifyHandler) respond(respErr *edge_ctrl_pb.Error, msg *channel2.Message, ch channel2.Channel) { +func (h *extendEnrollmentVerifyHandler) respond(respErr *edge_ctrl_pb.Error, msg *channel.Message, ch channel.Channel) { respErrBody, _ := proto.Marshal(respErr) - respMsg := channel2.NewMessage(int32(edge_ctrl_pb.ContentType_ErrorType), respErrBody) + respMsg := channel.NewMessage(int32(edge_ctrl_pb.ContentType_ErrorType), respErrBody) respMsg.ReplyTo(msg) if err := ch.Send(respMsg); err != nil { pfxlog.Logger().Errorf("could not send enrollment verification response, channel error: %v", err) } } -func (h *extendEnrollmentVerifyHandler) HandleReceive(msg *channel2.Message, ch channel2.Channel) { +func (h *extendEnrollmentVerifyHandler) HandleReceive(msg *channel.Message, ch channel.Channel) { go func() { verifyMsg := &edge_ctrl_pb.EnrollmentExtendRouterVerifyRequest{} diff --git a/controller/handler_edge_ctrl/health_event.go b/controller/handler_edge_ctrl/health_event.go index 433f65088..4eb2fb6b0 100644 --- a/controller/handler_edge_ctrl/health_event.go +++ b/controller/handler_edge_ctrl/health_event.go @@ -22,7 +22,7 @@ import ( "github.com/openziti/edge/controller/env" "github.com/openziti/edge/controller/persistence" "github.com/openziti/edge/pb/edge_ctrl_pb" - "github.com/openziti/foundation/channel2" + "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 channel2.Channel) channel2.ReceiveHandler { +func NewHealthEventHandler(appEnv *env.AppEnv, ch channel.Channel) channel.ReceiveHandler { serviceEventMetrics := appEnv.GetHostController().GetNetwork().GetServiceEventsMetricsRegistry() return &healthEventHandler{ baseRequestHandler: baseRequestHandler{ @@ -53,7 +53,7 @@ func (self *healthEventHandler) Label() string { return "health.event" } -func (self *healthEventHandler) HandleReceive(msg *channel2.Message, ch channel2.Channel) { +func (self *healthEventHandler) HandleReceive(msg *channel.Message, ch channel.Channel) { req := &edge_ctrl_pb.HealthEventRequest{} if err := proto.Unmarshal(msg.Body, req); err != nil { pfxlog.ContextLogger(ch.Label()).WithError(err).Error("could not unmarshal health event") diff --git a/controller/handler_edge_ctrl/heartbeat.go b/controller/handler_edge_ctrl/heartbeat.go index 39ef73eec..28425e15c 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/channel2" + "github.com/openziti/foundation/channel" ) type sessionHeartbeatHandler struct { @@ -38,7 +38,7 @@ func (h *sessionHeartbeatHandler) ContentType() int32 { return env.ApiSessionHeartbeatType } -func (h *sessionHeartbeatHandler) HandleReceive(msg *channel2.Message, ch channel2.Channel) { +func (h *sessionHeartbeatHandler) HandleReceive(msg *channel.Message, ch channel.Channel) { go func() { req := &edge_ctrl_pb.ApiSessionHeartbeat{} routerId := ch.Id().Token @@ -63,7 +63,7 @@ func (h *sessionHeartbeatHandler) HandleReceive(msg *channel2.Message, ch channe } msgBytes, _ := proto.Marshal(msgStruct) - msg := channel2.NewMessage(env.ApiSessionRemovedType, msgBytes) + msg := channel.NewMessage(env.ApiSessionRemovedType, msgBytes) _ = ch.Send(msg) } diff --git a/controller/handler_edge_ctrl/hello.go b/controller/handler_edge_ctrl/hello.go index 79d4580e1..3b6a3b635 100644 --- a/controller/handler_edge_ctrl/hello.go +++ b/controller/handler_edge_ctrl/hello.go @@ -22,7 +22,7 @@ import ( "github.com/openziti/edge/controller/env" "github.com/openziti/edge/pb/edge_ctrl_pb" "github.com/openziti/fabric/controller/network" - "github.com/openziti/foundation/channel2" + "github.com/openziti/foundation/channel" ) type helloHandler struct { @@ -41,7 +41,7 @@ func (h *helloHandler) ContentType() int32 { return env.ClientHelloType } -func (h *helloHandler) HandleReceive(msg *channel2.Message, ch channel2.Channel) { +func (h *helloHandler) HandleReceive(msg *channel.Message, ch channel.Channel) { respHello := &edge_ctrl_pb.ClientHello{} if err := proto.Unmarshal(msg.Body, respHello); err != nil { pfxlog.Logger().WithError(err).Error("could not unmarshal clientHello after serverHello") diff --git a/controller/handler_edge_ctrl/list_tunnel_services.go b/controller/handler_edge_ctrl/list_tunnel_services.go index bb298c8c5..48c2f4f1c 100644 --- a/controller/handler_edge_ctrl/list_tunnel_services.go +++ b/controller/handler_edge_ctrl/list_tunnel_services.go @@ -5,7 +5,7 @@ import ( "github.com/golang/protobuf/proto" "github.com/openziti/edge/controller/env" "github.com/openziti/edge/pb/edge_ctrl_pb" - "github.com/openziti/foundation/channel2" + "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 channel2.Channel, tunnelState *TunnelState) channel2.ReceiveHandler { +func NewListTunnelServicesHandler(appEnv *env.AppEnv, ch channel.Channel, tunnelState *TunnelState) channel.ReceiveHandler { return &listTunnelServicesHandler{ baseRequestHandler: baseRequestHandler{ch: ch, appEnv: appEnv}, TunnelState: tunnelState, @@ -35,7 +35,7 @@ func (self *listTunnelServicesHandler) Label() string { return "tunnel.list.services" } -func (self *listTunnelServicesHandler) HandleReceive(msg *channel2.Message, _ channel2.Channel) { +func (self *listTunnelServicesHandler) HandleReceive(msg *channel.Message, _ channel.Channel) { logger := logrus.WithField("router", self.ch.Id().Token) ctx := &listTunnelServicesRequestContext{ @@ -134,7 +134,7 @@ func (self *listTunnelServicesHandler) listServices(ctx *listTunnelServicesReque return } - serviceListMsg := channel2.NewMessage(serviceList.GetContentType(), body) + serviceListMsg := channel.NewMessage(serviceList.GetContentType(), body) if err = self.ch.Send(serviceListMsg); err != nil { logger.Error("failed to send services list") } diff --git a/controller/handler_edge_ctrl/remove_terminator.go b/controller/handler_edge_ctrl/remove_terminator.go index d0f087c30..3f14fd15f 100644 --- a/controller/handler_edge_ctrl/remove_terminator.go +++ b/controller/handler_edge_ctrl/remove_terminator.go @@ -23,7 +23,7 @@ import ( "github.com/openziti/edge/controller/persistence" "github.com/openziti/edge/edge_common" "github.com/openziti/edge/pb/edge_ctrl_pb" - "github.com/openziti/foundation/channel2" + "github.com/openziti/foundation/channel" "github.com/sirupsen/logrus" ) @@ -31,7 +31,7 @@ type removeTerminatorHandler struct { baseRequestHandler } -func NewRemoveTerminatorHandler(appEnv *env.AppEnv, ch channel2.Channel) channel2.ReceiveHandler { +func NewRemoveTerminatorHandler(appEnv *env.AppEnv, ch channel.Channel) channel.ReceiveHandler { return &removeTerminatorHandler{ baseRequestHandler{ ch: ch, @@ -51,14 +51,14 @@ func (self *removeTerminatorHandler) Label() string { func (self *removeTerminatorHandler) sendResponse(ctx *RemoveTerminatorRequestContext) { log := pfxlog.ContextLogger(self.ch.Label()) - responseMsg := channel2.NewMessage(int32(edge_ctrl_pb.ContentType_RemoveTerminatorResponseType), nil) + responseMsg := channel.NewMessage(int32(edge_ctrl_pb.ContentType_RemoveTerminatorResponseType), nil) responseMsg.ReplyTo(ctx.msg) if err := self.ch.Send(responseMsg); err != nil { log.WithError(err).WithField("token", ctx.req.SessionToken).Error("failed to send remove terminator response") } } -func (self *removeTerminatorHandler) HandleReceive(msg *channel2.Message, ch channel2.Channel) { +func (self *removeTerminatorHandler) HandleReceive(msg *channel.Message, ch channel.Channel) { req := &edge_ctrl_pb.RemoveTerminatorRequest{} if err := proto.Unmarshal(msg.Body, req); err != nil { pfxlog.ContextLogger(ch.Label()).WithError(err).Error("could not unmarshal RemoveTerminator") diff --git a/controller/handler_edge_ctrl/remove_tunnel_terminator.go b/controller/handler_edge_ctrl/remove_tunnel_terminator.go index 0c72bb414..531d50052 100644 --- a/controller/handler_edge_ctrl/remove_tunnel_terminator.go +++ b/controller/handler_edge_ctrl/remove_tunnel_terminator.go @@ -20,7 +20,7 @@ import ( "github.com/openziti/edge/controller/env" "github.com/openziti/edge/edge_common" "github.com/openziti/edge/pb/edge_ctrl_pb" - "github.com/openziti/foundation/channel2" + "github.com/openziti/foundation/channel" "github.com/sirupsen/logrus" ) @@ -28,7 +28,7 @@ type removeTunnelTerminatorHandler struct { baseRequestHandler } -func NewRemoveTunnelTerminatorHandler(appEnv *env.AppEnv, ch channel2.Channel) channel2.ReceiveHandler { +func NewRemoveTunnelTerminatorHandler(appEnv *env.AppEnv, ch channel.Channel) channel.ReceiveHandler { return &removeTunnelTerminatorHandler{ baseRequestHandler{ ch: ch, @@ -45,7 +45,7 @@ func (self *removeTunnelTerminatorHandler) Label() string { return "tunnel.remove.terminator" } -func (self *removeTunnelTerminatorHandler) HandleReceive(msg *channel2.Message, _ channel2.Channel) { +func (self *removeTunnelTerminatorHandler) HandleReceive(msg *channel.Message, _ channel.Channel) { ctx := &RemoveTunnelTerminatorRequestContext{ baseSessionRequestContext: baseSessionRequestContext{handler: self, msg: msg}, terminatorId: string(msg.Body), @@ -77,7 +77,7 @@ func (self *removeTunnelTerminatorHandler) RemoveTerminator(ctx *RemoveTunnelTer logger.Info("removed terminator") - responseMsg := channel2.NewMessage(int32(edge_ctrl_pb.ContentType_RemoveTunnelTerminatorResponseType), nil) + responseMsg := channel.NewMessage(int32(edge_ctrl_pb.ContentType_RemoveTunnelTerminatorResponseType), nil) responseMsg.ReplyTo(ctx.msg) if err := self.ch.Send(responseMsg); err != nil { logger.WithError(err).Error("failed to send remove tunnel terminator response") diff --git a/controller/handler_edge_ctrl/resync.go b/controller/handler_edge_ctrl/resync.go index 8c200f5da..53af17d93 100644 --- a/controller/handler_edge_ctrl/resync.go +++ b/controller/handler_edge_ctrl/resync.go @@ -22,7 +22,7 @@ import ( "github.com/openziti/edge/controller/env" "github.com/openziti/edge/pb/edge_ctrl_pb" "github.com/openziti/fabric/controller/network" - "github.com/openziti/foundation/channel2" + "github.com/openziti/foundation/channel" ) type resyncHandler struct { @@ -41,7 +41,7 @@ func (h *resyncHandler) ContentType() int32 { return env.RequestClientReSyncType } -func (h *resyncHandler) HandleReceive(msg *channel2.Message, ch channel2.Channel) { +func (h *resyncHandler) HandleReceive(msg *channel.Message, ch channel.Channel) { resyncReq := &edge_ctrl_pb.RequestClientReSync{} if err := proto.Unmarshal(msg.Body, resyncReq); err != nil { pfxlog.Logger().WithError(err).Error("could not unmarshal RequestClientReSync") diff --git a/controller/handler_edge_ctrl/tunnel_health_event.go b/controller/handler_edge_ctrl/tunnel_health_event.go index 1dd57e38e..7679f4e12 100644 --- a/controller/handler_edge_ctrl/tunnel_health_event.go +++ b/controller/handler_edge_ctrl/tunnel_health_event.go @@ -20,7 +20,7 @@ import ( "github.com/openziti/edge/controller/env" "github.com/openziti/edge/edge_common" "github.com/openziti/edge/pb/edge_ctrl_pb" - "github.com/openziti/foundation/channel2" + "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 channel2.Channel) channel2.ReceiveHandler { +func NewTunnelHealthEventHandler(appEnv *env.AppEnv, ch channel.Channel) channel.ReceiveHandler { serviceEventMetrics := appEnv.GetHostController().GetNetwork().GetServiceEventsMetricsRegistry() return &tunnelHealthEventHandler{ baseRequestHandler: baseRequestHandler{ @@ -51,7 +51,7 @@ func (self *tunnelHealthEventHandler) Label() string { return "tunnel.health.event" } -func (self *tunnelHealthEventHandler) HandleReceive(msg *channel2.Message, _ channel2.Channel) { +func (self *tunnelHealthEventHandler) HandleReceive(msg *channel.Message, _ channel.Channel) { terminatorId, _ := msg.GetStringHeader(int32(edge_ctrl_pb.Header_TerminatorId)) checkPassed, _ := msg.GetBoolHeader(int32(edge_ctrl_pb.Header_CheckPassed)) diff --git a/controller/handler_edge_ctrl/update_terminator.go b/controller/handler_edge_ctrl/update_terminator.go index 3f9e47b68..d784a4183 100644 --- a/controller/handler_edge_ctrl/update_terminator.go +++ b/controller/handler_edge_ctrl/update_terminator.go @@ -23,7 +23,7 @@ import ( "github.com/openziti/edge/controller/persistence" "github.com/openziti/edge/edge_common" "github.com/openziti/edge/pb/edge_ctrl_pb" - "github.com/openziti/foundation/channel2" + "github.com/openziti/foundation/channel" "github.com/sirupsen/logrus" ) @@ -31,7 +31,7 @@ type updateTerminatorHandler struct { baseRequestHandler } -func NewUpdateTerminatorHandler(appEnv *env.AppEnv, ch channel2.Channel) channel2.ReceiveHandler { +func NewUpdateTerminatorHandler(appEnv *env.AppEnv, ch channel.Channel) channel.ReceiveHandler { return &updateTerminatorHandler{ baseRequestHandler{ ch: ch, @@ -48,7 +48,7 @@ func (self *updateTerminatorHandler) Label() string { return "update.terminator" } -func (self *updateTerminatorHandler) HandleReceive(msg *channel2.Message, ch channel2.Channel) { +func (self *updateTerminatorHandler) HandleReceive(msg *channel.Message, ch channel.Channel) { req := &edge_ctrl_pb.UpdateTerminatorRequest{} if err := proto.Unmarshal(msg.Body, req); err != nil { pfxlog.ContextLogger(ch.Label()).WithError(err).Error("could not unmarshal UpdateTerminator") @@ -93,7 +93,7 @@ func (self *updateTerminatorHandler) UpdateTerminator(ctx *UpdateTerminatorReque logger = logger.WithField("serviceId", terminator.Service) logger.Info("updated terminator") - responseMsg := channel2.NewMessage(int32(edge_ctrl_pb.ContentType_UpdateTerminatorResponseType), nil) + responseMsg := channel.NewMessage(int32(edge_ctrl_pb.ContentType_UpdateTerminatorResponseType), nil) responseMsg.ReplyTo(ctx.msg) if err := self.ch.Send(responseMsg); err != nil { logger.WithError(err).Error("failed to send update terminator response") diff --git a/controller/handler_edge_ctrl/update_tunnel_terminator.go b/controller/handler_edge_ctrl/update_tunnel_terminator.go index 772040ff4..3e8e08b2a 100644 --- a/controller/handler_edge_ctrl/update_tunnel_terminator.go +++ b/controller/handler_edge_ctrl/update_tunnel_terminator.go @@ -22,7 +22,7 @@ import ( "github.com/openziti/edge/controller/env" "github.com/openziti/edge/edge_common" "github.com/openziti/edge/pb/edge_ctrl_pb" - "github.com/openziti/foundation/channel2" + "github.com/openziti/foundation/channel" "github.com/sirupsen/logrus" ) @@ -30,7 +30,7 @@ type updateTunnelTerminatorHandler struct { baseRequestHandler } -func NewUpdateTunnelTerminatorHandler(appEnv *env.AppEnv, ch channel2.Channel) channel2.ReceiveHandler { +func NewUpdateTunnelTerminatorHandler(appEnv *env.AppEnv, ch channel.Channel) channel.ReceiveHandler { return &updateTunnelTerminatorHandler{ baseRequestHandler{ ch: ch, @@ -47,7 +47,7 @@ func (self *updateTunnelTerminatorHandler) Label() string { return "tunnel.update.terminator" } -func (self *updateTunnelTerminatorHandler) HandleReceive(msg *channel2.Message, ch channel2.Channel) { +func (self *updateTunnelTerminatorHandler) HandleReceive(msg *channel.Message, ch channel.Channel) { req := &edge_ctrl_pb.UpdateTunnelTerminatorRequest{} if err := proto.Unmarshal(msg.Body, req); err != nil { pfxlog.ContextLogger(ch.Label()).WithError(err).Error("could not unmarshal UpdateTerminator") @@ -86,7 +86,7 @@ func (self *updateTunnelTerminatorHandler) UpdateTerminator(ctx *UpdateTunnelTer logger = logger.WithField("serviceId", terminator.Service) logger.Info("updated terminator") - responseMsg := channel2.NewMessage(int32(edge_ctrl_pb.ContentType_UpdateTunnelTerminatorResponseType), nil) + responseMsg := channel.NewMessage(int32(edge_ctrl_pb.ContentType_UpdateTunnelTerminatorResponseType), nil) responseMsg.ReplyTo(ctx.msg) if err := self.ch.Send(responseMsg); err != nil { logger.WithError(err).Error("failed to send update tunnel terminator response") diff --git a/controller/handler_edge_ctrl/validate_sessions.go b/controller/handler_edge_ctrl/validate_sessions.go index 4afe368f5..eb6202cc6 100644 --- a/controller/handler_edge_ctrl/validate_sessions.go +++ b/controller/handler_edge_ctrl/validate_sessions.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/channel2" + "github.com/openziti/foundation/channel" "go.etcd.io/bbolt" ) @@ -29,7 +29,7 @@ type validateSessionsHandler struct { baseRequestHandler } -func NewValidateSessionsHandler(appEnv *env.AppEnv, ch channel2.Channel) channel2.ReceiveHandler { +func NewValidateSessionsHandler(appEnv *env.AppEnv, ch channel.Channel) channel.ReceiveHandler { return &validateSessionsHandler{ baseRequestHandler{ ch: ch, @@ -46,7 +46,7 @@ func (self *validateSessionsHandler) Label() string { return "validate.sessions" } -func (self *validateSessionsHandler) HandleReceive(msg *channel2.Message, ch channel2.Channel) { +func (self *validateSessionsHandler) HandleReceive(msg *channel.Message, ch channel.Channel) { req := &edge_ctrl_pb.ValidateSessionsRequest{} if err := proto.Unmarshal(msg.Body, req); err != nil { pfxlog.ContextLogger(ch.Label()).WithError(err).Error("could not unmarshal ValidateSessionsRequest") @@ -86,7 +86,7 @@ func (self *validateSessionsHandler) validateSessions(req *edge_ctrl_pb.Validate return } - msg := channel2.NewMessage(int32(edge_ctrl_pb.ContentType_SessionRemovedType), body) + msg := channel.NewMessage(int32(edge_ctrl_pb.ContentType_SessionRemovedType), body) if err := self.ch.Send(msg); err != nil { pfxlog.ContextLogger(self.ch.Label()).WithError(err).Error("failed to send validate sessions request") return diff --git a/controller/server/controller.go b/controller/server/controller.go index 56a52eb75..e165d18f7 100644 --- a/controller/server/controller.go +++ b/controller/server/controller.go @@ -22,6 +22,7 @@ import ( 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" @@ -141,10 +142,10 @@ func (c *Controller) SetHostController(h env.HostController) { } } -func (c *Controller) GetCtrlHandlers(ch channel2.Channel) []channel2.ReceiveHandler { +func (c *Controller) GetCtrlHandlers(ch channel.Channel) []channel.ReceiveHandler { tunnelState := handler_edge_ctrl.NewTunnelState() - return []channel2.ReceiveHandler{ + return []channel.ReceiveHandler{ handler_edge_ctrl.NewSessionHeartbeatHandler(c.AppEnv), handler_edge_ctrl.NewCreateCircuitHandler(c.AppEnv, ch), handler_edge_ctrl.NewCreateTerminatorHandler(c.AppEnv, ch), @@ -321,12 +322,12 @@ func (c *Controller) Shutdown() { type subctrl struct { parent *Controller - channel channel2.Channel + channel channel.Channel } -func (c *subctrl) GetTraceDecoders() []channel2.TraceMessageDecoder { - return []channel2.TraceMessageDecoder{ - &edge_ctrl_pb.Decoder{}, +func (c *subctrl) GetTraceDecoders() []channel.TraceMessageDecoder { + return []channel.TraceMessageDecoder{ + edge_ctrl_pb.Decoder{}, } } @@ -341,7 +342,7 @@ func (c *subctrl) Enabled() bool { return c.parent.Enabled() } -func (c *subctrl) BindChannel(ch channel2.Channel) error { +func (c *subctrl) BindChannel(ch channel.Channel) error { for _, h := range c.parent.GetCtrlHandlers(ch) { ch.AddReceiveHandler(h) } @@ -349,7 +350,7 @@ func (c *subctrl) BindChannel(ch channel2.Channel) error { return nil } -func (c *subctrl) Run(channel2.Channel, boltz.Db, chan struct{}) error { +func (c *subctrl) Run(channel.Channel, boltz.Db, chan struct{}) error { return nil } diff --git a/controller/sync_strats/rtx.go b/controller/sync_strats/rtx.go index 84f9cae28..fd4960367 100644 --- a/controller/sync_strats/rtx.go +++ b/controller/sync_strats/rtx.go @@ -22,6 +22,7 @@ import ( "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" @@ -36,7 +37,7 @@ type RouterSender struct { Id string EdgeRouter *model.EdgeRouter Router *network.Router - send chan *channel2.Message + send chan *channel.Message closeNotify chan struct{} running concurrenz.AtomicBoolean @@ -48,7 +49,7 @@ func newRouterSender(edgeRouter *model.EdgeRouter, router *network.Router, sendB Id: eid.New(), EdgeRouter: edgeRouter, Router: router, - send: make(chan *channel2.Message, sendBufferSize), + send: make(chan *channel.Message, sendBufferSize), closeNotify: make(chan struct{}, 0), running: concurrenz.AtomicBoolean(1), RouterState: env.NewLockingRouterStatus(), @@ -93,7 +94,7 @@ func (rtx *RouterSender) logger() *logrus.Entry { WithField("routerChannelIsOpen", !rtx.Router.Control.IsClosed()) } -func (rtx *RouterSender) Send(msg *channel2.Message) error { +func (rtx *RouterSender) Send(msg *channel.Message) error { if !rtx.running.Get() { rtx.logger().Errorf("cannot send to router [%s], rtx'er is stopped", rtx.Router.Id) return errors.Errorf("cannot send to router [%s], rtx'er is stopped", rtx.Router.Id) diff --git a/controller/sync_strats/sync_instant.go b/controller/sync_strats/sync_instant.go index 52b782414..28d87de68 100644 --- a/controller/sync_strats/sync_instant.go +++ b/controller/sync_strats/sync_instant.go @@ -29,7 +29,7 @@ import ( "github.com/openziti/edge/pb/edge_ctrl_pb" "github.com/openziti/fabric/build" "github.com/openziti/fabric/controller/network" - "github.com/openziti/foundation/channel2" + "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 channel2.ReceiveHandler - resyncHandler channel2.ReceiveHandler + helloHandler channel.ReceiveHandler + resyncHandler channel.ReceiveHandler ae *env.AppEnv routerConnectedQueue chan *RouterSender @@ -222,7 +222,7 @@ func (strategy *InstantStrategy) ApiSessionUpdated(apiSession *persistence.ApiSe strategy.rtxMap.Range(func(rtx *RouterSender) { content, _ := proto.Marshal(apiSessionAdded) - msg := channel2.NewMessage(env.ApiSessionUpdatedType, content) + msg := channel.NewMessage(env.ApiSessionUpdatedType, content) msg.Headers[env.SyncStrategyTypeHeader] = []byte(strategy.Type()) msg.Headers[env.SyncStrategyStateHeader] = nil _ = rtx.Send(msg) @@ -236,7 +236,7 @@ func (strategy *InstantStrategy) ApiSessionDeleted(apiSession *persistence.ApiSe strategy.rtxMap.Range(func(rtx *RouterSender) { content, _ := proto.Marshal(sessionRemoved) - msg := channel2.NewMessage(env.ApiSessionRemovedType, content) + msg := channel.NewMessage(env.ApiSessionRemovedType, content) _ = rtx.Send(msg) }) } @@ -248,7 +248,7 @@ func (strategy *InstantStrategy) SessionDeleted(session *persistence.Session) { strategy.rtxMap.Range(func(rtx *RouterSender) { content, _ := proto.Marshal(sessionRemoved) - msg := channel2.NewMessage(env.SessionRemovedType, content) + msg := channel.NewMessage(env.SessionRemovedType, content) _ = rtx.Send(msg) }) } @@ -324,7 +324,7 @@ func (strategy *InstantStrategy) sendHello(rtx *RouterSender) { return } - if err = rtx.Router.Control.SendWithTimeout(channel2.NewMessage(env.ServerHelloType, buf), strategy.HelloSendTimeout); err != nil { + if err = rtx.Router.Control.Send(channel.NewMessage(env.ServerHelloType, buf).WithTimeout(strategy.HelloSendTimeout)); 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") @@ -482,7 +482,7 @@ func (strategy *InstantStrategy) sendApiSessionAdded(rtx *RouterSender, isFullSt msgContentBytes, _ := proto.Marshal(msgContent) - msg := channel2.NewMessage(env.ApiSessionAddedType, msgContentBytes) + msg := channel.NewMessage(env.ApiSessionAddedType, msgContentBytes) msg.Headers[env.SyncStrategyTypeHeader] = []byte(strategy.Type()) msg.Headers[env.SyncStrategyStateHeader] = stateBytes diff --git a/go.mod b/go.mod index cf833534d..2b4ab4c6d 100644 --- a/go.mod +++ b/go.mod @@ -42,8 +42,8 @@ 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.17 - github.com/openziti/foundation v0.16.13 + github.com/openziti/fabric v0.17.18 + github.com/openziti/foundation v0.16.14 github.com/openziti/sdk-golang v0.15.130 github.com/orcaman/concurrent-map v0.0.0-20210106121528-16402b402231 github.com/pkg/errors v0.9.1 diff --git a/go.sum b/go.sum index 858d39c65..cce9c7c89 100644 --- a/go.sum +++ b/go.sum @@ -705,10 +705,11 @@ github.com/opentracing/opentracing-go v1.2.0 h1:uEJPy/1a5RIPAJ0Ov+OIO8OxWu77jEv+ 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/dilithium v0.3.3/go.mod h1:vsCjI2AU/hon9e+dLhUFbCNGesJDj2ASgkySOcpmvjo= -github.com/openziti/fabric v0.17.17 h1:sjAQqmYoC3cF1Sh6RsSkKtNBDkO/NiLqqS1n8Is7Jbw= -github.com/openziti/fabric v0.17.17/go.mod h1:n1ysT2vSj1CFFXCsLB+LpzaVwPnuHUelSC8Q4n5knws= -github.com/openziti/foundation v0.16.13 h1:LvZ6pzTE9NayzqWBfW2J/bGv4X/67iJ3H5cqsB3DYI4= +github.com/openziti/fabric v0.17.18 h1:0nmCwTG3Tmwhjm4f3khcKtIS9IPBcNajNQ3OAYIf7xM= +github.com/openziti/fabric v0.17.18/go.mod h1:HIquY6ph5+yC9vL38EQh2n5qWSeu5y81Pgc1YPhtPVQ= github.com/openziti/foundation v0.16.13/go.mod h1:SHG4x9kLP+M7sn0BJpoc/RbSIh1Hm1BxwBPHFmCf7qE= +github.com/openziti/foundation v0.16.14 h1:fwiQK7bdhs+z9idFVpEfWHzk7p0DGIvbrFbYJ6UcU8c= +github.com/openziti/foundation v0.16.14/go.mod h1:SHG4x9kLP+M7sn0BJpoc/RbSIh1Hm1BxwBPHFmCf7qE= github.com/openziti/sdk-golang v0.15.130 h1:XnkWlUZEVJb1qusSyNFUiQc8Bc/IAJbv+W/va0O6MLg= github.com/openziti/sdk-golang v0.15.130/go.mod h1:gx60OgzPgn/7pbeOu5OerSA98jWL0s+jnITiefEJFNo= github.com/orcaman/concurrent-map v0.0.0-20190826125027-8c72a8bb44f6/go.mod h1:Lu3tH6HLW3feq74c2GC+jIMS/K2CFcDWnWD9XkenwhI= diff --git a/pb/edge_ctrl_pb/decoder.go b/pb/edge_ctrl_pb/decoder.go index 80ec235bf..b81dc2b48 100644 --- a/pb/edge_ctrl_pb/decoder.go +++ b/pb/edge_ctrl_pb/decoder.go @@ -20,6 +20,7 @@ import ( "fmt" "github.com/golang/protobuf/proto" "github.com/michaelquigley/pfxlog" + "github.com/openziti/foundation/channel" "github.com/openziti/foundation/channel2" "strings" ) @@ -28,7 +29,7 @@ type Decoder struct{} const DECODER = "edge_ctrl" -func (d Decoder) Decode(msg *channel2.Message) ([]byte, bool) { +func (d Decoder) Decode(msg *channel.Message) ([]byte, bool) { switch msg.ContentType { case int32(ContentType_ServerHelloType): request := &ServerHello{}