More receive handler, network, route sender wiring. (#168)

controller/handler_ctrl/route_result.go

@@ -20,6 +20,7 @@ import (
 	"github.com/openziti/fabric/controller/network"
 	"github.com/openziti/fabric/ctrl_msg"
 	"github.com/openziti/foundation/channel2"
+	"github.com/sirupsen/logrus"
 )
 
 type routeResultHandler struct {
@@ -39,4 +40,14 @@ func (self *routeResultHandler) ContentType() int32 {
 }
 
 func (self *routeResultHandler) HandleReceive(msg *channel2.Message, ch channel2.Channel) {
+	_, success := msg.Headers[ctrl_msg.RouteResultSuccessHeader]
+	sessionId := string(msg.Body)
+	routing := self.network.RouteResult(self.r, sessionId, success)
+	if !routing {
+		self.notRoutingSession(sessionId)
+	}
+}
+
+func (self *routeResultHandler) notRoutingSession(sessionId string) {
+	logrus.Warnf("not routing session [s/%s] for router [r/%s], sending unroute", sessionId, self.r.Id)
 }

controller/network/network.go

@@ -53,6 +53,7 @@ type Network struct {
 	linkController         *linkController
 	linkChanged            chan *Link
 	sessionController      *sessionController
+	routeSenderController  *routeSenderController
 	sequence               *sequence.Sequence
 	eventDispatcher        event.Dispatcher
 	traceController        trace.Controller
@@ -77,21 +78,22 @@ func NewNetwork(nodeId *identity.TokenId, options *Options, database boltz.Db, m
 	eventDispatcher := event.NewDispatcher()
 
 	network := &Network{
-		Controllers:       controllers,
-		nodeId:            nodeId,
-		options:           options,
-		routerChanged:     make(chan *Router),
-		linkController:    newLinkController(),
-		linkChanged:       make(chan *Link),
-		sessionController: newSessionController(),
-		sequence:          sequence.NewSequence(),
-		eventDispatcher:   eventDispatcher,
-		traceController:   trace.NewController(),
-		shutdownChan:      make(chan struct{}),
-		strategyRegistry:  xt.GlobalRegistry(),
-		lastSnapshot:      time.Now().Add(-time.Hour),
-		metricsRegistry:   metrics.NewRegistry(nodeId.Token, nil),
-		VersionProvider:   versionProvider,
+		Controllers:           controllers,
+		nodeId:                nodeId,
+		options:               options,
+		routerChanged:         make(chan *Router),
+		linkController:        newLinkController(),
+		linkChanged:           make(chan *Link),
+		sessionController:     newSessionController(),
+		routeSenderController: newRouteSenderController(),
+		sequence:              sequence.NewSequence(),
+		eventDispatcher:       eventDispatcher,
+		traceController:       trace.NewController(),
+		shutdownChan:          make(chan struct{}),
+		strategyRegistry:      xt.GlobalRegistry(),
+		lastSnapshot:          time.Now().Add(-time.Hour),
+		metricsRegistry:       metrics.NewRegistry(nodeId.Token, nil),
+		VersionProvider:       versionProvider,
 	}
 	metrics.Init(metricsCfg)
 	events.AddMetricsEventHandler(network)
@@ -176,6 +178,10 @@ func (network *Network) GetAllSessions() []*session {
 	return network.sessionController.all()
 }
 
+func (network *Network) RouteResult(r *Router, sessionId string, success bool) bool {
+	return network.routeSenderController.forwardRouteResult(r, sessionId, success)
+}
+
 func (network *Network) GetEventDispatcher() event.Dispatcher {
 	return network.eventDispatcher
 }

controller/network/routesender.go

@@ -18,9 +18,22 @@ package network
 
 import (
 	"github.com/openziti/fabric/pb/ctrl_pb"
+	cmap "github.com/orcaman/concurrent-map"
 	"time"
 )
 
+type routeSenderController struct {
+	senders cmap.ConcurrentMap // map[string]*routeSender
+}
+
+func newRouteSenderController() *routeSenderController {
+	return &routeSenderController{}
+}
+
+func (self *routeSenderController) forwardRouteResult(r *Router, sessionId string, success bool) bool {
+	return false
+}
+
 type routeSender struct {
 	sessionId string
 	circuit   *Circuit
@@ -39,4 +52,4 @@ func newRouteSender(sessionId string, timeout time.Duration, maxTries int) *rout
 
 func (self *routeSender) send(circuit *Circuit, routeMsgs []*ctrl_pb.Route) error {
 	return nil
-}
+}