Merge pull request #1511 from libp2p/merge-muxer-multistream

move go-stream-muxer-multistream here

config/muxer.go

@@ -3,7 +3,7 @@ package config
 import (
 	"fmt"
 
-	msmux "github.com/libp2p/go-stream-muxer-multistream"
+	msmux "github.com/libp2p/go-libp2p/p2p/muxer/muxer-multistream"
 
 	"github.com/libp2p/go-libp2p-core/network"
 

go.mod

@@ -30,7 +30,6 @@ require (
 	github.com/libp2p/go-nat v0.1.0
 	github.com/libp2p/go-netroute v0.2.0
 	github.com/libp2p/go-reuseport v0.1.0
-	github.com/libp2p/go-stream-muxer-multistream v0.4.0
 	github.com/libp2p/go-yamux/v3 v3.1.1
 	github.com/libp2p/zeroconf/v2 v2.1.1
 	github.com/lucas-clemente/quic-go v0.27.1

p2p/muxer/muxer-multistream/multistream.go

@@ -0,0 +1,75 @@
+// Package muxer_multistream implements a peerstream transport using
+// go-multistream to select the underlying stream muxer
+package muxer_multistream
+
+import (
+	"fmt"
+	"net"
+	"time"
+
+	"github.com/libp2p/go-libp2p-core/network"
+
+	mss "github.com/multiformats/go-multistream"
+)
+
+var DefaultNegotiateTimeout = time.Second * 60
+
+type Transport struct {
+	mux *mss.MultistreamMuxer
+
+	tpts map[string]network.Multiplexer
+
+	NegotiateTimeout time.Duration
+
+	OrderPreference []string
+}
+
+func NewBlankTransport() *Transport {
+	return &Transport{
+		mux:              mss.NewMultistreamMuxer(),
+		tpts:             make(map[string]network.Multiplexer),
+		NegotiateTimeout: DefaultNegotiateTimeout,
+	}
+}
+
+func (t *Transport) AddTransport(path string, tpt network.Multiplexer) {
+	t.mux.AddHandler(path, nil)
+	t.tpts[path] = tpt
+	t.OrderPreference = append(t.OrderPreference, path)
+}
+
+func (t *Transport) NewConn(nc net.Conn, isServer bool, scope network.PeerScope) (network.MuxedConn, error) {
+	if t.NegotiateTimeout != 0 {
+		if err := nc.SetDeadline(time.Now().Add(t.NegotiateTimeout)); err != nil {
+			return nil, err
+		}
+	}
+
+	var proto string
+	if isServer {
+		selected, _, err := t.mux.Negotiate(nc)
+		if err != nil {
+			return nil, err
+		}
+		proto = selected
+	} else {
+		selected, err := mss.SelectOneOf(t.OrderPreference, nc)
+		if err != nil {
+			return nil, err
+		}
+		proto = selected
+	}
+
+	if t.NegotiateTimeout != 0 {
+		if err := nc.SetDeadline(time.Time{}); err != nil {
+			return nil, err
+		}
+	}
+
+	tpt, ok := t.tpts[proto]
+	if !ok {
+		return nil, fmt.Errorf("selected protocol we don't have a transport for")
+	}
+
+	return tpt.NewConn(nc, isServer, scope)
+}

p2p/net/swarm/dial_worker_test.go

@@ -8,6 +8,7 @@ import (
 	"testing"
 	"time"
 
+	msmux "github.com/libp2p/go-libp2p/p2p/muxer/muxer-multistream"
 	"github.com/libp2p/go-libp2p/p2p/muxer/yamux"
 	csms "github.com/libp2p/go-libp2p/p2p/net/conn-security-multistream"
 	tptu "github.com/libp2p/go-libp2p/p2p/net/upgrader"
@@ -20,7 +21,6 @@ import (
 
 	"github.com/libp2p/go-libp2p-peerstore/pstoremem"
 	tnet "github.com/libp2p/go-libp2p-testing/net"
-	msmux "github.com/libp2p/go-stream-muxer-multistream"
 	ma "github.com/multiformats/go-multiaddr"
 
 	"github.com/stretchr/testify/require"

p2p/net/swarm/testing/testing.go

@@ -4,6 +4,7 @@ import (
 	"testing"
 	"time"
 
+	msmux "github.com/libp2p/go-libp2p/p2p/muxer/muxer-multistream"
 	"github.com/libp2p/go-libp2p/p2p/muxer/yamux"
 	csms "github.com/libp2p/go-libp2p/p2p/net/conn-security-multistream"
 	"github.com/libp2p/go-libp2p/p2p/net/swarm"
@@ -23,7 +24,6 @@ import (
 
 	"github.com/libp2p/go-libp2p-peerstore/pstoremem"
 	tnet "github.com/libp2p/go-libp2p-testing/net"
-	msmux "github.com/libp2p/go-stream-muxer-multistream"
 	ma "github.com/multiformats/go-multiaddr"
 	"github.com/stretchr/testify/require"
 )