mirror of
https://github.com/openziti/ziti.git
synced 2026-10-13 15:51:21 +00:00
More transport configuration wiring.
This commit is contained in:
+29
-18
@@ -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
|
||||
}
|
||||
|
||||
|
||||
+3
-3
@@ -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
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user