Merge pull request #4805 from ipfs/feat/coreapi/pubsub

coreapi: PubSub API

core/commands/pubsub.go

@@ -3,23 +3,16 @@ package commands
 import (
 	"context"
 	"encoding/binary"
-	"errors"
 	"fmt"
 	"io"
 	"net/http"
 	"sort"
-	"sync"
-	"time"
 
-	core "github.com/ipfs/go-ipfs/core"
 	cmdenv "github.com/ipfs/go-ipfs/core/commands/cmdenv"
 	e "github.com/ipfs/go-ipfs/core/commands/e"
+	options "github.com/ipfs/go-ipfs/core/coreapi/interface/options"
 
-	cid "gx/ipfs/QmPSQnBKM9g7BaUcZCvswUJVscQ1ipjmwxN5PXCjkp9EQ7/go-cid"
-	blocks "gx/ipfs/QmRcHuYzAyswytBuMF78rj3LTChYszomRFXNg4685ZN1WM/go-block-format"
-	pstore "gx/ipfs/QmSJ36wcYQyEViJUWUEhJU81tw1KdakTKqLLHbvYbA9zDv/go-libp2p-peerstore"
 	cmdkit "gx/ipfs/QmSP88ryZkHSRn1fnngAaV2Vcn63WUJzAavnRM9CVdU1Ky/go-ipfs-cmdkit"
-	floodsub "gx/ipfs/QmUK4h113Hh7bR2gPpsMcbUEbbzc7hspocmPi91Bmi69nH/go-libp2p-floodsub"
 	cmds "gx/ipfs/QmXTmUCBtDUrzDYVzASogLiNph7EBuYqEgPL7QoHNMzUnz/go-ipfs-cmds"
 )
 
@@ -48,6 +41,13 @@ const (
 	pubsubDiscoverOptionName = "discover"
 )
 
+type pubsubMessage struct {
+	From     []byte   `json:"from,omitempty"`
+	Data     []byte   `json:"data,omitempty"`
+	Seqno    []byte   `json:"seqno,omitempty"`
+	TopicIDs []string `json:"topicIDs,omitempty"`
+}
+
 var PubsubSubCmd = &cmds.Command{
 	Helptext: cmdkit.HelpText{
 		Tagline: "Subscribe to messages on a given topic.",
@@ -79,40 +79,19 @@ This command outputs data in the following encodings:
 		cmdkit.BoolOption(pubsubDiscoverOptionName, "try to discover other peers subscribed to the same topic"),
 	},
 	Run: func(req *cmds.Request, res cmds.ResponseEmitter, env cmds.Environment) error {
-		n, err := cmdenv.GetNode(env)
+		api, err := cmdenv.GetApi(env)
 		if err != nil {
 			return err
 		}
 
-		// Must be online!
-		if !n.OnlineMode() {
-			return cmdkit.Errorf(cmdkit.ErrClient, ErrNotOnline.Error())
-		}
-
-		if n.Floodsub == nil {
-			return fmt.Errorf("experimental pubsub feature not enabled. Run daemon with --enable-pubsub-experiment to use")
-		}
-
 		topic := req.Arguments[0]
-		sub, err := n.Floodsub.Subscribe(topic)
+		discover, _ := req.Options[pubsubDiscoverOptionName].(bool)
+
+		sub, err := api.PubSub().Subscribe(req.Context, topic, options.PubSub.Discover(discover))
 		if err != nil {
 			return err
 		}
-		defer sub.Cancel()
-
-		discover, _ := req.Options[pubsubDiscoverOptionName].(bool)
-		if discover {
-			go func() {
-				blk := blocks.NewBlock([]byte("floodsub:" + topic))
-				err := n.Blocks.AddBlock(blk)
-				if err != nil {
-					log.Error("pubsub discovery: ", err)
-					return
-				}
-
-				connectToPubSubPeers(req.Context, n, blk.Cid())
-			}()
-		}
+		defer sub.Close()
 
 		if f, ok := res.(http.Flusher); ok {
 			f.Flush()
@@ -126,15 +105,17 @@ This command outputs data in the following encodings:
 				return err
 			}
 
-			err = res.Emit(msg)
-			if err != nil {
-				return err
-			}
+			res.Emit(&pubsubMessage{
+				Data:     msg.Data(),
+				From:     []byte(msg.From()),
+				Seqno:    msg.Seq(),
+				TopicIDs: msg.Topics(),
+			})
 		}
 	},
 	Encoders: cmds.EncoderMap{
 		cmds.Text: cmds.MakeEncoder(func(req *cmds.Request, w io.Writer, v interface{}) error {
-			m, ok := v.(*floodsub.Message)
+			m, ok := v.(*pubsubMessage)
 			if !ok {
 				return fmt.Errorf("unexpected type: %T", v)
 			}
@@ -143,7 +124,7 @@ This command outputs data in the following encodings:
 			return err
 		}),
 		"ndpayload": cmds.MakeEncoder(func(req *cmds.Request, w io.Writer, v interface{}) error {
-			m, ok := v.(*floodsub.Message)
+			m, ok := v.(*pubsubMessage)
 			if !ok {
 				return fmt.Errorf("unexpected type: %T", v)
 			}
@@ -153,7 +134,7 @@ This command outputs data in the following encodings:
 			return err
 		}),
 		"lenpayload": cmds.MakeEncoder(func(req *cmds.Request, w io.Writer, v interface{}) error {
-			m, ok := v.(*floodsub.Message)
+			m, ok := v.(*pubsubMessage)
 			if !ok {
 				return fmt.Errorf("unexpected type: %T", v)
 			}
@@ -166,31 +147,7 @@ This command outputs data in the following encodings:
 			return err
 		}),
 	},
-	Type: floodsub.Message{},
-}
-
-func connectToPubSubPeers(ctx context.Context, n *core.IpfsNode, cid cid.Cid) {
-	ctx, cancel := context.WithCancel(ctx)
-	defer cancel()
-
-	provs := n.Routing.FindProvidersAsync(ctx, cid, 10)
-	wg := &sync.WaitGroup{}
-	for p := range provs {
-		wg.Add(1)
-		go func(pi pstore.PeerInfo) {
-			defer wg.Done()
-			ctx, cancel := context.WithTimeout(ctx, time.Second*10)
-			defer cancel()
-			err := n.PeerHost.Connect(ctx, pi)
-			if err != nil {
-				log.Info("pubsub discover: ", err)
-				return
-			}
-			log.Info("connected to pubsub peer:", pi.ID)
-		}(p)
-	}
-
-	wg.Wait()
+	Type: pubsubMessage{},
 }
 
 var PubsubPubCmd = &cmds.Command{
@@ -210,20 +167,11 @@ To use, the daemon must be run with '--enable-pubsub-experiment'.
 		cmdkit.StringArg("data", true, true, "Payload of message to publish.").EnableStdin(),
 	},
 	Run: func(req *cmds.Request, res cmds.ResponseEmitter, env cmds.Environment) error {
-		n, err := cmdenv.GetNode(env)
+		api, err := cmdenv.GetApi(env)
 		if err != nil {
 			return err
 		}
 
-		// Must be online!
-		if !n.OnlineMode() {
-			return cmdkit.Errorf(cmdkit.ErrClient, ErrNotOnline.Error())
-		}
-
-		if n.Floodsub == nil {
-			return errors.New("experimental pubsub feature not enabled. Run daemon with --enable-pubsub-experiment to use.")
-		}
-
 		topic := req.Arguments[0]
 
 		err = req.ParseBodyArgs()
@@ -232,7 +180,7 @@ To use, the daemon must be run with '--enable-pubsub-experiment'.
 		}
 
 		for _, data := range req.Arguments[1:] {
-			if err := n.Floodsub.Publish(topic, []byte(data)); err != nil {
+			if err := api.PubSub().Publish(req.Context, topic, []byte(data)); err != nil {
 				return err
 			}
 		}
@@ -254,21 +202,17 @@ To use, the daemon must be run with '--enable-pubsub-experiment'.
 `,
 	},
 	Run: func(req *cmds.Request, res cmds.ResponseEmitter, env cmds.Environment) error {
-		n, err := cmdenv.GetNode(env)
+		api, err := cmdenv.GetApi(env)
 		if err != nil {
 			return err
 		}
 
-		// Must be online!
-		if !n.OnlineMode() {
-			return cmdkit.Errorf(cmdkit.ErrClient, ErrNotOnline.Error())
-		}
-
-		if n.Floodsub == nil {
-			return errors.New("experimental pubsub feature not enabled. Run daemon with --enable-pubsub-experiment to use.")
+		l, err := api.PubSub().Ls(req.Context)
+		if err != nil {
+			return err
 		}
 
-		return cmds.EmitOnce(res, stringList{n.Floodsub.GetTopics()})
+		return cmds.EmitOnce(res, stringList{l})
 	},
 	Type: stringList{},
 	Encoders: cmds.EncoderMap{
@@ -308,26 +252,21 @@ To use, the daemon must be run with '--enable-pubsub-experiment'.
 		cmdkit.StringArg("topic", false, false, "topic to list connected peers of"),
 	},
 	Run: func(req *cmds.Request, res cmds.ResponseEmitter, env cmds.Environment) error {
-		n, err := cmdenv.GetNode(env)
+		api, err := cmdenv.GetApi(env)
 		if err != nil {
 			return err
 		}
 
-		// Must be online!
-		if !n.OnlineMode() {
-			return cmdkit.Errorf(cmdkit.ErrClient, ErrNotOnline.Error())
-		}
-
-		if n.Floodsub == nil {
-			return errors.New("experimental pubsub feature not enabled. Run daemon with --enable-pubsub-experiment to use")
-		}
-
 		var topic string
 		if len(req.Arguments) == 1 {
 			topic = req.Arguments[0]
 		}
 
-		peers := n.Floodsub.ListPeers(topic)
+		peers, err := api.PubSub().Peers(req.Context, options.PubSub.Topic(topic))
+		if err != nil {
+			return err
+		}
+
 		list := &stringList{make([]string, 0, len(peers))}
 
 		for _, peer := range peers {

core/coreapi/coreapi.go

@@ -16,8 +16,12 @@ package coreapi
 import (
 	core "github.com/ipfs/go-ipfs/core"
 	coreiface "github.com/ipfs/go-ipfs/core/coreapi/interface"
+
+	logging "gx/ipfs/QmZChCsSt8DctjceaL56Eibc29CVQq4dGKRXC5JRZ6Ppae/go-log"
 )
 
+var log = logging.Logger("core/coreapi")
+
 type CoreAPI struct {
 	node *core.IpfsNode
 }
@@ -72,3 +76,8 @@ func (api *CoreAPI) Dht() coreiface.DhtAPI {
 func (api *CoreAPI) Swarm() coreiface.SwarmAPI {
 	return (*SwarmAPI)(api)
 }
+
+// PubSub returns the PubSubAPI interface implementation backed by the go-ipfs node
+func (api *CoreAPI) PubSub() coreiface.PubSubAPI {
+	return (*PubSubAPI)(api)
+}

core/coreapi/interface/coreapi.go

@@ -37,6 +37,9 @@ type CoreAPI interface {
 	// Swarm returns an implementation of Swarm API
 	Swarm() SwarmAPI
 
+	// PubSub returns an implementation of PubSub API
+	PubSub() PubSubAPI
+
 	// ResolvePath resolves the path using Unixfs resolver
 	ResolvePath(context.Context, Path) (ResolvedPath, error)
 

core/coreapi/interface/errors.go

@@ -4,5 +4,5 @@ import "errors"
 
 var (
 	ErrIsDir   = errors.New("object is a directory")
-	ErrOffline = errors.New("can't resolve, ipfs node is offline")
+	ErrOffline = errors.New("this action must be run in online mode, try running 'ipfs daemon' first")
 )

core/coreapi/interface/options/pubsub.go

+package options
+
+type PubSubPeersSettings struct {
+	Topic string
+}
+
+type PubSubSubscribeSettings struct {
+	Discover bool
+}
+
+type PubSubPeersOption func(*PubSubPeersSettings) error
+type PubSubSubscribeOption func(*PubSubSubscribeSettings) error
+
+func PubSubPeersOptions(opts ...PubSubPeersOption) (*PubSubPeersSettings, error) {
+	options := &PubSubPeersSettings{
+		Topic: "",
+	}
+
+	for _, opt := range opts {
+		err := opt(options)
+		if err != nil {
+			return nil, err
+		}
+	}
+	return options, nil
+}
+
+func PubSubSubscribeOptions(opts ...PubSubSubscribeOption) (*PubSubSubscribeSettings, error) {
+	options := &PubSubSubscribeSettings{
+		Discover: false,
+	}
+
+	for _, opt := range opts {
+		err := opt(options)
+		if err != nil {
+			return nil, err
+		}
+	}
+	return options, nil
+}
+
+type pubsubOpts struct{}
+
+var PubSub pubsubOpts
+
+func (pubsubOpts) Topic(topic string) PubSubPeersOption {
+	return func(settings *PubSubPeersSettings) error {
+		settings.Topic = topic
+		return nil
+	}
+}
+
+func (pubsubOpts) Discover(discover bool) PubSubSubscribeOption {
+	return func(settings *PubSubSubscribeSettings) error {
+		settings.Discover = discover
+		return nil
+	}
+}

core/coreapi/interface/pubsub.go

+package iface
+
+import (
+	"context"
+	"io"
+
+	options "github.com/ipfs/go-ipfs/core/coreapi/interface/options"
+
+	peer "gx/ipfs/QmbNepETomvmXfz1X5pHNFD2QuPqnqi47dTd94QJWSorQ3/go-libp2p-peer"
+)
+
+// PubSubSubscription is an active PubSub subscription
+type PubSubSubscription interface {
+	io.Closer
+
+	// Next return the next incoming message
+	Next(context.Context) (PubSubMessage, error)
+}
+
+// PubSubMessage is a single PubSub message
+type PubSubMessage interface {
+	// From returns id of a peer from which the message has arrived
+	From() peer.ID
+
+	// Data returns the message body
+	Data() []byte
+
+	// Seq returns message identifier
+	Seq() []byte
+
+	// Topics returns list of topics this message was set to
+	Topics() []string
+}
+
+// PubSubAPI specifies the interface to PubSub
+type PubSubAPI interface {
+	// Ls lists subscribed topics by name
+	Ls(context.Context) ([]string, error)
+
+	// Peers list peers we are currently pubsubbing with
+	Peers(context.Context, ...options.PubSubPeersOption) ([]peer.ID, error)
+
+	// Publish a message to a given pubsub topic
+	Publish(context.Context, string, []byte) error
+
+	// Subscribe to messages on a given topic
+	Subscribe(context.Context, string, ...options.PubSubSubscribeOption) (PubSubSubscription, error)
+}

core/coreapi/pubsub.go

+package coreapi
+
+import (
+	"context"
+	"errors"
+	"strings"
+	"sync"
+	"time"
+
+	core "github.com/ipfs/go-ipfs/core"
+	coreiface "github.com/ipfs/go-ipfs/core/coreapi/interface"
+	caopts "github.com/ipfs/go-ipfs/core/coreapi/interface/options"
+
+	cid "gx/ipfs/QmPSQnBKM9g7BaUcZCvswUJVscQ1ipjmwxN5PXCjkp9EQ7/go-cid"
+	pstore "gx/ipfs/QmSJ36wcYQyEViJUWUEhJU81tw1KdakTKqLLHbvYbA9zDv/go-libp2p-peerstore"
+	floodsub "gx/ipfs/QmUK4h113Hh7bR2gPpsMcbUEbbzc7hspocmPi91Bmi69nH/go-libp2p-floodsub"
+	peer "gx/ipfs/QmbNepETomvmXfz1X5pHNFD2QuPqnqi47dTd94QJWSorQ3/go-libp2p-peer"
+)
+
+type PubSubAPI CoreAPI
+
+type pubSubSubscription struct {
+	cancel       context.CancelFunc
+	subscription *floodsub.Subscription
+}
+
+type pubSubMessage struct {
+	msg *floodsub.Message
+}
+
+func (api *PubSubAPI) Ls(ctx context.Context) ([]string, error) {
+	if err := api.checkNode(); err != nil {
+		return nil, err
+	}
+
+	return api.node.Floodsub.GetTopics(), nil
+}
+
+func (api *PubSubAPI) Peers(ctx context.Context, opts ...caopts.PubSubPeersOption) ([]peer.ID, error) {
+	if err := api.checkNode(); err != nil {
+		return nil, err
+	}
+
+	settings, err := caopts.PubSubPeersOptions(opts...)
+	if err != nil {
+		return nil, err
+	}
+
+	peers := api.node.Floodsub.ListPeers(settings.Topic)
+	out := make([]peer.ID, len(peers))
+
+	for i, peer := range peers {
+		out[i] = peer
+	}
+
+	return out, nil
+}
+
+func (api *PubSubAPI) Publish(ctx context.Context, topic string, data []byte) error {
+	if err := api.checkNode(); err != nil {
+		return err
+	}
+
+	return api.node.Floodsub.Publish(topic, data)
+}
+
+func (api *PubSubAPI) Subscribe(ctx context.Context, topic string, opts ...caopts.PubSubSubscribeOption) (coreiface.PubSubSubscription, error) {
+	options, err := caopts.PubSubSubscribeOptions(opts...)
+
+	if err := api.checkNode(); err != nil {
+		return nil, err
+	}
+
+	sub, err := api.node.Floodsub.Subscribe(topic)
+	if err != nil {
+		return nil, err
+	}
+
+	pubctx, cancel := context.WithCancel(api.node.Context())
+
+	if options.Discover {
+		go func() {
+			blk, err := api.core().Block().Put(pubctx, strings.NewReader("floodsub:"+topic))
+			if err != nil {
+				log.Error("pubsub discovery: ", err)
+				return
+			}
+
+			connectToPubSubPeers(pubctx, api.node, blk.Path().Cid())
+		}()
+	}
+
+	return &pubSubSubscription{cancel, sub}, nil
+}
+
+func connectToPubSubPeers(ctx context.Context, n *core.IpfsNode, cid cid.Cid) {
+	ctx, cancel := context.WithCancel(ctx)
+	defer cancel()
+
+	provs := n.Routing.FindProvidersAsync(ctx, cid, 10)
+	var wg sync.WaitGroup
+	for p := range provs {
+		wg.Add(1)
+		go func(pi pstore.PeerInfo) {
+			defer wg.Done()
+			ctx, cancel := context.WithTimeout(ctx, time.Second*10)
+			defer cancel()
+			err := n.PeerHost.Connect(ctx, pi)
+			if err != nil {
+				log.Info("pubsub discover: ", err)
+				return
+			}
+			log.Info("connected to pubsub peer:", pi.ID)
+		}(p)
+	}
+
+	wg.Wait()
+}
+
+func (api *PubSubAPI) checkNode() error {
+	if !api.node.OnlineMode() {
+		return coreiface.ErrOffline
+	}
+
+	if api.node.Floodsub == nil {
+		return errors.New("experimental pubsub feature not enabled. Run daemon with --enable-pubsub-experiment to use.")
+	}
+
+	return nil
+}
+
+func (sub *pubSubSubscription) Close() error {
+	sub.cancel()
+	sub.subscription.Cancel()
+	return nil
+}
+
+func (sub *pubSubSubscription) Next(ctx context.Context) (coreiface.PubSubMessage, error) {
+	msg, err := sub.subscription.Next(ctx)
+	if err != nil {
+		return nil, err
+	}
+
+	return &pubSubMessage{msg}, nil
+}
+
+func (msg *pubSubMessage) From() peer.ID {
+	return peer.ID(msg.msg.From)
+}
+
+func (msg *pubSubMessage) Data() []byte {
+	return msg.msg.Data
+}
+
+func (msg *pubSubMessage) Seq() []byte {
+	return msg.msg.Seqno
+}
+
+func (msg *pubSubMessage) Topics() []string {
+	return msg.msg.TopicIDs
+}
+
+func (api *PubSubAPI) core() coreiface.CoreAPI {
+	return (*CoreAPI)(api)
+}

core/coreapi/pubsub_test.go

+package coreapi_test
+
+import (
+	"context"
+	"github.com/ipfs/go-ipfs/core/coreapi/interface/options"
+	"testing"
+	"time"
+)
+
+func TestBasicPubSub(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	nds, apis, err := makeAPISwarm(ctx, true, 2)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	sub, err := apis[0].PubSub().Subscribe(ctx, "testch")
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	go func() {
+		tick := time.Tick(100 * time.Millisecond)
+
+		for {
+			err = apis[1].PubSub().Publish(ctx, "testch", []byte("hello world"))
+			if err != nil {
+				t.Fatal(err)
+			}
+			select {
+			case <-tick:
+			case <-ctx.Done():
+				return
+			}
+		}
+	}()
+
+	m, err := sub.Next(ctx)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if string(m.Data()) != "hello world" {
+		t.Errorf("got invalid data: %s", string(m.Data()))
+	}
+
+	if m.From() != nds[1].Identity {
+		t.Errorf("m.From didn't match")
+	}
+
+	peers, err := apis[1].PubSub().Peers(ctx, options.PubSub.Topic("testch"))
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if len(peers) != 1 {
+		t.Fatalf("got incorrect number of peers: %d", len(peers))
+	}
+
+	if peers[0] != nds[0].Identity {
+		t.Errorf("peer didn't match")
+	}
+
+	peers, err = apis[1].PubSub().Peers(ctx, options.PubSub.Topic("nottestch"))
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if len(peers) != 0 {
+		t.Fatalf("got incorrect number of peers: %d", len(peers))
+	}
+
+	topics, err := apis[0].PubSub().Ls(ctx)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if len(topics) != 1 {
+		t.Fatalf("got incorrect number of topics: %d", len(peers))
+	}
+
+	if topics[0] != "testch" {
+		t.Errorf("topic didn't match")
+	}
+
+	topics, err = apis[1].PubSub().Ls(ctx)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if len(topics) != 0 {
+		t.Fatalf("got incorrect number of topics: %d", len(peers))
+	}
+}

core/coreapi/unixfs_test.go

@@ -94,6 +94,9 @@ func makeAPISwarm(ctx context.Context, fullIdentity bool, n int) ([]*core.IpfsNo
 			Repo:   r,
 			Host:   mock.MockHostOption(mn),
 			Online: fullIdentity,
+			ExtraOpts: map[string]bool{
+				"pubsub": true,
+			},
 		})
 		if err != nil {
 			return nil, nil, err