Support for P2P listeners and streams

.travis.yml

@@ -18,7 +18,11 @@ services:
 before_install:
 - docker pull ipfs/go-ipfs:master
 - mkdir /tmp/ipfs && chmod 0777 /tmp/ipfs
-- docker run -d -v /tmp/ipfs:/data/ipfs -p 8080:8080 -p 4001:4001 -p 5001:5001 ipfs/go-ipfs:master --enable-pubsub-experiment
+- docker run -v /tmp/ipfs:/data/ipfs --user ipfs --entrypoint "" ipfs/go-ipfs:master ipfs init
+- docker run -v /tmp/ipfs:/data/ipfs --user ipfs --entrypoint "" ipfs/go-ipfs:master ipfs config --json Experimental.Libp2pStreamMounting true
+- docker run -v /tmp/ipfs:/data/ipfs --user ipfs --entrypoint "" ipfs/go-ipfs:master ipfs config Addresses.API /ip4/0.0.0.0/tcp/5001
+- docker run -v /tmp/ipfs:/data/ipfs --user ipfs --entrypoint "" ipfs/go-ipfs:master ipfs config Addresses.Gateway /ip4/0.0.0.0/tcp/8080
+- docker run -d -v /tmp/ipfs:/data/ipfs -p 8080:8080 -p 4001:4001 -p 5001:5001 ipfs/go-ipfs:master daemon --enable-pubsub-experiment
 
 install:
   - go get -t -v ./...

p2p.go

@@ -0,0 +1,129 @@
+package shell
+
+import (
+	"context"
+	"strconv"
+
+	ma "github.com/multiformats/go-multiaddr"
+)
+
+// P2PListener describes a P2P listener.
+type P2PListener struct {
+	Protocol string
+	Address  string
+}
+
+// P2POpenListener forwards P2P connections to a network multiaddr.
+func (s *Shell) P2POpenListener(ctx context.Context, protocol, maddr string) (*P2PListener, error) {
+	if _, err := ma.NewMultiaddr(maddr); err != nil {
+		return nil, err
+	}
+
+	var response *P2PListener
+	err := s.Request("p2p/listener/open").
+		Arguments(protocol, maddr).Exec(ctx, &response)
+	if err != nil {
+		return nil, err
+	}
+
+	return response, nil
+}
+
+// P2PCloseListener closes one or all active P2P listeners.
+func (s *Shell) P2PCloseListener(ctx context.Context, protocol string, closeAll bool) error {
+	req := s.Request("p2p/listener/close").
+		Option("all", strconv.FormatBool(closeAll))
+
+	if protocol != "" {
+		req.Arguments(protocol)
+	}
+
+	if err := req.Exec(ctx, nil); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+// P2PListenerList contains a slice of P2PListeners.
+type P2PListenerList struct {
+	Listeners []*P2PListener
+}
+
+// P2PListListeners lists all P2P listeners.
+func (s *Shell) P2PListListeners(ctx context.Context) (*P2PListenerList, error) {
+	var response *P2PListenerList
+
+	if err := s.Request("p2p/listener/ls").Exec(ctx, &response); err != nil {
+		return nil, err
+	}
+
+	return response, nil
+}
+
+// P2PStream describes a P2P stream.
+type P2PStream struct {
+	Protocol string
+	Address  string
+}
+
+// P2PStreamDial dials to a peer's P2P listener.
+func (s *Shell) P2PStreamDial(ctx context.Context, peerID, protocol, listenerMaddr string) (*P2PStream, error) {
+	var response *P2PStream
+	req := s.Request("p2p/stream/dial").
+		Arguments(peerID, protocol)
+
+	if listenerMaddr != "" {
+		if _, err := ma.NewMultiaddr(listenerMaddr); err != nil {
+			return nil, err
+		}
+		req.Arguments(listenerMaddr)
+	}
+
+	if err := req.Exec(ctx, &response); err != nil {
+		return nil, err
+	}
+
+	return response, nil
+}
+
+// P2PCloseStream closes one or all active P2P streams.
+func (s *Shell) P2PCloseStream(ctx context.Context, handlerID string, closeAll bool) error {
+	req := s.Request("p2p/stream/close").
+		Option("all", strconv.FormatBool(closeAll))
+
+	if handlerID != "" {
+		req.Arguments(handlerID)
+	}
+
+	if err := req.Exec(ctx, nil); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+// P2PStreamsList contains a slice of streams.
+type P2PStreamsList struct {
+	Streams []*struct {
+		HandlerID     string
+		Protocol      string
+		LocalPeer     string
+		LocalAddress  string
+		RemotePeer    string
+		RemoteAddress string
+	}
+}
+
+// P2PListStreams lists all P2P streams.
+func (s *Shell) P2PListStreams(ctx context.Context) (*P2PStreamsList, error) {
+	var response *P2PStreamsList
+	req := s.Request("p2p/stream/ls").
+		Option("headers", strconv.FormatBool(true))
+
+	if err := req.Exec(ctx, &response); err != nil {
+		return nil, err
+	}
+
+	return response, nil
+}

p2p_test.go

@@ -0,0 +1,39 @@
+package shell
+
+import (
+	"context"
+	"testing"
+
+	"github.com/cheekybits/is"
+)
+
+func TestP2PListener(t *testing.T) {
+	is := is.New(t)
+	s := NewShell(shellUrl)
+
+	listener, err := s.P2POpenListener(context.Background(), "p2p-open-listener-test", "/ip4/127.0.0.1/udp/3000")
+	is.Nil(err)
+	is.Equal(listener.Address, "/ip4/127.0.0.1/udp/3000")
+	is.Equal(listener.Protocol, "/p2p/p2p-open-listener-test")
+
+	listenerList, err := s.P2PListListeners(context.Background())
+	is.Nil(err)
+	is.Equal(len(listenerList.Listeners), 1)
+	is.Equal(listenerList.Listeners[0].Address, "/ip4/127.0.0.1/udp/3000")
+	is.Equal(listenerList.Listeners[0].Protocol, "/p2p/p2p-open-listener-test")
+
+	is.Nil(s.P2PCloseListener(context.Background(), "p2p-open-listener-test", false))
+
+	listenerList, err = s.P2PListListeners(context.Background())
+	is.Nil(err)
+	is.Equal(len(listenerList.Listeners), 0)
+}
+
+func TestP2PStreams(t *testing.T) {
+	is := is.New(t)
+	s := NewShell(shellUrl)
+
+	streamsList, err := s.P2PListStreams(context.Background())
+	is.Nil(err)
+	is.Equal(len(streamsList.Streams), 0)
+}

shell.go

@@ -594,3 +594,16 @@ func (s *Shell) SwarmPeers(ctx context.Context) (*SwarmConnInfos, error) {
 	err := s.Request("swarm/peers").Exec(ctx, &v)
 	return v, err
 }
+
+type swarmConnection struct {
+	Strings []string
+}
+
+// SwarmConnect opens a swarm connection to a specific address.
+func (s *Shell) SwarmConnect(ctx context.Context, addr string) error {
+	var conn *swarmConnection
+	err := s.Request("swarm/connect").
+		Arguments(addr).
+		Exec(ctx, &conn)
+	return err
+}