Ensure we can't start xgress rx more than once

router/xgress/xgress.go

@@ -108,6 +108,7 @@ type Xgress struct {
 	closed         concurrenz.AtomicBoolean
 	closeHandler   CloseHandler
 	peekHandlers   []PeekHandler
+	rxerStarted    concurrenz.AtomicBoolean
 }
 
 func NewXgress(sessionId *identity.TokenId, address Address, peer Connection, originator Originator, options *Options) *Xgress {
@@ -250,7 +251,7 @@ func (self *Xgress) SendAcknowledgement(acknowledgement *Acknowledgement) error
 }
 
 func (self *Xgress) payloadIngester(payload *Payload) {
-	if payload.IsSessionStartFlagSet() {
+	if payload.IsSessionStartFlagSet() && self.rxerStarted.CompareAndSwap(false, true) {
 		pfxlog.ContextLogger(self.Label()).WithFields(payload.GetLoggerFields()).Debug("received session start, starting xgress receiver")
 		go self.rx()
 	}
@@ -351,7 +352,7 @@ func (self *Xgress) tx() {
 func (self *Xgress) rx() {
 	log := pfxlog.ContextLogger(self.Label())
 
-	log.Debug("started")
+	log.Debugf("started with peer: %v", self.peer.LogContext())
 	defer log.Debug("exited")
 
 	defer func() {
@@ -378,39 +379,37 @@ func (self *Xgress) rx() {
 			return
 		}
 
-		if !self.closed.Get() {
-			start := 0
-			remaining := n
-			payloads := 0
-			for {
-				length := mathz.MinInt(remaining, int(self.Options.Mtu))
-				payload := &Payload{
-					Header: Header{
-						SessionId: self.sessionId,
-						Flags:     SetOriginatorFlag(0, self.originator),
-					},
-					Sequence: self.nextReceiveSequence(),
-					Data:     buffer[start : start+length],
-					Headers:  headers,
-				}
-				start += length
-				remaining -= length
-				payloads++
-
-				self.forwardPayload(payload)
-				payloadLogger := log.WithFields(payload.GetLoggerFields())
-				payloadLogger.Debugf("received [%s]", info.ByteCount(int64(n)))
-
-				if remaining < 1 {
-					break
-				}
-			}
-
-			logrus.Debugf("received [%d] payloads for [%d] bytes", payloads, n)
-
-		} else {
+		if self.closed.Get() {
 			return
 		}
+		start := 0
+		remaining := n
+		payloads := 0
+		for {
+			length := mathz.MinInt(remaining, int(self.Options.Mtu))
+			payload := &Payload{
+				Header: Header{
+					SessionId: self.sessionId,
+					Flags:     SetOriginatorFlag(0, self.originator),
+				},
+				Sequence: self.nextReceiveSequence(),
+				Data:     buffer[start : start+length],
+				Headers:  headers,
+			}
+			start += length
+			remaining -= length
+			payloads++
+
+			self.forwardPayload(payload)
+			payloadLogger := log.WithFields(payload.GetLoggerFields())
+			payloadLogger.Debugf("received [%s]", info.ByteCount(int64(n)))
+
+			if remaining < 1 {
+				break
+			}
+		}
+
+		logrus.Debugf("received [%d] payloads for [%d] bytes", payloads, n)
 	}
 }
 

router/xgress_transport/connection.go

@@ -32,15 +32,13 @@ func (c *transportXgressConn) ReadPayload() ([]byte, map[uint8][]byte, error) {
 	buffer := make([]byte, 10240)
 	n, err := c.Reader().Read(buffer)
 	if err == nil {
-		if len(buffer) < (5 * 1024) {
+		if n < (5 * 1024) {
 			buffer = append([]byte(nil), buffer[:n]...)
 		}
 	}
-	return buffer[:n], nil, err
-}
 
-func (c *transportXgressConn) Write(p []byte) (n int, err error) {
-	return c.Writer().Write(p)
+	data := buffer[:n]
+	return data, nil, err
 }
 
 func (c *transportXgressConn) WritePayload(p []byte, _ map[uint8][]byte) (n int, err error) {

router/xgress_transport/dialer.go

@@ -56,7 +56,7 @@ func (txd *dialer) Dial(destination string, sessionId *identity.TokenId, address
 		return nil, err
 	}
 
-	conn := &transportXgressConn{peer}
+	conn := &transportXgressConn{Connection: peer}
 	x := xgress.NewXgress(sessionId, address, conn, xgress.Terminator, txd.options)
 	bindHandler.HandleXgressBind(x)
 	x.Start()

router/xgress_transport/listener.go

@@ -71,7 +71,7 @@ func (listener *listener) Close() error {
 }
 
 func (listener *listener) handleConnect(peer transport.Connection, bindHandler xgress.BindHandler) {
-	conn := &transportXgressConn{peer}
+	conn := &transportXgressConn{Connection: peer}
 	log := pfxlog.ContextLogger(conn.LogContext())
 
 	request, err := xgress.ReceiveRequest(peer)