pkg/p2p: add IncompatibleStreamError

go.mod

@@ -25,6 +25,7 @@ require (
 	github.com/libp2p/go-libp2p-transport v0.1.0 // indirect
 	github.com/libp2p/go-stream-muxer v0.1.0 // indirect
 	github.com/multiformats/go-multiaddr v0.2.0
+	github.com/multiformats/go-multistream v0.1.0
 	github.com/whyrusleeping/go-smux-multiplex v3.0.16+incompatible // indirect
 	github.com/whyrusleeping/go-smux-multistream v2.0.2+incompatible // indirect
 	github.com/whyrusleeping/go-smux-yamux v2.0.9+incompatible // indirect

pkg/p2p/libp2p/libp2p.go

@@ -5,6 +5,8 @@ import (
 	"fmt"
 	"time"
 
+	"github.com/multiformats/go-multistream"
+
 	"github.com/janos/bee/pkg/p2p"
 
 	"github.com/libp2p/go-libp2p"
@@ -20,7 +22,6 @@ import (
 	libp2pquic "github.com/libp2p/go-libp2p-quic-transport"
 	secio "github.com/libp2p/go-libp2p-secio"
 	libp2ptls "github.com/libp2p/go-libp2p-tls"
-	"github.com/multiformats/go-multiaddr"
 	ma "github.com/multiformats/go-multiaddr"
 )
 
@@ -129,7 +130,7 @@ func (s *Service) Addresses() (addrs []string, err error) {
 	return addrs, nil
 }
 
-func (s *Service) Connect(ctx context.Context, addr multiaddr.Multiaddr) (peerID string, err error) {
+func (s *Service) Connect(ctx context.Context, addr ma.Multiaddr) (peerID string, err error) {
 	// Extract the peer ID from the multiaddr.
 	info, err := peer.AddrInfoFromP2pAddr(addr)
 	if err != nil {
@@ -149,6 +150,9 @@ func (s *Service) NewStream(ctx context.Context, peerID, protocolName, streamNam
 	swarmStreamName := p2p.NewSwarmStreamName(protocolName, streamName, version)
 	st, err := s.host.NewStream(ctx, id, protocol.ID(swarmStreamName))
 	if err != nil {
+		if err == multistream.ErrNotSupported || err == multistream.ErrIncorrectVersion {
+			return nil, p2p.NewIncompatibleStreamError(err)
+		}
 		return nil, fmt.Errorf("create stream %q to %q: %w", swarmStreamName, peerID, err)
 	}
 	return stream{st}, nil

pkg/p2p/p2p.go

@@ -2,6 +2,7 @@ package p2p
 
 import (
 	"context"
+	"fmt"
 	"io"
 
 	ma "github.com/multiformats/go-multiaddr"
@@ -44,6 +45,20 @@ type StreamSpec struct {
 	Handler func(Peer)
 }
 
+type IncompatibleStreamError struct {
+	err error
+}
+
+func NewIncompatibleStreamError(err error) *IncompatibleStreamError {
+	return &IncompatibleStreamError{err: err}
+}
+
+func (e *IncompatibleStreamError) Unwrap() error { return e.err }
+
+func (e *IncompatibleStreamError) Error() string {
+	return fmt.Sprintf("incompatible stream: %v", e.err)
+}
+
 func NewSwarmStreamName(protocol, stream, version string) string {
 	return "/swarm/" + protocol + "/" + stream + "/" + version
 }