From eb8aeb99bc797728e534a21f619a52a27afd0085 Mon Sep 17 00:00:00 2001 From: Michael Quigley Date: Wed, 29 Jul 2020 17:12:07 -0400 Subject: [PATCH] More transport configuration wiring. --- router/config.go | 47 ++++++++++++++++++------------ router/router.go | 6 ++-- router/xlink_transport/dialer.go | 3 +- router/xlink_transport/factory.go | 8 +++-- router/xlink_transport/listener.go | 6 ++-- 5 files changed, 44 insertions(+), 26 deletions(-) diff --git a/router/config.go b/router/config.go index e8511850e..00007ff1f 100644 --- a/router/config.go +++ b/router/config.go @@ -27,9 +27,11 @@ import ( "github.com/openziti/foundation/config" "github.com/openziti/foundation/identity/identity" "github.com/openziti/foundation/transport" + "github.com/pkg/errors" "github.com/spf13/pflag" "gopkg.in/yaml.v2" "io/ioutil" + "reflect" "time" ) @@ -82,6 +84,7 @@ type Config struct { } Dialers map[string]xgress.OptionsData Listeners []listenerBinding + Transport map[interface{}]interface{} src map[interface{}]interface{} } @@ -258,24 +261,6 @@ func LoadConfig(path string) (*Config, error) { } } - if value, found := cfgmap["listeners"]; found { - if subarr, ok := value.([]interface{}); ok { - for _, value := range subarr { - if submap, ok := value.(map[interface{}]interface{}); ok { - binding, found := submap["binding"].(string) - if !found { - return nil, fmt.Errorf("[listener] must provide [binding] (%v)", submap) - } - _, found = submap["address"] - if !found { - return nil, fmt.Errorf("[listener] must provide [address] (%v)", submap) - } - cfg.Listeners = append(cfg.Listeners, listenerBinding{name: binding, options: submap}) - } - } - } - } - if value, found := cfgmap["dialers"]; found { if subarr, ok := value.([]interface{}); ok { for _, value := range subarr { @@ -294,6 +279,32 @@ func LoadConfig(path string) (*Config, error) { } } + if value, found := cfgmap["listeners"]; found { + if subarr, ok := value.([]interface{}); ok { + for _, value := range subarr { + if submap, ok := value.(map[interface{}]interface{}); ok { + binding, found := submap["binding"].(string) + if !found { + return nil, fmt.Errorf("[listener] must provide [binding] (%v)", submap) + } + _, found = submap["address"] + if !found { + return nil, fmt.Errorf("[listener] must provide [address] (%v)", submap) + } + cfg.Listeners = append(cfg.Listeners, listenerBinding{name: binding, options: submap}) + } + } + } + } + + if value, found := cfgmap["transport"]; found { + if submap, ok := value.(map[interface{}]interface{}); ok { + cfg.Transport = submap + } else { + return nil, errors.Errorf("invalid [transport] type (%v)", reflect.TypeOf(value)) + } + } + return cfg, nil } diff --git a/router/router.go b/router/router.go index 29d88751d..b81b7769c 100644 --- a/router/router.go +++ b/router/router.go @@ -176,11 +176,11 @@ func (self *Router) registerComponents() error { self.config.Forwarder, self.metricsRegistry, ) - self.xlinkFactories["transport"] = xlink_transport.NewFactory(xlinkAccepter, xlinkChAccepter) + self.xlinkFactories["transport"] = xlink_transport.NewFactory(xlinkAccepter, xlinkChAccepter, self.config.Transport) - xgress.GlobalRegistry().Register("proxy", xgress_proxy.NewFactory(self.config.Id, self, nil)) + xgress.GlobalRegistry().Register("proxy", xgress_proxy.NewFactory(self.config.Id, self, self.config.Transport)) xgress.GlobalRegistry().Register("proxy_udp", xgress_proxy_udp.NewFactory(self)) - xgress.GlobalRegistry().Register("transport", xgress_transport.NewFactory(self.config.Id, self, nil)) + xgress.GlobalRegistry().Register("transport", xgress_transport.NewFactory(self.config.Id, self, self.config.Transport)) xgress.GlobalRegistry().Register("transport_udp", xgress_transport_udp.NewFactory(self.config.Id, self)) return nil diff --git a/router/xlink_transport/dialer.go b/router/xlink_transport/dialer.go index 33f41670a..186f0f785 100644 --- a/router/xlink_transport/dialer.go +++ b/router/xlink_transport/dialer.go @@ -30,7 +30,7 @@ func (self *dialer) Dial(addressString string, linkId *identity.TokenId) error { logrus.Infof("dialing link [l/%s]", linkId.Token) dialer := channel2.NewClassicDialer(linkId, address, nil) - ch, err := channel2.NewChannel("l/"+linkId.Token, dialer, self.config.options) + ch, err := channel2.NewChannelWithTransportConfig("l/"+linkId.Token, dialer, self.config.options, self.c) if err == nil { xlink := &impl{id: linkId, ch: ch} @@ -59,4 +59,5 @@ type dialer struct { config *dialerConfig accepter xlink.Accepter chAccepter ChannelAccepter + c transport.Configuration } diff --git a/router/xlink_transport/factory.go b/router/xlink_transport/factory.go index bf5d18647..2ee7782d4 100644 --- a/router/xlink_transport/factory.go +++ b/router/xlink_transport/factory.go @@ -20,10 +20,11 @@ import ( "fmt" "github.com/openziti/fabric/router/xlink" "github.com/openziti/foundation/identity/identity" + "github.com/openziti/foundation/transport" ) -func NewFactory(accepter xlink.Accepter, chAccepter ChannelAccepter) xlink.Factory { - return &factory{accepter: accepter, chAccepter: chAccepter} +func NewFactory(accepter xlink.Accepter, chAccepter ChannelAccepter, c transport.Configuration) xlink.Factory { + return &factory{accepter: accepter, chAccepter: chAccepter, c: c} } func (self *factory) CreateListener(id *identity.TokenId, _ xlink.Forwarder, configData map[interface{}]interface{}) (xlink.Listener, error) { @@ -36,6 +37,7 @@ func (self *factory) CreateListener(id *identity.TokenId, _ xlink.Forwarder, con config: config, accepter: self.accepter, chAccepter: self.chAccepter, + c: self.c, }, nil } @@ -49,10 +51,12 @@ func (self *factory) CreateDialer(id *identity.TokenId, _ xlink.Forwarder, confi config: config, accepter: self.accepter, chAccepter: self.chAccepter, + c: self.c, }, nil } type factory struct { accepter xlink.Accepter chAccepter ChannelAccepter + c transport.Configuration } diff --git a/router/xlink_transport/listener.go b/router/xlink_transport/listener.go index 619694a5d..97fe4e605 100644 --- a/router/xlink_transport/listener.go +++ b/router/xlink_transport/listener.go @@ -22,11 +22,12 @@ import ( "github.com/openziti/fabric/router/xlink" "github.com/openziti/foundation/channel2" "github.com/openziti/foundation/identity/identity" + "github.com/openziti/foundation/transport" "github.com/sirupsen/logrus" ) func (self *listener) Listen() error { - listener := channel2.NewClassicListener(self.id, self.config.bind, self.config.options.ConnectOptions) + listener := channel2.NewClassicListenerWithConfiguration(self.id, self.config.bind, self.config.options.ConnectOptions, self.c) self.listener = listener if err := self.listener.Listen(); err != nil { @@ -46,7 +47,7 @@ func (self *listener) Close() error { func (self *listener) acceptLoop() { for { - ch, err := channel2.NewChannel("link", self.listener, self.config.options) + ch, err := channel2.NewChannelWithTransportConfig("link", self.listener, self.config.options, self.c) if err == nil { xlink := &impl{id: ch.Id(), ch: ch} logrus.Infof("accepted link id [l/%s]", xlink.Id().Token) @@ -78,4 +79,5 @@ type listener struct { listener channel2.UnderlayListener accepter xlink.Accepter chAccepter ChannelAccepter + c transport.Configuration }