From 84be6c49d4bdd7b6c00fff26bbd4226a7073c877 Mon Sep 17 00:00:00 2001 From: Michael Quigley Date: Wed, 29 Jul 2020 12:35:17 -0400 Subject: [PATCH 1/5] Wiring transport.Configuration throughout channel2 and transport. Lint removal. --- router/router.go | 6 +- router/xgress_proxy/factory.go | 8 +- router/xgress_proxy/listener.go | 6 +- router/xgress_proxy_udp/listener.go | 4 +- router/xgress_transport/client.go | 4 +- router/xgress_transport/dialer.go | 6 +- router/xgress_transport/factory.go | 10 +- router/xgress_transport/listener.go | 6 +- router/xgress_transport_udp/connection.go | 4 +- router/xgress_transport_udp/listener.go | 4 +- router/xlink_transwarp/buffer.go | 111 ----- router/xlink_transwarp/config.go | 85 ---- router/xlink_transwarp/dialer.go | 96 ---- router/xlink_transwarp/factory.go | 59 --- router/xlink_transwarp/listener.go | 99 ---- router/xlink_transwarp/message.go | 465 ------------------ .../message_acknowledgement.go | 99 ---- router/xlink_transwarp/message_payload.go | 99 ---- router/xlink_transwarp/message_window.go | 129 ----- router/xlink_transwarp/xlink.go | 179 ------- 20 files changed, 33 insertions(+), 1446 deletions(-) delete mode 100644 router/xlink_transwarp/buffer.go delete mode 100644 router/xlink_transwarp/config.go delete mode 100644 router/xlink_transwarp/dialer.go delete mode 100644 router/xlink_transwarp/factory.go delete mode 100644 router/xlink_transwarp/listener.go delete mode 100644 router/xlink_transwarp/message.go delete mode 100644 router/xlink_transwarp/message_acknowledgement.go delete mode 100644 router/xlink_transwarp/message_payload.go delete mode 100644 router/xlink_transwarp/message_window.go delete mode 100644 router/xlink_transwarp/xlink.go diff --git a/router/router.go b/router/router.go index db65b5e88..29d88751d 100644 --- a/router/router.go +++ b/router/router.go @@ -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" @@ -178,11 +177,10 @@ func (self *Router) registerComponents() error { self.metricsRegistry, ) 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", xgress_proxy.NewFactory(self.config.Id, self, nil)) 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, nil)) xgress.GlobalRegistry().Register("transport_udp", xgress_transport_udp.NewFactory(self.config.Id, self)) return nil diff --git a/router/xgress_proxy/factory.go b/router/xgress_proxy/factory.go index 3b4e780da..c607149dd 100644 --- a/router/xgress_proxy/factory.go +++ b/router/xgress_proxy/factory.go @@ -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, c transport.Configuration) xgress.Factory { + return &factory{id: id, ctrl: ctrl, c: c} } 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.c, 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 + c transport.Configuration } diff --git a/router/xgress_proxy/listener.go b/router/xgress_proxy/listener.go index 0da7abf49..085eb9fd1 100644 --- a/router/xgress_proxy/listener.go +++ b/router/xgress_proxy/listener.go @@ -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, c transport.Configuration, service string) xgress.Listener { return &listener{ id: id, ctrl: ctrl, options: options, + c: c, 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.c)) go func() { for { select { @@ -73,6 +74,7 @@ type listener struct { id *identity.TokenId ctrl xgress.CtrlChannel options *xgress.Options + c transport.Configuration service string closeHelper *xgress.CloseHelper } diff --git a/router/xgress_proxy_udp/listener.go b/router/xgress_proxy_udp/listener.go index 485564798..49c441aa8 100644 --- a/router/xgress_proxy_udp/listener.go +++ b/router/xgress_proxy_udp/listener.go @@ -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.MaxPacketSize) read, addr, err := l.conn.ReadFrom(buf) if err != nil { logrus.Errorf("error reading packet (%v)", err) diff --git a/router/xgress_transport/client.go b/router/xgress_transport/client.go index e08e32eec..540b2de5e 100644 --- a/router/xgress_transport/client.go +++ b/router/xgress_transport/client.go @@ -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, c transport.Configuration) (transport.Connection, error) { + peer, err := addr.Dial("i/"+id.Token, id, c) if err != nil { return nil, err } diff --git a/router/xgress_transport/dialer.go b/router/xgress_transport/dialer.go index 76297c61a..de2ba1303 100644 --- a/router/xgress_transport/dialer.go +++ b/router/xgress_transport/dialer.go @@ -27,17 +27,19 @@ type dialer struct { id *identity.TokenId ctrl xgress.CtrlChannel options *xgress.Options + c 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, c transport.Configuration) (xgress.Dialer, error) { txd := &dialer{ id: id, ctrl: ctrl, options: options, + c: c, } return txd, nil } @@ -48,7 +50,7 @@ func (txd *dialer) Dial(destination string, sessionId *identity.TokenId, address return 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.c) if err != nil { return err } diff --git a/router/xgress_transport/factory.go b/router/xgress_transport/factory.go index 0d000a424..198a0a237 100644 --- a/router/xgress_transport/factory.go +++ b/router/xgress_transport/factory.go @@ -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 + c 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, c transport.Configuration) xgress.Factory { + return &factory{id: id, ctrl: ctrl, c: c} } 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.c), 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.c) } diff --git a/router/xgress_transport/listener.go b/router/xgress_transport/listener.go index 049a43b9e..d891331df 100644 --- a/router/xgress_transport/listener.go +++ b/router/xgress_transport/listener.go @@ -28,14 +28,16 @@ type listener struct { id *identity.TokenId ctrl xgress.CtrlChannel options *xgress.Options + c 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, c transport.Configuration) xgress.Listener { return &listener{ id: id, ctrl: ctrl, options: options, + c: c, 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.c)) go func() { for { select { diff --git a/router/xgress_transport_udp/connection.go b/router/xgress_transport_udp/connection.go index 6f205fcc6..230cd4a75 100644 --- a/router/xgress_transport_udp/connection.go +++ b/router/xgress_transport_udp/connection.go @@ -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.MaxPacketSize) n, err := p.Read(buffer) if err != nil { return nil, nil, err diff --git a/router/xgress_transport_udp/listener.go b/router/xgress_transport_udp/listener.go index 0c58f9bcb..9a285d6df 100644 --- a/router/xgress_transport_udp/listener.go +++ b/router/xgress_transport_udp/listener.go @@ -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.MaxPacketSize) logger.Debugf("Trying to read next packet") bytesRead, addr, err := l.conn.ReadFrom(buf) logger.Debugf("Packet read complete: %v bytes read", bytesRead) diff --git a/router/xlink_transwarp/buffer.go b/router/xlink_transwarp/buffer.go deleted file mode 100644 index 621771301..000000000 --- a/router/xlink_transwarp/buffer.go +++ /dev/null @@ -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 diff --git a/router/xlink_transwarp/config.go b/router/xlink_transwarp/config.go deleted file mode 100644 index ac2bdd873..000000000 --- a/router/xlink_transwarp/config.go +++ /dev/null @@ -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 -} diff --git a/router/xlink_transwarp/dialer.go b/router/xlink_transwarp/dialer.go deleted file mode 100644 index 09a7b2f4a..000000000 --- a/router/xlink_transwarp/dialer.go +++ /dev/null @@ -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{} -} diff --git a/router/xlink_transwarp/factory.go b/router/xlink_transwarp/factory.go deleted file mode 100644 index 4f9affced..000000000 --- a/router/xlink_transwarp/factory.go +++ /dev/null @@ -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 -} diff --git a/router/xlink_transwarp/listener.go b/router/xlink_transwarp/listener.go deleted file mode 100644 index 794fcb96e..000000000 --- a/router/xlink_transwarp/listener.go +++ /dev/null @@ -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 -} diff --git a/router/xlink_transwarp/message.go b/router/xlink_transwarp/message.go deleted file mode 100644 index 48a8e96bc..000000000 --- a/router/xlink_transwarp/message.go +++ /dev/null @@ -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 --------------------------------------------------------------------------------- // - * - * 0 1 2 3 - * 4 5 6 7 - * 8 - * 9 - * 10 - * 11 12 - * 13 14 - * - * // --- data section ------------------------------------------------------------------------------------ // - * - * 15 -> (15 + headers_length) - * (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 - * - * 0 - * 1 - * 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 -} diff --git a/router/xlink_transwarp/message_acknowledgement.go b/router/xlink_transwarp/message_acknowledgement.go deleted file mode 100644 index 82f7f68c4..000000000 --- a/router/xlink_transwarp/message_acknowledgement.go +++ /dev/null @@ -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 - * - * 0 1 2 3 - * 4 5 6 7 - * 8 9 10 11 - * 12 -> (12 + session_id_length) - * (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 -} diff --git a/router/xlink_transwarp/message_payload.go b/router/xlink_transwarp/message_payload.go deleted file mode 100644 index 5351cc121..000000000 --- a/router/xlink_transwarp/message_payload.go +++ /dev/null @@ -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 - * - * 0 1 2 3 - * 4 5 6 7 - * 8 9 10 11 - * 12 13 14 15 - * 16 -> (16 + session_id_length) - * (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 -} diff --git a/router/xlink_transwarp/message_window.go b/router/xlink_transwarp/message_window.go deleted file mode 100644 index 61485d269..000000000 --- a/router/xlink_transwarp/message_window.go +++ /dev/null @@ -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 -} diff --git a/router/xlink_transwarp/xlink.go b/router/xlink_transwarp/xlink.go deleted file mode 100644 index 51adaa9d9..000000000 --- a/router/xlink_transwarp/xlink.go +++ /dev/null @@ -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 From eb8aeb99bc797728e534a21f619a52a27afd0085 Mon Sep 17 00:00:00 2001 From: Michael Quigley Date: Wed, 29 Jul 2020 17:12:07 -0400 Subject: [PATCH 2/5] 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 } From 921675ac38bc4b1bfb12fcdf1696580dc1d0a816 Mon Sep 17 00:00:00 2001 From: Michael Quigley Date: Fri, 31 Jul 2020 11:19:24 -0400 Subject: [PATCH 3/5] Better max UDP packet size naming. --- router/xgress_proxy_udp/listener.go | 2 +- router/xgress_transport_udp/connection.go | 2 +- router/xgress_transport_udp/listener.go | 2 +- 3 files changed, 3 insertions(+), 3 deletions(-) diff --git a/router/xgress_proxy_udp/listener.go b/router/xgress_proxy_udp/listener.go index 49c441aa8..ceb1e93fc 100644 --- a/router/xgress_proxy_udp/listener.go +++ b/router/xgress_proxy_udp/listener.go @@ -77,7 +77,7 @@ func (l *listener) relay() { }() for { - buf := make([]byte, info.MaxPacketSize) + buf := make([]byte, info.MaxUdpPacketSize) read, addr, err := l.conn.ReadFrom(buf) if err != nil { logrus.Errorf("error reading packet (%v)", err) diff --git a/router/xgress_transport_udp/connection.go b/router/xgress_transport_udp/connection.go index 230cd4a75..6aca79283 100644 --- a/router/xgress_transport_udp/connection.go +++ b/router/xgress_transport_udp/connection.go @@ -27,7 +27,7 @@ func (p *packetConn) LogContext() string { } func (p *packetConn) ReadPayload() ([]byte, map[uint8][]byte, error) { - buffer := make([]byte, info.MaxPacketSize) + buffer := make([]byte, info.MaxUdpPacketSize) n, err := p.Read(buffer) if err != nil { return nil, nil, err diff --git a/router/xgress_transport_udp/listener.go b/router/xgress_transport_udp/listener.go index 9a285d6df..5878a35b0 100644 --- a/router/xgress_transport_udp/listener.go +++ b/router/xgress_transport_udp/listener.go @@ -88,7 +88,7 @@ func (l *listener) relayIncomingPackets() { }() for { - buf := make([]byte, info.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) From 8649954475a0ee478b09bb597382aed586ecd2be Mon Sep 17 00:00:00 2001 From: Michael Quigley Date: Fri, 31 Jul 2020 12:53:21 -0400 Subject: [PATCH 4/5] Lint removal and polish application. --- router/xgress_proxy/factory.go | 8 ++++---- router/xgress_proxy/listener.go | 8 ++++---- router/xgress_transport/client.go | 4 ++-- router/xgress_transport/dialer.go | 8 ++++---- router/xgress_transport/factory.go | 10 +++++----- router/xgress_transport/listener.go | 8 ++++---- router/xlink_transport/dialer.go | 4 ++-- router/xlink_transport/factory.go | 8 ++++---- router/xlink_transport/listener.go | 6 +++--- 9 files changed, 32 insertions(+), 32 deletions(-) diff --git a/router/xgress_proxy/factory.go b/router/xgress_proxy/factory.go index c607149dd..210801d8e 100644 --- a/router/xgress_proxy/factory.go +++ b/router/xgress_proxy/factory.go @@ -23,8 +23,8 @@ import ( "github.com/openziti/foundation/transport" ) -func NewFactory(id *identity.TokenId, ctrl xgress.CtrlChannel, c transport.Configuration) xgress.Factory { - return &factory{id: id, ctrl: ctrl, c: c} +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) { @@ -35,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, factory.c, service), nil + return newListener(factory.id, factory.ctrl, options, factory.tcfg, service), nil } func (factory *factory) CreateDialer(optionsData xgress.OptionsData) (xgress.Dialer, error) { @@ -46,5 +46,5 @@ type factory struct { id *identity.TokenId ctrl xgress.CtrlChannel options *xgress.Options - c transport.Configuration + tcfg transport.Configuration } diff --git a/router/xgress_proxy/listener.go b/router/xgress_proxy/listener.go index 085eb9fd1..50ef89e5b 100644 --- a/router/xgress_proxy/listener.go +++ b/router/xgress_proxy/listener.go @@ -24,12 +24,12 @@ import ( "github.com/openziti/foundation/transport" ) -func newListener(id *identity.TokenId, ctrl xgress.CtrlChannel, options *xgress.Options, c transport.Configuration, 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, - c: c, + tcfg: tcfg, service: service, closeHelper: &xgress.CloseHelper{}, } @@ -42,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, listener.c)) + go listener.closeHelper.Init(txAddress.MustListen("tcp", listener.id, incomingPeers, listener.tcfg)) go func() { for { select { @@ -74,7 +74,7 @@ type listener struct { id *identity.TokenId ctrl xgress.CtrlChannel options *xgress.Options - c transport.Configuration + tcfg transport.Configuration service string closeHelper *xgress.CloseHelper } diff --git a/router/xgress_transport/client.go b/router/xgress_transport/client.go index 540b2de5e..82beea282 100644 --- a/router/xgress_transport/client.go +++ b/router/xgress_transport/client.go @@ -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, c transport.Configuration) (transport.Connection, error) { - peer, err := addr.Dial("i/"+id.Token, id, c) +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 } diff --git a/router/xgress_transport/dialer.go b/router/xgress_transport/dialer.go index 1b947f212..e7e222c90 100644 --- a/router/xgress_transport/dialer.go +++ b/router/xgress_transport/dialer.go @@ -28,19 +28,19 @@ type dialer struct { id *identity.TokenId ctrl xgress.CtrlChannel options *xgress.Options - c transport.Configuration + tcfg transport.Configuration } func (txd *dialer) IsTerminatorValid(string, string) bool { return true } -func newDialer(id *identity.TokenId, ctrl xgress.CtrlChannel, options *xgress.Options, c transport.Configuration) (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, - c: c, + tcfg: tcfg, } return txd, nil } @@ -51,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, txd.c) + peer, err := txDestination.Dial("x/"+sessionId.Token, sessionId, txd.tcfg) if err != nil { return nil, err } diff --git a/router/xgress_transport/factory.go b/router/xgress_transport/factory.go index 198a0a237..da48756ba 100644 --- a/router/xgress_transport/factory.go +++ b/router/xgress_transport/factory.go @@ -26,20 +26,20 @@ type factory struct { id *identity.TokenId ctrl xgress.CtrlChannel options *xgress.Options - c transport.Configuration + tcfg transport.Configuration } // NewFactory returns a new Transport Xgress factory -func NewFactory(id *identity.TokenId, ctrl xgress.CtrlChannel, c transport.Configuration) xgress.Factory { - return &factory{id: id, ctrl: ctrl, c: c} +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, factory.c), 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, factory.c) + return newDialer(factory.id, factory.ctrl, options, factory.tcfg) } diff --git a/router/xgress_transport/listener.go b/router/xgress_transport/listener.go index d891331df..73780b0f5 100644 --- a/router/xgress_transport/listener.go +++ b/router/xgress_transport/listener.go @@ -28,16 +28,16 @@ type listener struct { id *identity.TokenId ctrl xgress.CtrlChannel options *xgress.Options - c transport.Configuration + tcfg transport.Configuration closeHelper *xgress.CloseHelper } -func newListener(id *identity.TokenId, ctrl xgress.CtrlChannel, options *xgress.Options, c transport.Configuration) 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, - c: c, + tcfg: tcfg, closeHelper: &xgress.CloseHelper{}, } } @@ -49,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, listener.c)) + go listener.closeHelper.Init(txAddress.MustListen("tcp", listener.id, incomingPeers, listener.tcfg)) go func() { for { select { diff --git a/router/xlink_transport/dialer.go b/router/xlink_transport/dialer.go index 186f0f785..4ca936d43 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.NewChannelWithTransportConfig("l/"+linkId.Token, dialer, self.config.options, self.c) + ch, err := channel2.NewChannelWithTransportConfiguration("l/"+linkId.Token, dialer, self.config.options, self.tcfg) if err == nil { xlink := &impl{id: linkId, ch: ch} @@ -59,5 +59,5 @@ type dialer struct { config *dialerConfig accepter xlink.Accepter chAccepter ChannelAccepter - c transport.Configuration + tcfg transport.Configuration } diff --git a/router/xlink_transport/factory.go b/router/xlink_transport/factory.go index 2ee7782d4..68f8b346c 100644 --- a/router/xlink_transport/factory.go +++ b/router/xlink_transport/factory.go @@ -24,7 +24,7 @@ import ( ) func NewFactory(accepter xlink.Accepter, chAccepter ChannelAccepter, c transport.Configuration) xlink.Factory { - return &factory{accepter: accepter, chAccepter: chAccepter, c: c} + return &factory{accepter: accepter, chAccepter: chAccepter, tcfg: c} } func (self *factory) CreateListener(id *identity.TokenId, _ xlink.Forwarder, configData map[interface{}]interface{}) (xlink.Listener, error) { @@ -37,7 +37,7 @@ func (self *factory) CreateListener(id *identity.TokenId, _ xlink.Forwarder, con config: config, accepter: self.accepter, chAccepter: self.chAccepter, - c: self.c, + tcfg: self.tcfg, }, nil } @@ -51,12 +51,12 @@ func (self *factory) CreateDialer(id *identity.TokenId, _ xlink.Forwarder, confi config: config, accepter: self.accepter, chAccepter: self.chAccepter, - c: self.c, + tcfg: self.tcfg, }, nil } type factory struct { accepter xlink.Accepter chAccepter ChannelAccepter - c transport.Configuration + tcfg transport.Configuration } diff --git a/router/xlink_transport/listener.go b/router/xlink_transport/listener.go index 97fe4e605..57ec4e421 100644 --- a/router/xlink_transport/listener.go +++ b/router/xlink_transport/listener.go @@ -27,7 +27,7 @@ import ( ) func (self *listener) Listen() error { - listener := channel2.NewClassicListenerWithConfiguration(self.id, self.config.bind, self.config.options.ConnectOptions, self.c) + listener := channel2.NewClassicListenerWithTransportConfiguration(self.id, self.config.bind, self.config.options.ConnectOptions, self.tcfg) self.listener = listener if err := self.listener.Listen(); err != nil { @@ -47,7 +47,7 @@ func (self *listener) Close() error { func (self *listener) acceptLoop() { for { - ch, err := channel2.NewChannelWithTransportConfig("link", self.listener, self.config.options, self.c) + 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) @@ -79,5 +79,5 @@ type listener struct { listener channel2.UnderlayListener accepter xlink.Accepter chAccepter ChannelAccepter - c transport.Configuration + tcfg transport.Configuration } From 74e07ebce6adb9aa28726ba7e495ce8e67051e05 Mon Sep 17 00:00:00 2001 From: Michael Quigley Date: Mon, 3 Aug 2020 10:02:44 -0400 Subject: [PATCH 5/5] Allow empty 'transport:' configurations to succeed. --- router/config.go | 4 ---- 1 file changed, 4 deletions(-) diff --git a/router/config.go b/router/config.go index 00007ff1f..0d05aaff0 100644 --- a/router/config.go +++ b/router/config.go @@ -27,11 +27,9 @@ 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" ) @@ -300,8 +298,6 @@ func LoadConfig(path string) (*Config, error) { 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)) } }