xlink_transport roughed-in using roughly the same scruffy wiring code as the single-listener router implementation. Will revisit/refactor as xlink_transwarp develops. (#54)

router/router.go

@@ -32,6 +32,7 @@ import (
 	"github.com/netfoundry/ziti-fabric/router/xgress_proxy_udp"
 	"github.com/netfoundry/ziti-fabric/router/xgress_transport"
 	"github.com/netfoundry/ziti-fabric/router/xgress_transport_udp"
+	"github.com/netfoundry/ziti-fabric/router/xlink"
 	"github.com/netfoundry/ziti-foundation/channel2"
 	"github.com/netfoundry/ziti-foundation/profiler"
 	"github.com/netfoundry/ziti-foundation/util/info"
@@ -49,6 +50,7 @@ type Router struct {
 	forwarder       *forwarder.Forwarder
 	xctrls          []xctrl.Xctrl
 	xctrlDone       chan struct{}
+	xlinks          map[string]xlink.Xlink
 	metricsRegistry metrics.Registry
 }
 

router/xlink/xlink.go

@@ -24,6 +24,6 @@ type Factory interface {
 
 type Xlink interface {
 	Listen() error
-	Dial(address string) error
+	Dial(address, linkId string) error
 	GetAdvertisement() string
 }

router/xlink_transport/factory.go

@@ -18,6 +18,7 @@ package xlink_transport
 
 import (
 	"fmt"
+	"github.com/netfoundry/ziti-fabric/metrics"
 	forwarder2 "github.com/netfoundry/ziti-fabric/router/forwarder"
 	"github.com/netfoundry/ziti-fabric/router/xgress"
 	"github.com/netfoundry/ziti-fabric/router/xlink"
@@ -25,12 +26,13 @@ import (
 	"github.com/netfoundry/ziti-foundation/identity/identity"
 )
 
-func NewFactory(ctrl xgress.CtrlChannel, forwarder *forwarder2.Forwarder, forwarderOptions *forwarder2.Options, options *channel2.Options) xlink.Factory {
+func NewFactory(ctrl xgress.CtrlChannel, f *forwarder2.Forwarder, fo *forwarder2.Options, mr metrics.Registry, lo *channel2.Options) xlink.Factory {
 	return &factory{
 		ctrl:             ctrl,
-		forwarder:        forwarder,
-		forwarderOptions: forwarderOptions,
-		options:          options,
+		forwarder:        f,
+		forwarderOptions: fo,
+		metricsRegistry:  mr,
+		linkOptions:      lo,
 	}
 }
 
@@ -45,7 +47,7 @@ func (self *factory) Create(id *identity.TokenId, configData map[interface{}]int
 		ctrl:             self.ctrl,
 		forwarder:        self.forwarder,
 		forwarderOptions: self.forwarderOptions,
-		options:          self.options,
+		linkOptions:      self.linkOptions,
 	}, nil
 }
 
@@ -53,5 +55,6 @@ type factory struct {
 	ctrl             xgress.CtrlChannel
 	forwarder        *forwarder2.Forwarder
 	forwarderOptions *forwarder2.Options
-	options          *channel2.Options
+	metricsRegistry  metrics.Registry
+	linkOptions      *channel2.Options
 }

router/xlink_transport/xlink.go

@@ -18,11 +18,16 @@ package xlink_transport
 
 import (
 	"fmt"
+	"github.com/golang/protobuf/proto"
+	"github.com/netfoundry/ziti-fabric/metrics"
+	"github.com/netfoundry/ziti-fabric/pb/ctrl_pb"
 	"github.com/netfoundry/ziti-fabric/router/forwarder"
 	"github.com/netfoundry/ziti-fabric/router/handler_link"
 	"github.com/netfoundry/ziti-fabric/router/xgress"
 	"github.com/netfoundry/ziti-foundation/channel2"
 	"github.com/netfoundry/ziti-foundation/identity/identity"
+	"github.com/netfoundry/ziti-foundation/transport"
+	log "github.com/sirupsen/logrus"
 )
 
 func (self *impl) Listen() error {
@@ -30,11 +35,63 @@ func (self *impl) Listen() error {
 	if err := self.listener.Listen(); err != nil {
 		return fmt.Errorf("error listening (%w)", err)
 	}
-	go handler_link.NewAccepter(self.id, self.ctrl, self.forwarder, self.listener, self.options, self.forwarderOptions)
+	go handler_link.NewAccepter(self.id, self.ctrl, self.forwarder, self.listener, self.linkOptions, self.forwarderOptions)
 	return nil
 }
 
-func (_ *impl) Dial(address string) error {
+func (self *impl) Dial(addressString, linkId string) error {
+	address, err := transport.ParseAddress(addressString)
+	if err == nil {
+		linkId := self.id.ShallowCloneWithNewToken(linkId)
+		name := "l/" + linkId.Token
+
+		dialer := channel2.NewClassicDialer(linkId, address, nil)
+		ch, err := channel2.NewChannel(name, dialer, self.linkOptions)
+		if err == nil {
+			link := forwarder.NewLink(linkId, ch)
+			if err := link.Channel.Bind(handler_link.NewBindHandler(self.id, link, self.ctrl, self.forwarder)); err == nil {
+				self.forwarder.RegisterLink(link)
+
+				go metrics.ProbeLatency(
+					link.Channel,
+					self.metricsRegistry.Histogram("link."+linkId.Token+".latency"),
+					self.forwarderOptions.LatencyProbeInterval,
+				)
+
+				linkMsg := &ctrl_pb.Link{Id: link.Id.Token}
+				body, err := proto.Marshal(linkMsg)
+				if err == nil {
+					msg := channel2.NewMessage(int32(ctrl_pb.ContentType_LinkType), body)
+					if err := self.ctrl.Channel().Send(msg); err == nil {
+						log.Infof("link [l/%s] up", link.Id.Token)
+					} else {
+						log.Errorf("unexpected error sending link (%s)", err)
+					}
+
+				} else {
+					log.Errorf("unexpected error (%s)", err)
+				}
+			} else {
+				log.Errorf("error binding link (%s)", err)
+			}
+		} else {
+			log.Errorf("link dialing failed [%s]", address.String())
+
+			fault := &ctrl_pb.Fault{Subject: ctrl_pb.FaultSubject_LinkFault, Id: linkId.Token}
+			body, err := proto.Marshal(fault)
+			if err == nil {
+				msg := channel2.NewMessage(int32(ctrl_pb.ContentType_FaultType), body)
+				if err := self.ctrl.Channel().Send(msg); err != nil {
+					log.Errorf("error sending fault (%s)", err)
+				}
+			} else {
+				log.Errorf("unexpected error (%s)", err)
+			}
+		}
+	} else {
+		log.Errorf("link address parsing failed (%s)", err)
+	}
+
 	return nil
 }
 
@@ -49,5 +106,6 @@ type impl struct {
 	ctrl             xgress.CtrlChannel
 	forwarder        *forwarder.Forwarder
 	forwarderOptions *forwarder.Options
-	options          *channel2.Options
+	metricsRegistry  metrics.Registry
+	linkOptions      *channel2.Options
 }