P2P demo fixes

1 files changed, 34 insertions, 25 deletions

diff --git a/build/gobind-pinecone/monolith.go b/build/gobind-pinecone/monolith.go
index 500403ae..241adcf7 100644
--- a/build/gobind-pinecone/monolith.go
+++ b/build/gobind-pinecone/monolith.go
@@ -28,6 +28,7 @@ import (
 	"path/filepath"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/gorilla/mux"
@@ -66,6 +67,7 @@ const (
 	PeerTypeRemote    = pineconeRouter.PeerTypeRemote
 	PeerTypeMulticast = pineconeRouter.PeerTypeMulticast
 	PeerTypeBluetooth = pineconeRouter.PeerTypeBluetooth
+	PeerTypeBonjour   = pineconeRouter.PeerTypeBonjour
 )
 
 type DendriteMonolith struct {
@@ -82,6 +84,10 @@ type DendriteMonolith struct {
 	userAPI           userapiAPI.UserInternalAPI
 }
 
+func (m *DendriteMonolith) PublicKey() string {
+	return m.PineconeRouter.PublicKey().String()
+}
+
 func (m *DendriteMonolith) BaseURL() string {
 	return fmt.Sprintf("http://%s", m.listener.Addr().String())
 }
@@ -134,32 +140,21 @@ func (m *DendriteMonolith) Conduit(zone string, peertype int) (*Conduit, error)
 	go func() {
 		conduit.portMutex.Lock()
 		defer conduit.portMutex.Unlock()
-	loop:
-		for i := 1; i <= 10; i++ {
-			logrus.Errorf("Attempting authenticated connect (attempt %d)", i)
-			var err error
-			conduit.port, err = m.PineconeRouter.Connect(
-				l,
-				pineconeRouter.ConnectionZone(zone),
-				pineconeRouter.ConnectionPeerType(peertype),
-			)
-			switch err {
-			case io.ErrClosedPipe:
-				logrus.Errorf("Authenticated connect failed due to closed pipe (attempt %d)", i)
-				return
-			case io.EOF:
-				logrus.Errorf("Authenticated connect failed due to EOF (attempt %d)", i)
-				break loop
-			case nil:
-				logrus.Errorf("Authenticated connect succeeded, connected to port %d (attempt %d)", conduit.port, i)
-				return
-			default:
-				logrus.WithError(err).Errorf("Authenticated connect failed (attempt %d)", i)
-				time.Sleep(time.Second)
-			}
+
+		logrus.Errorf("Attempting authenticated connect")
+		var err error
+		if conduit.port, err = m.PineconeRouter.Connect(
+			l,
+			pineconeRouter.ConnectionZone(zone),
+			pineconeRouter.ConnectionPeerType(peertype),
+		); err != nil {
+			logrus.Errorf("Authenticated connect failed: %s", err)
+			_ = l.Close()
+			_ = r.Close()
+			_ = conduit.Close()
+			return
 		}
-		_ = l.Close()
-		_ = r.Close()
+		logrus.Infof("Authenticated connect succeeded (port %d)", conduit.port)
 	}()
 	return conduit, nil
 }
@@ -395,6 +390,7 @@ func (m *DendriteMonolith) Stop() {
 const MaxFrameSize = types.MaxFrameSize
 
 type Conduit struct {
+	closed    atomic.Bool
 	conn      net.Conn
 	port      types.SwitchPortID
 	portMutex sync.Mutex
@@ -407,10 +403,16 @@ func (c *Conduit) Port() int {
 }
 
 func (c *Conduit) Read(b []byte) (int, error) {
+	if c.closed.Load() {
+		return 0, io.EOF
+	}
 	return c.conn.Read(b)
 }
 
 func (c *Conduit) ReadCopy() ([]byte, error) {
+	if c.closed.Load() {
+		return nil, io.EOF
+	}
 	var buf [65535 * 2]byte
 	n, err := c.conn.Read(buf[:])
 	if err != nil {
@@ -420,9 +422,16 @@ func (c *Conduit) ReadCopy() ([]byte, error) {
 }
 
 func (c *Conduit) Write(b []byte) (int, error) {
+	if c.closed.Load() {
+		return 0, io.EOF
+	}
 	return c.conn.Write(b)
 }
 
 func (c *Conduit) Close() error {
+	if c.closed.Load() {
+		return io.ErrClosedPipe
+	}
+	c.closed.Store(true)
 	return c.conn.Close()
 }