Rounding out the xlink_transwarp skeleton. (#44)

This commit is contained in:
Michael Quigley
2020-03-24 11:51:21 -04:00
parent a09a608449
commit 75aab14ef8
9 changed files with 198 additions and 24 deletions
+4 -1
View File
@@ -34,6 +34,7 @@ import (
"github.com/netfoundry/ziti-fabric/router/xgress_transport_udp"
"github.com/netfoundry/ziti-fabric/router/xlink"
"github.com/netfoundry/ziti-fabric/router/xlink_transport"
"github.com/netfoundry/ziti-fabric/router/xlink_transwarp"
"github.com/netfoundry/ziti-foundation/channel2"
"github.com/netfoundry/ziti-foundation/profiler"
"github.com/netfoundry/ziti-foundation/util/info"
@@ -135,12 +136,14 @@ func (self *Router) startProfiling() {
func (self *Router) registerComponents() error {
self.xlinkFactories = make(map[string]xlink.Factory)
xlinkAccepter := newXlinkAccepter(self.forwarder)
xlinkChAccepter := handler_link.NewChannelAccepter(self,
self.forwarder,
self.config.Forwarder,
self.metricsRegistry,
)
self.xlinkFactories["transport"] = xlink_transport.NewFactory(newXlinkAccepter(self.forwarder), xlinkChAccepter)
self.xlinkFactories["transport"] = xlink_transport.NewFactory(xlinkAccepter, xlinkChAccepter)
self.xlinkFactories["transwarp"] = xlink_transwarp.NewFactory(xlinkAccepter)
xgress.GlobalRegistry().Register("proxy", xgress_proxy.NewFactory(self.config.Id, self))
xgress.GlobalRegistry().Register("proxy_udp", xgress_proxy_udp.NewFactory(self))
+3
View File
@@ -5,6 +5,9 @@ import (
"github.com/netfoundry/ziti-foundation/channel2"
)
// ChannelAccepter can be implemented and provided to the factory to perform channel2 binding and other channel setup
// tasks at accept time.
//
type ChannelAccepter interface {
AcceptChannel(xlink xlink.Xlink, ch channel2.Channel) error
}
+22 -8
View File
@@ -1,3 +1,19 @@
/*
(c) Copyright NetFoundry, Inc.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
https://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package xlink_transport
import (
@@ -9,14 +25,12 @@ import (
"github.com/sirupsen/logrus"
)
func (self *dialerImpl) Dial(addressString string, linkId *identity.TokenId) error {
address, err := transport.ParseAddress(addressString)
if err == nil {
name := "l/" + linkId.Token
logrus.Infof("dialing link [%s]", name)
func (self *dialer) Dial(addressString string, linkId *identity.TokenId) error {
if address, err := transport.ParseAddress(addressString); err == nil {
logrus.Infof("dialing link [l/%s]", linkId.Token)
dialer := channel2.NewClassicDialer(linkId, address, nil)
ch, err := channel2.NewChannel(name, dialer, self.config.options)
ch, err := channel2.NewChannel("l/"+linkId.Token, dialer, self.config.options)
if err == nil {
xlink := &impl{id: linkId, ch: ch}
@@ -33,14 +47,14 @@ func (self *dialerImpl) Dial(addressString string, linkId *identity.TokenId) err
return nil
} else {
return fmt.Errorf("error dialing link [%s] (%w)", name, err)
return fmt.Errorf("error dialing link [l/%s] (%w)", linkId.Token, err)
}
} else {
return fmt.Errorf("error parsing link address [%s] (%w)", addressString, err)
}
}
type dialerImpl struct {
type dialer struct {
id *identity.TokenId
config *dialerConfig
accepter xlink.Accepter
+8 -8
View File
@@ -27,26 +27,26 @@ func NewFactory(accepter xlink.Accepter, chAccepter ChannelAccepter) xlink.Facto
}
func (self *factory) CreateListener(id *identity.TokenId, configData map[interface{}]interface{}) (xlink.Listener, error) {
c, err := loadListenerConfig(configData)
config, err := loadListenerConfig(configData)
if err != nil {
return nil, fmt.Errorf("error loading configuration (%w)", err)
return nil, fmt.Errorf("error loading listener configuration (%w)", err)
}
return &listenerImpl{
return &listener{
id: id,
config: c,
config: config,
accepter: self.accepter,
chAccepter: self.chAccepter,
}, nil
}
func (self *factory) CreateDialer(id *identity.TokenId, configData map[interface{}]interface{}) (xlink.Dialer, error) {
c, err := loadDialerConfig(configData)
config, err := loadDialerConfig(configData)
if err != nil {
return nil, fmt.Errorf("error loading configuration (%w)", err)
return nil, fmt.Errorf("error loading dialer configuration (%w)", err)
}
return &dialerImpl{
return &dialer{
id: id,
config: c,
config: config,
accepter: self.accepter,
chAccepter: self.chAccepter,
}, nil
+4 -4
View File
@@ -24,7 +24,7 @@ import (
"github.com/sirupsen/logrus"
)
func (self *listenerImpl) Listen() error {
func (self *listener) Listen() error {
self.listener = channel2.NewClassicListener(self.id, self.config.bind)
if err := self.listener.Listen(); err != nil {
return fmt.Errorf("error listening (%w)", err)
@@ -33,11 +33,11 @@ func (self *listenerImpl) Listen() error {
return nil
}
func (self *listenerImpl) GetAdvertisement() string {
func (self *listener) GetAdvertisement() string {
return self.config.advertise.String()
}
func (self *listenerImpl) acceptLoop() {
func (self *listener) acceptLoop() {
for {
ch, err := channel2.NewChannel("link", self.listener, self.config.options)
if err == nil {
@@ -62,7 +62,7 @@ func (self *listenerImpl) acceptLoop() {
}
}
type listenerImpl struct {
type listener struct {
id *identity.TokenId
config *listenerConfig
listener channel2.UnderlayListener
+55
View File
@@ -0,0 +1,55 @@
/*
(c) Copyright NetFoundry, Inc.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
https://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package xlink_transwarp
import (
"fmt"
"github.com/netfoundry/ziti-fabric/router/xlink"
"github.com/netfoundry/ziti-foundation/identity/identity"
"github.com/sirupsen/logrus"
"net"
)
func (self *dialer) Dial(addressString string, linkId *identity.TokenId) error {
if address, err := net.ResolveUDPAddr("udp", addressString); err == nil {
name := "l/" + linkId.Token
logrus.Infof("dialing link [%s]", name)
if conn, err := net.DialUDP("udp", self.config.bindAddress, address); err == nil {
xlink := &impl{id: linkId, conn: conn}
if err := self.accepter.Accept(xlink); err != nil {
return fmt.Errorf("error accepting outgoing Xlink (%w)", err)
}
logrus.Infof("bound to [%s]", self.config.bindAddress)
return nil
} else {
return fmt.Errorf("error dialing link [%s] (%w)", name, err)
}
} else {
return fmt.Errorf("error parsing link address [%s] (%w)", addressString, err)
}
}
type dialer struct {
id *identity.TokenId
config *dialerConfig
accepter xlink.Accepter
}
+55
View File
@@ -0,0 +1,55 @@
/*
(c) Copyright NetFoundry, Inc.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
https://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package xlink_transwarp
import (
"fmt"
"github.com/netfoundry/ziti-fabric/router/xlink"
"github.com/netfoundry/ziti-foundation/identity/identity"
)
func NewFactory(accepter xlink.Accepter) xlink.Factory {
return &factory{accepter: accepter}
}
func (self *factory) CreateListener(id *identity.TokenId, configData map[interface{}]interface{}) (xlink.Listener, error) {
config, err := loadListenerConfig(configData)
if err != nil {
return nil, fmt.Errorf("error loading listener configuration (%w)", err)
}
return &listener{
id: id,
config: config,
accepter: self.accepter,
}, nil
}
func (self *factory) CreateDialer(id *identity.TokenId, configData map[interface{}]interface{}) (xlink.Dialer, error) {
config, err := loadDialerConfig(configData)
if err != nil {
return nil, fmt.Errorf("error loading dialer configuration (%w)", err)
}
return &dialer{
id: id,
config: config,
accepter: self.accepter,
}, nil
}
type factory struct {
accepter xlink.Accepter
}
+44
View File
@@ -0,0 +1,44 @@
/*
(c) Copyright NetFoundry, Inc.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
https://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package xlink_transwarp
import (
"fmt"
"github.com/netfoundry/ziti-fabric/router/xlink"
"github.com/netfoundry/ziti-foundation/identity/identity"
"net"
)
func (self *listener) Listen() error {
listener, err := net.ListenUDP("udp", self.config.bindAddress)
if err != nil {
return fmt.Errorf("error listening (%w)", err)
}
self.listener = listener
return nil
}
func (self *listener) GetAdvertisement() string {
return self.config.advertiseAddress.String()
}
type listener struct {
id *identity.TokenId
config *listenerConfig
listener *net.UDPConn
accepter xlink.Accepter
}
+3 -3
View File
@@ -40,6 +40,6 @@ func (self *impl) Close() error {
}
type impl struct {
id *identity.TokenId
conn net.UDPConn
}
id *identity.TokenId
conn *net.UDPConn
}