gsadmin/ziti
1
0
Fork 0
mirror of https://github.com/openziti/ziti.git synced 2026-10-13 15:51:21 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request openziti/fabric#110 from openziti/transwarp_1

Browse Source 
transwarp_1
This commit is contained in:
Michael Quigley
2020-08-03 11:21:58 -04:00
committed by GitHub
parent 000ec1e235 74e07ebce6
commit f62f598f25
24 changed files with 71 additions and 1470 deletions
Download Patch File Download Diff File Expand all files Collapse all files
router/config.go
+25 -18
View File
@@ -82,6 +82,7 @@ type Config struct {
}
Dialers map[string]xgress.OptionsData
Listeners []listenerBinding
Transport map[interface{}]interface{}
src map[interface{}]interface{}
}
@@ -258,24 +259,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 +277,30 @@ 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
}
}
return cfg, nil
}
router/router.go
+3 -5
View File
@@ -33,7 +33,6 @@ import (
"github.com/openziti/fabric/router/xgress_transport_udp"
"github.com/openziti/fabric/router/xlink"
"github.com/openziti/fabric/router/xlink_transport"
"github.com/openziti/fabric/router/xlink_transwarp"
"github.com/openziti/foundation/channel2"
"github.com/openziti/foundation/metrics"
"github.com/openziti/foundation/profiler"
@@ -177,12 +176,11 @@ func (self *Router) registerComponents() error {
self.config.Forwarder,
self.metricsRegistry,
)
self.xlinkFactories["transport"] = xlink_transport.NewFactory(xlinkAccepter, xlinkChAccepter)
self.xlinkFactories["transwarp"] = xlink_transwarp.NewFactory(xlinkAccepter)
self.xlinkFactories["transport"] = xlink_transport.NewFactory(xlinkAccepter, xlinkChAccepter, self.config.Transport)
xgress.GlobalRegistry().Register("proxy", xgress_proxy.NewFactory(self.config.Id, self))
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))
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
router/xgress_proxy/factory.go
+5 -3
View File
@@ -20,10 +20,11 @@ import (
"fmt"
"github.com/openziti/fabric/router/xgress"
"github.com/openziti/foundation/identity/identity"
"github.com/openziti/foundation/transport"
)
func NewFactory(id *identity.TokenId, ctrl xgress.CtrlChannel) xgress.Factory {
return &factory{id: id, ctrl: ctrl}
func NewFactory(id *identity.TokenId, ctrl xgress.CtrlChannel, tcfg transport.Configuration) xgress.Factory {
return &factory{id: id, ctrl: ctrl, tcfg: tcfg}
}
func (factory *factory) CreateListener(optionsData xgress.OptionsData) (xgress.Listener, error) {
@@ -34,7 +35,7 @@ func (factory *factory) CreateListener(optionsData xgress.OptionsData) (xgress.L
} else {
return nil, fmt.Errorf("missing 'service' configuration option")
}
return newListener(factory.id, factory.ctrl, options, service), nil
return newListener(factory.id, factory.ctrl, options, factory.tcfg, service), nil
}
func (factory *factory) CreateDialer(optionsData xgress.OptionsData) (xgress.Dialer, error) {
@@ -45,4 +46,5 @@ type factory struct {
id *identity.TokenId
ctrl xgress.CtrlChannel
options *xgress.Options
tcfg transport.Configuration
}
router/xgress_proxy/listener.go
+4 -2
View File
@@ -24,11 +24,12 @@ import (
"github.com/openziti/foundation/transport"
)
func newListener(id *identity.TokenId, ctrl xgress.CtrlChannel, options *xgress.Options, service string) xgress.Listener {
func newListener(id *identity.TokenId, ctrl xgress.CtrlChannel, options *xgress.Options, tcfg transport.Configuration, service string) xgress.Listener {
return &listener{
id: id,
ctrl: ctrl,
options: options,
tcfg: tcfg,
service: service,
closeHelper: &xgress.CloseHelper{},
}
@@ -41,7 +42,7 @@ func (listener *listener) Listen(address string, bindHandler xgress.BindHandler)
}
incomingPeers := make(chan transport.Connection)
go listener.closeHelper.Init(txAddress.MustListen("tcp", listener.id, incomingPeers))
go listener.closeHelper.Init(txAddress.MustListen("tcp", listener.id, incomingPeers, listener.tcfg))
go func() {
for {
select {
@@ -73,6 +74,7 @@ type listener struct {
id *identity.TokenId
ctrl xgress.CtrlChannel
options *xgress.Options
tcfg transport.Configuration
service string
closeHelper *xgress.CloseHelper
}
router/xgress_proxy_udp/listener.go
+2 -2
View File
@@ -20,7 +20,7 @@ import (
"fmt"
"github.com/openziti/fabric/router/xgress"
"github.com/openziti/fabric/router/xgress_udp"
"github.com/openziti/foundation/transport/udp"
"github.com/openziti/foundation/util/info"
"github.com/sirupsen/logrus"
"net"
"time"
@@ -77,7 +77,7 @@ func (l *listener) relay() {
}()
for {
buf := make([]byte, udp.MaxPacketSize)
buf := make([]byte, info.MaxUdpPacketSize)
read, addr, err := l.conn.ReadFrom(buf)
if err != nil {
logrus.Errorf("error reading packet (%v)", err)
router/xgress_transport/client.go
+2 -2
View File
@@ -24,8 +24,8 @@ import (
)
// ClientDial dials the given xgress address and handles authentication, returning an authed connection or an error
func ClientDial(addr transport.Address, id *identity.TokenId, serviceId *identity.TokenId) (transport.Connection, error) {
peer, err := addr.Dial("i/"+id.Token, id)
func ClientDial(addr transport.Address, id *identity.TokenId, serviceId *identity.TokenId, tcfg transport.Configuration) (transport.Connection, error) {
peer, err := addr.Dial("i/"+id.Token, id, tcfg)
if err != nil {
return nil, err
}
router/xgress_transport/dialer.go
+4 -2
View File
@@ -28,17 +28,19 @@ type dialer struct {
id *identity.TokenId
ctrl xgress.CtrlChannel
options *xgress.Options
tcfg transport.Configuration
}
func (txd *dialer) IsTerminatorValid(string, string) bool {
return true
}
func newDialer(id *identity.TokenId, ctrl xgress.CtrlChannel, options *xgress.Options) (xgress.Dialer, error) {
func newDialer(id *identity.TokenId, ctrl xgress.CtrlChannel, options *xgress.Options, tcfg transport.Configuration) (xgress.Dialer, error) {
txd := &dialer{
id: id,
ctrl: ctrl,
options: options,
tcfg: tcfg,
}
return txd, nil
}
@@ -49,7 +51,7 @@ func (txd *dialer) Dial(destination string, sessionId *identity.TokenId, address
return nil, fmt.Errorf("cannot dial on invalid address [%s] (%s)", destination, err)
}
peer, err := txDestination.Dial("x/"+sessionId.Token, sessionId)
peer, err := txDestination.Dial("x/"+sessionId.Token, sessionId, txd.tcfg)
if err != nil {
return nil, err
}
router/xgress_transport/factory.go
+6 -4
View File
@@ -19,25 +19,27 @@ package xgress_transport
import (
"github.com/openziti/fabric/router/xgress"
"github.com/openziti/foundation/identity/identity"
"github.com/openziti/foundation/transport"
)
type factory struct {
id *identity.TokenId
ctrl xgress.CtrlChannel
options *xgress.Options
tcfg transport.Configuration
}
// NewFactory returns a new Transport Xgress factory
func NewFactory(id *identity.TokenId, ctrl xgress.CtrlChannel) xgress.Factory {
return &factory{id: id, ctrl: ctrl}
func NewFactory(id *identity.TokenId, ctrl xgress.CtrlChannel, tcfg transport.Configuration) xgress.Factory {
return &factory{id: id, ctrl: ctrl, tcfg: tcfg}
}
func (factory *factory) CreateListener(optionsData xgress.OptionsData) (xgress.Listener, error) {
options := xgress.LoadOptions(optionsData)
return newListener(factory.id, factory.ctrl, options), nil
return newListener(factory.id, factory.ctrl, options, factory.tcfg), nil
}
func (factory *factory) CreateDialer(optionsData xgress.OptionsData) (xgress.Dialer, error) {
options := xgress.LoadOptions(optionsData)
return newDialer(factory.id, factory.ctrl, options)
return newDialer(factory.id, factory.ctrl, options, factory.tcfg)
}
router/xgress_transport/listener.go
+4 -2
View File
@@ -28,14 +28,16 @@ type listener struct {
id *identity.TokenId
ctrl xgress.CtrlChannel
options *xgress.Options
tcfg transport.Configuration
closeHelper *xgress.CloseHelper
}
func newListener(id *identity.TokenId, ctrl xgress.CtrlChannel, options *xgress.Options) xgress.Listener {
func newListener(id *identity.TokenId, ctrl xgress.CtrlChannel, options *xgress.Options, tcfg transport.Configuration) xgress.Listener {
return &listener{
id: id,
ctrl: ctrl,
options: options,
tcfg: tcfg,
closeHelper: &xgress.CloseHelper{},
}
}
@@ -47,7 +49,7 @@ func (listener *listener) Listen(address string, bindHandler xgress.BindHandler)
}
incomingPeers := make(chan transport.Connection)
go listener.closeHelper.Init(txAddress.MustListen("tcp", listener.id, incomingPeers))
go listener.closeHelper.Init(txAddress.MustListen("tcp", listener.id, incomingPeers, listener.tcfg))
go func() {
for {
select {
router/xgress_transport_udp/connection.go
+2 -2
View File
@@ -18,7 +18,7 @@ package xgress_transport_udp
import (
"github.com/openziti/fabric/router/xgress"
"github.com/openziti/foundation/transport/udp"
"github.com/openziti/foundation/util/info"
"net"
)
@@ -27,7 +27,7 @@ func (p *packetConn) LogContext() string {
}
func (p *packetConn) ReadPayload() ([]byte, map[uint8][]byte, error) {
buffer := make([]byte, udp.MaxPacketSize)
buffer := make([]byte, info.MaxUdpPacketSize)
n, err := p.Read(buffer)
if err != nil {
return nil, nil, err
router/xgress_transport_udp/listener.go
+2 -2
View File
@@ -22,7 +22,7 @@ import (
"github.com/openziti/fabric/router/xgress"
"github.com/openziti/fabric/router/xgress_udp"
"github.com/openziti/foundation/identity/identity"
"github.com/openziti/foundation/transport/udp"
"github.com/openziti/foundation/util/info"
"net"
"time"
)
@@ -88,7 +88,7 @@ func (l *listener) relayIncomingPackets() {
}()
for {
buf := make([]byte, udp.MaxPacketSize)
buf := make([]byte, info.MaxUdpPacketSize)
logger.Debugf("Trying to read next packet")
bytesRead, addr, err := l.conn.ReadFrom(buf)
logger.Debugf("Packet read complete: %v bytes read", bytesRead)
router/xlink_transport/dialer.go
+2 -1
View File
@@ -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.NewChannelWithTransportConfiguration("l/"+linkId.Token, dialer, self.config.options, self.tcfg)
if err == nil {
xlink := &impl{id: linkId, ch: ch}
@@ -59,4 +59,5 @@ type dialer struct {
config *dialerConfig
accepter xlink.Accepter
chAccepter ChannelAccepter
tcfg transport.Configuration
}
router/xlink_transport/factory.go
+6 -2
View File
@@ -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, tcfg: 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,
tcfg: self.tcfg,
}, nil
}
@@ -49,10 +51,12 @@ func (self *factory) CreateDialer(id *identity.TokenId, _ xlink.Forwarder, confi
config: config,
accepter: self.accepter,
chAccepter: self.chAccepter,
tcfg: self.tcfg,
}, nil
}
type factory struct {
accepter xlink.Accepter
chAccepter ChannelAccepter
tcfg transport.Configuration
}
router/xlink_transport/listener.go
+4 -2
View File
@@ -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.NewClassicListenerWithTransportConfiguration(self.id, self.config.bind, self.config.options.ConnectOptions, self.tcfg)
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.NewChannelWithTransportConfiguration("link", self.listener, self.config.options, self.tcfg)
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
tcfg transport.Configuration
}
router/xlink_transwarp/buffer.go
-111
View File
@@ -1,111 +0,0 @@
/*
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 (
"github.com/sirupsen/logrus"
"sync"
"time"
)
func (self *transmitBuffer) ack(gaps int) {
self.cond.L.Lock()
if gaps > 0 {
if self.windowSize > 10 {
self.windowSize /= 2
}
}
self.windowContains = 0
self.cond.L.Unlock()
}
func (self *transmitBuffer) accept(m *message) {
self.cond.L.Lock()
for self.windowContains >= self.windowSize {
self.cond.Wait()
}
self.highWater = m.sequence
self.windowContains++
self.cond.L.Unlock()
}
func newTransmitBuffer() *transmitBuffer {
txb := &transmitBuffer{
windowSize: startingWindowSize,
}
txb.lock = new(sync.Mutex)
txb.cond = sync.NewCond(txb.lock)
return txb
}
type transmitBuffer struct {
windowContains int
windowSize int
highWater int32
lock *sync.Mutex
cond *sync.Cond
}
func (self *receiveBuffer) receive(m *message) {
self.lock.Lock()
if m.sequence != self.highWater+1 {
self.gaps++
}
if m.sequence < self.lowWater {
self.lowWater = m.sequence
}
if m.sequence > self.highWater {
self.highWater = m.sequence
}
self.lock.Unlock()
}
func (self *receiveBuffer) acker() {
for {
time.Sleep(1 * time.Second)
self.lock.Lock()
if time.Since(self.lastReport).Milliseconds() > 1000 {
// send report
logrus.Infof("sending transwarp window report")
self.lastReport = time.Now()
}
}
}
func newReceiveBuffer(xlinkImpl *impl) *receiveBuffer {
rxb := &receiveBuffer{
lastReport: time.Now(),
windowSize: startingWindowSize,
xlinkImpl: xlinkImpl,
}
rxb.lock = new(sync.Mutex)
return rxb
}
type receiveBuffer struct {
lowWater int32
highWater int32
count int32
gaps int32
windowSize int
lastReport time.Time
xlinkImpl *impl
lock *sync.Mutex
}
const startingWindowSize = 10
router/xlink_transwarp/config.go
-85
View File
@@ -1,85 +0,0 @@
/*
(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"
"net"
"reflect"
)
func loadListenerConfig(data map[interface{}]interface{}) (*listenerConfig, error) {
config := &listenerConfig{}
if value, found := data["bind"]; found {
if addressString, ok := value.(string); ok {
if address, err := net.ResolveUDPAddr("udp", addressString); err == nil {
config.bindAddress = address
config.advertiseAddress = address
} else {
return nil, fmt.Errorf("error parsing 'bind' address (%w)", err)
}
} else {
return nil, fmt.Errorf("invalid 'bind' address in listener config (%s)", reflect.TypeOf(value))
}
} else {
return nil, fmt.Errorf("missing 'bind' address in listener config")
}
if value, found := data["advertise"]; found {
if addressString, ok := value.(string); ok {
if address, err := net.ResolveUDPAddr("udp", addressString); err == nil {
config.advertiseAddress = address
} else {
return nil, fmt.Errorf("error parsing 'advertise' address in listener config (%w)", err)
}
} else {
return nil, fmt.Errorf("invalid 'advertise' address in listener config (%s)", reflect.TypeOf(value))
}
}
return config, nil
}
type listenerConfig struct {
bindAddress *net.UDPAddr
advertiseAddress *net.UDPAddr
}
func loadDialerConfig(data map[interface{}]interface{}) (*dialerConfig, error) {
config := &dialerConfig{}
if value, found := data["bind"]; found {
if addressString, ok := value.(string); ok {
if address, err := net.ResolveUDPAddr("udp", addressString); err == nil {
config.bindAddress = address
} else {
return nil, fmt.Errorf("error parsing 'bind' address in dialer config (%w)", err)
}
} else {
return nil, fmt.Errorf("invalid 'bind' address in dialer config (%s)", reflect.TypeOf(value))
}
} else {
return nil, fmt.Errorf("missing 'bind' address in dialer config")
}
return config, nil
}
type dialerConfig struct {
bindAddress *net.UDPAddr
}
router/xlink_transwarp/dialer.go
-96
View File
@@ -1,96 +0,0 @@
/*
(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/openziti/fabric/router/xlink"
"github.com/openziti/foundation/identity/identity"
"github.com/sirupsen/logrus"
"net"
"time"
)
/*
* xlink.Dialer
*/
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] at [%s]", name, address)
if conn, err := net.ListenUDP("udp", self.config.bindAddress); err == nil {
waitCh := make(chan struct{}, 0)
self.waiters[linkId.Token] = waitCh
if err := writeHello(linkId, conn, address); err == nil {
if m, peer, err := readMessage(conn); err == nil {
if err := handleHello(m, conn, peer, self); err == nil {
select {
case <-waitCh:
xlinkImpl := newImpl(linkId, conn, address, self.forwarder)
if err := self.accepter.Accept(xlinkImpl); err != nil {
return fmt.Errorf("error accepting outgoing Xlink (%w)", err)
}
go xlinkImpl.listener()
go xlinkImpl.pinger()
return nil
case <-time.After(5 * time.Second):
delete(self.waiters, linkId.Token)
return fmt.Errorf("timeout in hello response")
}
} else {
return fmt.Errorf("error handling hello from [%s] (%w)", peer, err)
}
} else {
return fmt.Errorf("error receiving hello from peer [%s] (%w)", peer, err)
}
} else {
return fmt.Errorf("error sending hello to peer from [%s] (%w)", address, err)
}
} else {
return fmt.Errorf("error dialing link [%s] (%w)", name, err)
}
} else {
return fmt.Errorf("error parsing link address [%s] (%w)", addressString, err)
}
}
/*
* xlink_transwarp.HelloHandler
*/
func (self *dialer) HandleHello(linkId *identity.TokenId, _ *net.UDPConn, peer *net.UDPAddr) {
if ch, found := self.waiters[linkId.Token]; found {
logrus.Infof("[hello->l/%s] <- %s", linkId.Token, peer)
close(ch)
} else {
logrus.Errorf("invalid hello for link [%s] from peer [%s], failure", linkId.Token, peer)
}
}
/*
* xlink_transwarp.dialer
*/
type dialer struct {
id *identity.TokenId
config *dialerConfig
accepter xlink.Accepter
forwarder xlink.Forwarder
waiters map[string]chan struct{}
}
router/xlink_transwarp/factory.go
-59
View File
@@ -1,59 +0,0 @@
/*
(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/openziti/fabric/router/xlink"
"github.com/openziti/foundation/identity/identity"
)
func NewFactory(accepter xlink.Accepter) xlink.Factory {
return &factory{accepter: accepter}
}
func (self *factory) CreateListener(id *identity.TokenId, f xlink.Forwarder, 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,
forwarder: f,
peers: make(map[string]*impl),
}, nil
}
func (self *factory) CreateDialer(id *identity.TokenId, f xlink.Forwarder, 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,
forwarder: f,
waiters: make(map[string]chan struct{}),
}, nil
}
type factory struct {
accepter xlink.Accepter
}
router/xlink_transwarp/listener.go
-99
View File
@@ -1,99 +0,0 @@
/*
(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/openziti/fabric/router/xlink"
"github.com/openziti/foundation/identity/identity"
"github.com/openziti/foundation/util/concurrenz"
"github.com/sirupsen/logrus"
"net"
)
/*
* xlink.Listener
*/
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
go self.listen()
return nil
}
func (self *listener) GetAdvertisement() string {
return self.config.advertiseAddress.String()
}
/*
* xlink_transwarp.HelloHandler
*/
func (self *listener) HandleHello(linkId *identity.TokenId, conn *net.UDPConn, peer *net.UDPAddr) {
xlinkImpl := newImpl(linkId, conn, peer, self.forwarder)
if err := self.accepter.Accept(xlinkImpl); err == nil {
self.peers[peer.String()] = xlinkImpl
if err := writeHello(linkId, self.listener, peer); err == nil {
go xlinkImpl.pinger()
logrus.Infof("[hello->l/%s] -> %s", linkId.Token, peer)
} else {
logrus.Errorf("error sending hello for [l/%s] to peer at [%s] (%v)", linkId.Token, peer, err)
}
}
}
func (self *listener) Close() error {
defer self.closed.Set(true)
return self.listener.Close()
}
/*
* xlink_transwarp.listener
*/
func (self *listener) listen() {
for !self.closed.Get() {
if m, peer, err := readMessage(self.listener); err == nil {
if m.messageType == Hello {
if err := handleHello(m, self.listener, peer, self); err != nil {
logrus.Errorf("error handling hello from [%s] (%v)", peer, err)
}
} else {
if xlinkImpl, found := self.peers[peer.String()]; found {
if err := handleMessage(m, self.listener, peer, xlinkImpl); err != nil {
logrus.Errorf("error handling message from [%s] (%v)", peer, err)
}
}
}
} else {
if nerr, ok := err.(net.Error); ok && !nerr.Timeout() {
logrus.Errorf("error reading message (%v)", err)
}
}
}
}
type listener struct {
id *identity.TokenId
config *listenerConfig
listener *net.UDPConn
accepter xlink.Accepter
forwarder xlink.Forwarder
peers map[string]*impl
closed concurrenz.AtomicBoolean
}
router/xlink_transwarp/message.go
-465
View File
@@ -1,465 +0,0 @@
/*
(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 (
"bytes"
"encoding/binary"
"fmt"
"github.com/openziti/fabric/router/xgress"
"github.com/openziti/foundation/identity/identity"
"github.com/openziti/foundation/transport/udp"
"net"
"time"
)
type HelloHandler interface {
HandleHello(linkId *identity.TokenId, conn *net.UDPConn, addr *net.UDPAddr)
}
type MessageHandler interface {
HandlePing(sequence int32, replyFor int32, conn *net.UDPConn, addr *net.UDPAddr)
HandlePayload(p *xgress.Payload, sequence int32, conn *net.UDPConn, addr *net.UDPAddr)
HandleAcknowledgement(a *xgress.Acknowledgement, sequence int32, conn *net.UDPConn, addr *net.UDPAddr)
HandleWindowReport(lowWater, highWater, gaps, count int32, conn *net.UDPConn, addr *net.UDPAddr)
HandleWindowSizeRequest(newWindowSize int32, conn *net.UDPConn, addr *net.UDPAddr)
}
/**
* TRANSWARP v1 Wire Format
*
* // --- message section --------------------------------------------------------------------------------- //
*
* <version:[]byte> 0 1 2 3
* <sequence:int32> 4 5 6 7
* <fragment:uint8> 8
* <of_fragments:uint8> 9
* <content_type:uint8> 10
* <headers_length:uint16> 11 12
* <payload_length:uint16> 13 14
*
* // --- data section ------------------------------------------------------------------------------------ //
*
* <headers> 15 -> (15 + headers_length)
* <body> (15 + headers_length) -> (15 + headers_length + body_length)
*/
var magicV1 = []byte{0x01, 0x02, 0x02, 0x00}
const messageSectionLength = 15
type messageType uint8
const (
Hello messageType = iota
Ping
Payload
Acknowledgement
WindowReport
WindowSizeRequest
)
const timeoutSeconds = 5
const mss = 1472
const noReplyFor = -1
func writeHello(linkId *identity.TokenId, conn *net.UDPConn, peer *net.UDPAddr) error {
payload := new(bytes.Buffer)
payload.Write([]byte(linkId.Token))
data, err := encodeMessage(&message{
sequence: -1,
fragment: 0,
ofFragments: 1,
messageType: Hello,
payload: payload.Bytes(),
})
if err != nil {
return fmt.Errorf("error creating message (%w)", err)
}
if err := conn.SetWriteDeadline(time.Now().Add(timeoutSeconds * time.Second)); err != nil {
return fmt.Errorf("unable to set write deadline (%w)", err)
}
if _, err := conn.WriteToUDP(data, peer); err != nil {
return err
}
return nil
}
func writePing(sequence int32, conn *net.UDPConn, peer *net.UDPAddr, replyFor int32) error {
payload := new(bytes.Buffer)
if err := binary.Write(payload, binary.LittleEndian, replyFor); err != nil {
return fmt.Errorf("reply for write (%w)", err)
}
data, err := encodeMessage(&message{
sequence: sequence,
fragment: 0,
ofFragments: 1,
messageType: Ping,
payload: payload.Bytes(),
})
if err != nil {
return fmt.Errorf("error creating message (%w)", err)
}
if err := conn.SetWriteDeadline(time.Now().Add(timeoutSeconds * time.Second)); err != nil {
return fmt.Errorf("unable to set write deadline (%w)", err)
}
if _, err := conn.WriteToUDP(data, peer); err != nil {
return err
}
return nil
}
func writePayload(sequence int32, p *xgress.Payload, conn *net.UDPConn, peer *net.UDPAddr) error {
m, err := encodePayload(p, sequence)
if err != nil {
return err
}
data, err := encodeMessage(m)
if err != nil {
return fmt.Errorf("error creating message (%w)", err)
}
if err := conn.SetWriteDeadline(time.Now().Add(timeoutSeconds * time.Second)); err != nil {
return fmt.Errorf("unable to set write deadline (%w)", err)
}
if _, err := conn.WriteToUDP(data, peer); err != nil {
return err
}
return nil
}
func writeAcknowledgement(sequence int32, a *xgress.Acknowledgement, conn *net.UDPConn, peer *net.UDPAddr) error {
m, err := encodeAcnowledgement(a, sequence)
if err != nil {
return err
}
data, err := encodeMessage(m)
if err != nil {
return fmt.Errorf("error creating message (%w)", err)
}
if err := conn.SetWriteDeadline(time.Now().Add(timeoutSeconds * time.Second)); err != nil {
return fmt.Errorf("unable to set write deadline (%w)", err)
}
if _, err := conn.WriteToUDP(data, peer); err != nil {
return err
}
return nil
}
func writeWindowReport(sequence int32, lowWater, highWater, gaps, count int32, conn *net.UDPConn, peer *net.UDPAddr) error {
m, err := encodeWindowReport(sequence, lowWater, highWater, gaps, count)
if err != nil {
return err
}
data, err := encodeMessage(m)
if err != nil {
return fmt.Errorf("error creating message (%w)", err)
}
if err := conn.SetWriteDeadline(time.Now().Add(timeoutSeconds * time.Second)); err != nil {
return fmt.Errorf("unable to set write deadline (%w)", err)
}
if _, err := conn.WriteToUDP(data, peer); err != nil {
return err
}
return nil
}
func writeWindowSizeRequest(sequence, newWindowSize int32, conn *net.UDPConn, peer *net.UDPAddr) error {
m, err := encodeWindowSizeRequest(sequence, newWindowSize)
if err != nil {
return err
}
data, err := encodeMessage(m)
if err != nil {
return fmt.Errorf("error creating message (%w)", err)
}
if err := conn.SetWriteDeadline(time.Now().Add(timeoutSeconds * time.Second)); err != nil {
return fmt.Errorf("unable to set write deadline (%w)", err)
}
if _, err := conn.WriteToUDP(data, peer); err != nil {
return err
}
return nil
}
func readMessage(conn *net.UDPConn) (*message, *net.UDPAddr, error) {
data := make([]byte, udp.MaxPacketSize)
if err := conn.SetReadDeadline(time.Now().Add(5 * time.Second)); err != nil {
return nil, nil, fmt.Errorf("error setting read deadline (%w)", err)
}
if n, peer, err := conn.ReadFromUDP(data); err == nil {
if m, err := decodeMessage(data[:n]); err == nil {
return m, peer, nil
} else {
return nil, nil, fmt.Errorf("error decoding message from [%s] (%w)", peer, err)
}
} else {
return nil, nil, err
}
}
func handleHello(m *message, conn *net.UDPConn, peer *net.UDPAddr, handler HelloHandler) error {
if m != nil {
switch m.messageType {
case Hello:
if m.sequence != -1 {
return fmt.Errorf("hello expects sequence -1 [%s]", peer)
}
if m.fragment != 0 || m.ofFragments != 1 {
return fmt.Errorf("hello expects single fragment [%s]", peer)
}
linkId := &identity.TokenId{Token: string(m.payload)}
handler.HandleHello(linkId, conn, peer)
return nil
default:
return fmt.Errorf("expected hello, not [%d] from [%s]", m.messageType, peer)
}
} else {
return fmt.Errorf("nil message")
}
}
func handleMessage(m *message, conn *net.UDPConn, peer *net.UDPAddr, handler MessageHandler) error {
if m.fragment != 0 || m.ofFragments != 1 {
return fmt.Errorf("ping expects single fragment [%s]", peer)
}
switch m.messageType {
case Ping:
replyFor, err := readInt32(m.payload)
if err != nil {
return fmt.Errorf("ping expects replyFor in payload [%s] (%w)", peer, err)
}
handler.HandlePing(m.sequence, replyFor, conn, peer)
return nil
case Payload:
p, err := decodePayload(m)
if err != nil {
return fmt.Errorf("error decoding payload for peer [%s] (%w)", peer, err)
}
handler.HandlePayload(p, m.sequence, conn, peer)
return nil
case Acknowledgement:
a, err := decodeAcknowledgement(m)
if err != nil {
return fmt.Errorf("error decoding acknowledgement for peer [%s] (%w)", peer, err)
}
handler.HandleAcknowledgement(a, m.sequence, conn, peer)
return nil
case WindowReport:
lowWater, highWater, gaps, count, err := decodeWindowReport(m)
if err != nil {
return fmt.Errorf("error decoding window report for peer [%s] (%w)", peer, err)
}
handler.HandleWindowReport(lowWater, highWater, gaps, count, conn, peer)
return nil
case WindowSizeRequest:
newWindowSize, err := decodeWindowSizeRequest(m)
if err != nil {
return fmt.Errorf("error decoding window size request for peer [%s] (%w)", peer, err)
}
handler.HandleWindowSizeRequest(newWindowSize, conn, peer)
return nil
default:
return fmt.Errorf("unexpected message type [%d] from [%s]", m.messageType, peer)
}
}
func encodeMessage(m *message) ([]byte, error) {
data := new(bytes.Buffer)
data.Write(magicV1)
if err := binary.Write(data, binary.LittleEndian, m.sequence); err != nil {
return nil, fmt.Errorf("sequence write (%w)", err)
}
data.Write([]byte{m.fragment, m.ofFragments, uint8(m.messageType)})
if err := binary.Write(data, binary.LittleEndian, uint16(0)); err != nil { // headers length
return nil, fmt.Errorf("headers length write (%w)", err)
}
if err := binary.Write(data, binary.LittleEndian, uint16(len(m.payload))); err != nil {
return nil, fmt.Errorf("payload length write (%w)", err)
}
data.Write(m.payload)
buffer := make([]byte, data.Len())
_, err := data.Read(buffer)
if err != nil {
return nil, fmt.Errorf("error reading buffer (%w)", err)
}
/*
if n > mss {
return nil, fmt.Errorf("message too long [%d]", n)
}
*/
return buffer, nil
}
/*
* TRANSWARP v1 Headers Wire Format
*
* <key:uint8> 0
* <length:uint8> 1
* <data> 2 -> (2 + length)
*/
func encodeHeaders(headers map[uint8][]byte) ([]byte, error) {
data := new(bytes.Buffer)
for k, v := range headers {
if _, err := data.Write([]byte{k}); err != nil {
return nil, err
}
if err := binary.Write(data, binary.LittleEndian, uint8(len(v))); err != nil {
return nil, err
}
if n, err := data.Write(v); err == nil {
if n != len(v) {
return nil, fmt.Errorf("short header write")
}
} else {
return nil, err
}
}
return data.Bytes(), nil
}
func decodeHeaders(data []byte) (map[uint8][]byte, error) {
headers := make(map[uint8][]byte)
if len(data) > 0 && len(data) < 2 {
return nil, fmt.Errorf("truncated header data")
}
i := 0
for i < len(data) {
key := data[i]
length := data[i+1]
if i+2+int(length) > len(data) {
return nil, fmt.Errorf("short header data (%d > %d)", i+2+int(length), len(data))
}
headerData := data[i+2 : i+2+int(length)]
headers[key] = headerData
i += 2 + int(length)
}
return headers, nil
}
func decodeMessage(data []byte) (*message, error) {
m := &message{}
if len(data) < messageSectionLength {
return nil, fmt.Errorf("short read")
}
for i := 0; i < len(magicV1); i++ {
if data[i] != magicV1[i] {
return nil, fmt.Errorf("bad magic")
}
}
sequence, err := readInt32(data[4:8])
if err != nil {
return nil, fmt.Errorf("error reading sequence (%w)", err)
}
m.sequence = sequence
m.fragment = data[8]
m.ofFragments = data[9]
m.messageType = messageType(data[10])
headersLength, err := readUint16(data[11:13])
if err != nil {
return nil, fmt.Errorf("error reading headers length (%w)", err)
}
if headers, err := decodeHeaders(data[15 : 15+headersLength]); err == nil {
m.headers = headers
} else {
return nil, fmt.Errorf("headers error (%w)", err)
}
payloadLength, err := readUint16(data[13:15])
if err != nil {
return nil, fmt.Errorf("error reading payload length (%w)", err)
}
m.payload = data[15+headersLength : 15+headersLength+payloadLength]
return m, nil
}
func readInt32(data []byte) (ret int32, err error) {
buf := bytes.NewBuffer(data)
err = binary.Read(buf, binary.LittleEndian, &ret)
return
}
func writeInt32(value int32) ([]byte, error) {
buf := new(bytes.Buffer)
err := binary.Write(buf, binary.LittleEndian, value)
if err != nil {
return nil, err
}
return buf.Bytes(), nil
}
func readUint32(data []byte) (ret uint32, err error) {
buf := bytes.NewBuffer(data)
err = binary.Read(buf, binary.LittleEndian, &ret)
return
}
func readUint16(data []byte) (ret uint16, err error) {
buf := bytes.NewBuffer(data)
err = binary.Read(buf, binary.LittleEndian, &ret)
return
}
type message struct {
sequence int32
fragment uint8
ofFragments uint8
messageType messageType
headers map[uint8][]byte
payload []byte
}
router/xlink_transwarp/message_acknowledgement.go
-99
View File
@@ -1,99 +0,0 @@
/*
(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 (
"bytes"
"encoding/binary"
"github.com/openziti/fabric/router/xgress"
)
/*
* TRANSWARP v1 Acknowledgement Format
*
* <session_id_length:int32> 0 1 2 3
* <flags:uint32> 4 5 6 7
* <sequence_ids_count:int32> 8 9 10 11
* <session_id> 12 -> (12 + session_id_length)
* <sequence_ids:[]int32> (12 + session_id_length) -> ((12 + session_id_length) + (4 * sequence_ids_count))
*/
func encodeAcnowledgement(a *xgress.Acknowledgement, sequence int32) (m *message, err error) {
payload := new(bytes.Buffer)
if err := binary.Write(payload, binary.LittleEndian, int32(len(a.SessionId))); err != nil {
return nil, err
}
if err := binary.Write(payload, binary.LittleEndian, a.Flags); err != nil {
return nil, err
}
if err := binary.Write(payload, binary.LittleEndian, int32(len(a.Sequence))); err != nil {
return nil, err
}
if _, err := payload.Write([]byte(a.SessionId)); err != nil {
return nil, err
}
for _, sequenceId := range a.Sequence {
if err := binary.Write(payload, binary.LittleEndian, sequenceId); err != nil {
return nil, err
}
}
m = &message{
sequence: sequence,
fragment: 0,
ofFragments: 1,
messageType: Acknowledgement,
payload: payload.Bytes(),
}
return
}
func decodeAcknowledgement(m *message) (a *xgress.Acknowledgement, err error) {
sessionIdLength, err := readInt32(m.payload[0:4])
if err != nil {
return nil, err
}
flags, err := readUint32(m.payload[4:8])
if err != nil {
return nil, err
}
sequenceIdsCount, err := readInt32(m.payload[8:12])
if err != nil {
return nil, err
}
sessionId := m.payload[12 : 12+sessionIdLength]
nextSequenceId := 12 + sessionIdLength
var sequenceIds []int32
for i := 0; i < int(sequenceIdsCount); i++ {
sequenceId, err := readInt32(m.payload[nextSequenceId : nextSequenceId+4])
if err != nil {
return nil, err
}
sequenceIds = append(sequenceIds, sequenceId)
nextSequenceId += 4
}
a = &xgress.Acknowledgement{
Header: xgress.Header{
SessionId: string(sessionId),
Flags: flags,
},
Sequence: sequenceIds,
}
return
}
router/xlink_transwarp/message_payload.go
-99
View File
@@ -1,99 +0,0 @@
/*
(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 (
"bytes"
"encoding/binary"
"github.com/openziti/fabric/router/xgress"
)
/*
* TRANSWARP v1 Payload Format
*
* <session_id_length:int32> 0 1 2 3
* <flags:uint32> 4 5 6 7
* <sequence:int32> 8 9 10 11
* <data_length:int32> 12 13 14 15
* <session_id> 16 -> (16 + session_id_length)
* <data> (16 + session_id_length) -> (16 + session_id_length + data_length)
*/
func encodePayload(p *xgress.Payload, sequence int32) (m *message, err error) {
payload := new(bytes.Buffer)
if err := binary.Write(payload, binary.LittleEndian, int32(len(p.SessionId))); err != nil {
return nil, err
}
if err := binary.Write(payload, binary.LittleEndian, p.Flags); err != nil {
return nil, err
}
if err := binary.Write(payload, binary.LittleEndian, p.Sequence); err != nil {
return nil, err
}
if err := binary.Write(payload, binary.LittleEndian, int32(len(p.Data))); err != nil {
return nil, err
}
if _, err := payload.Write([]byte(p.SessionId)); err != nil {
return nil, err
}
if _, err := payload.Write(p.Data); err != nil {
return nil, err
}
m = &message{
sequence: sequence,
fragment: 0,
ofFragments: 1,
messageType: Payload,
headers: p.Headers,
payload: payload.Bytes(),
}
return
}
func decodePayload(m *message) (p *xgress.Payload, err error) {
sessionIdLength, err := readInt32(m.payload[0:4])
if err != nil {
return nil, err
}
flags, err := readUint32(m.payload[4:8])
if err != nil {
return nil, err
}
sequence, err := readInt32(m.payload[8:12])
if err != nil {
return nil, err
}
dataLength, err := readInt32(m.payload[12:16])
if err != nil {
return nil, err
}
sessionId := m.payload[16 : 16+sessionIdLength]
data := m.payload[16+sessionIdLength : 16+sessionIdLength+dataLength]
p = &xgress.Payload{
Header: xgress.Header{
SessionId: string(sessionId),
Flags: flags,
},
Sequence: sequence,
Headers: m.headers,
Data: data,
}
return p, nil
}
router/xlink_transwarp/message_window.go
-129
View File
@@ -1,129 +0,0 @@
/*
(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 (
"bytes"
"encoding/binary"
"fmt"
)
func encodeWindowReport(sequence, lowWater, highWater, gaps, count int32) (*message, error) {
headers := make(map[uint8][]byte)
if data, err := writeInt32(lowWater); err == nil {
headers[0] = data
} else {
return nil, err
}
if data, err := writeInt32(highWater); err == nil {
headers[1] = data
} else {
return nil, err
}
if data, err := writeInt32(gaps); err != nil {
headers[2] = data
} else {
return nil, err
}
if data, err := writeInt32(count); err != nil {
headers[3] = data
} else {
return nil, err
}
return &message{
sequence: sequence,
fragment: 0,
ofFragments: 1,
messageType: WindowReport,
headers: headers,
}, nil
}
func decodeWindowReport(m *message) (lowWater, highWater, gaps, count int32, err error) {
if m.headers != nil {
if value, found := m.headers[0]; found {
i32v, err := readInt32(value)
if err == nil {
lowWater = i32v
} else {
return 0, 0, 0, 0, fmt.Errorf("malformed lowWater (%w)", err)
}
} else {
return 0, 0, 0, 0, fmt.Errorf("missing lowWater")
}
if value, found := m.headers[1]; found {
i32v, err := readInt32(value)
if err == nil {
highWater = i32v
} else {
return 0, 0, 0, 0, fmt.Errorf("malformed highWater (%w)", err)
}
} else {
return 0, 0, 0, 0, fmt.Errorf("missing highWater")
}
if value, found := m.headers[2]; found {
i32v, err := readInt32(value)
if err == nil {
gaps = i32v
} else {
return 0, 0, 0, 0, fmt.Errorf("malformed gaps (%w)", err)
}
} else {
return 0, 0, 0, 0, fmt.Errorf("missing gaps")
}
if value, found := m.headers[3]; found {
i32v, err := readInt32(value)
if err == nil {
count = i32v
} else {
return 0, 0, 0, 0, fmt.Errorf("malformed count (%w)", err)
}
} else {
return 0, 0, 0, 0, fmt.Errorf("missing count")
}
return
} else {
return 0, 0, 0, 0, fmt.Errorf("missing headers")
}
}
func encodeWindowSizeRequest(sequence, newSize int32) (*message, error) {
payload := new(bytes.Buffer)
if err := binary.Write(payload, binary.LittleEndian, newSize); err != nil {
return nil, err
}
m := &message{
sequence: sequence,
fragment: 0,
ofFragments: 1,
messageType: WindowSizeRequest,
payload: payload.Bytes(),
}
return m, nil
}
func decodeWindowSizeRequest(m *message) (int32, error) {
newWindowSize, err := readInt32(m.payload[0:4])
if err != nil {
return 0, err
}
return newWindowSize, nil
}
router/xlink_transwarp/xlink.go
-179
View File
@@ -1,179 +0,0 @@
/*
(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/openziti/fabric/router/xgress"
"github.com/openziti/fabric/router/xlink"
"github.com/openziti/foundation/identity/identity"
"github.com/sirupsen/logrus"
"net"
"sync"
"time"
)
/*
* xlink.Xlink
*/
func (self *impl) Id() *identity.TokenId {
return self.id
}
func (self *impl) SendPayload(payload *xgress.Payload) error {
return writePayload(self.nextSequence(), payload, self.conn, self.peer)
}
func (self *impl) SendAcknowledgement(acknowledgement *xgress.Acknowledgement) error {
return writeAcknowledgement(self.nextSequence(), acknowledgement, self.conn, self.peer)
}
func (self *impl) Close() error {
return self.conn.Close()
}
/*
* xlink_transwarp.MessageHandler
*/
func (self *impl) HandlePing(sequence int32, replyFor int32, conn *net.UDPConn, addr *net.UDPAddr) {
if replyFor == -1 {
if err := self.sendPingReply(sequence); err != nil {
logrus.Errorf("error sending ping (%v)", err)
}
} else {
self.receivePing(replyFor)
}
}
func (self *impl) HandlePayload(p *xgress.Payload, sequence int32, conn *net.UDPConn, addr *net.UDPAddr) {
if err := self.forwarder.ForwardPayload(xgress.Address(self.id.Token), p); err != nil {
logrus.Errorf("[l/%s] => error forwarding payload (%v)", self.id.Token, err)
}
}
func (self *impl) HandleAcknowledgement(a *xgress.Acknowledgement, sequence int32, conn *net.UDPConn, addr *net.UDPAddr) {
if err := self.forwarder.ForwardAcknowledgement(xgress.Address(self.id.Token), a); err != nil {
logrus.Errorf("[l/%s] => error forwarding acknowledgement (%v)", self.id.Token, err)
}
}
func (self *impl) HandleWindowReport(lowWater, highWater, gaps, count int32, conn *net.UDPConn, peer *net.UDPAddr) {
logrus.Errorf("window report not connected")
}
func (self *impl) HandleWindowSizeRequest(newWindowSize int32, conn *net.UDPConn, peer *net.UDPAddr) {
logrus.Errorf("window size request not connected")
}
/*
* impl
*/
func (self *impl) listener() {
for {
if m, peer, err := readMessage(self.conn); err == nil {
if err := handleMessage(m, self.conn, peer, self); err != nil {
logrus.Errorf("error handling message from [%s] (%v)", peer, err)
}
}
}
}
func (self *impl) pinger() {
for {
time.Sleep(pingCycleDelayMs * time.Millisecond)
if time.Since(self.lastPingTx).Milliseconds() >= pingDelayMs {
if err := self.sendPingRequest(); err != nil {
logrus.Errorf("error sending ping request (%v)", err)
}
}
logrus.Debugf("time since last ping receipt [%d ms.]", time.Since(self.lastPingRx).Milliseconds())
}
}
func (self *impl) sendPingRequest() error {
sequence := self.nextSequence()
if err := writePing(sequence, self.conn, self.peer, noReplyFor); err == nil {
self.lastPingTxSequence = sequence
self.lastPingTx = time.Now()
logrus.Infof("[ping:%d] => %s", sequence, self.peer)
return nil
} else {
return fmt.Errorf("error sending ping (%w)", err)
}
}
func (self *impl) sendPingReply(forSequence int32) error {
sequence := self.nextSequence()
if err := writePing(sequence, self.conn, self.peer, forSequence); err == nil {
logrus.Infof("[ping:%d] <= %s", forSequence, self.peer)
return nil
} else {
return fmt.Errorf("error sending ping reply to [%s] (%w)", self.peer, err)
}
}
func (self *impl) receivePing(replyFor int32) {
if replyFor == self.lastPingTxSequence {
self.lastPingRx = time.Now()
logrus.Debugf("received outstanding ping for [l/%s]", self.id.Token)
}
}
func (self *impl) nextSequence() int32 {
self.sequenceLock.Lock()
defer self.sequenceLock.Unlock()
sequence := self.sequence
self.sequence++
return sequence
}
func newImpl(id *identity.TokenId, conn *net.UDPConn, peer *net.UDPAddr, f xlink.Forwarder) *impl {
now := time.Now()
xlinkImpl := &impl{
id: id,
conn: conn,
peer: peer,
lastPingRx: now,
lastPingTx: now,
txBuffer: newTransmitBuffer(),
forwarder: f,
}
xlinkImpl.rxBuffer = newReceiveBuffer(xlinkImpl)
return xlinkImpl
}
type impl struct {
id *identity.TokenId
conn *net.UDPConn
peer *net.UDPAddr
sequence int32
sequenceLock sync.Mutex
lastPingRx time.Time
lastPingTx time.Time
lastPingTxSequence int32
txBuffer *transmitBuffer
rxBuffer *receiveBuffer
forwarder xlink.Forwarder
}
const pingDelayMs = 5000
const pingCycleDelayMs = 500