Merge pull request #3402 from keks/feat/newfloodsub

use floodsub version 0.8.1

core/commands/pubsub.go

@@ -13,7 +13,7 @@ import (
 	cmds "github.com/ipfs/go-ipfs/commands"
 	core "github.com/ipfs/go-ipfs/core"
 
-	floodsub "gx/ipfs/QmWiLbk7eE1jGePDAuS26E2A9bMK3e3PMH3dcSeRY3MEBR/floodsub"
+	floodsub "gx/ipfs/QmRJs5veT3gnuYpLAagC3NbzixbkgwjSdUXTKfh3hMo6XM/floodsub"
 	pstore "gx/ipfs/QmXXCcQ7CLg5a81Ui9TTR35QcR4y7ZyihxwfjqaHfUVcVo/go-libp2p-peerstore"
 	u "gx/ipfs/Qmb912gdngC1UWwTkhuW8knyRbcWeu5kqkxBpveLmW8bSr/go-ipfs-util"
 	cid "gx/ipfs/QmcEcrBAMrwMyhSjXt4yfyPpzgSuV8HLHavnfmiKCSRqZU/go-cid"
@@ -77,7 +77,7 @@ To use, the daemon must be run with '--enable-pubsub-experiment'.
 		}
 
 		topic := req.Arguments()[0]
-		msgs, err := n.Floodsub.Subscribe(req.Context(), topic)
+		sub, err := n.Floodsub.Subscribe(topic)
 		if err != nil {
 			res.SetError(err, cmds.ErrNormal)
 			return
@@ -86,19 +86,19 @@ To use, the daemon must be run with '--enable-pubsub-experiment'.
 		out := make(chan interface{})
 		res.SetOutput((<-chan interface{})(out))
 
-		ctx := req.Context()
 		go func() {
+			defer sub.Cancel()
 			defer close(out)
 			for {
-				select {
-				case msg, ok := <-msgs:
-					if !ok {
-						return
-					}
-					out <- msg
-				case <-ctx.Done():
-					n.Floodsub.Unsub(topic)
+				msg, err := sub.Next(req.Context())
+				if err == io.EOF || err == context.Canceled {
+					break
+				} else if err != nil {
+					res.SetError(err, cmds.ErrNormal)
+					return
 				}
+
+				out <- msg
 			}
 		}()
 

core/core.go

@@ -38,6 +38,7 @@ import (
 	ft "github.com/ipfs/go-ipfs/unixfs"
 
 	swarm "gx/ipfs/QmNafAGBU21iQmLudMT2z1kqgEGhjUrNoK9a3v4azd8ei4/go-libp2p-swarm"
+	floodsub "gx/ipfs/QmRJs5veT3gnuYpLAagC3NbzixbkgwjSdUXTKfh3hMo6XM/floodsub"
 	goprocess "gx/ipfs/QmSF8fPo3jgVBAy8fpdjjYqgG87dkJgUprRBHRd2tmfgpP/goprocess"
 	mamask "gx/ipfs/QmSMZwvs3n4GBikZ7hKzT17c3bk65FmyZo2JqtJ16swqCv/multiaddr-filter"
 	logging "gx/ipfs/QmSpJByNKFX1sCsHBEp3R73FL4NF6FnQTEGyNAXHm2GS52/go-log"
@@ -45,7 +46,6 @@ import (
 	ma "gx/ipfs/QmUAQaWbKxGCUTuoQVvvicbQNZ9APF5pDGWyAZSe93AtKH/go-multiaddr"
 	routing "gx/ipfs/QmUrCwTDvJgmBbJVHu1HGEyqDaod3dR6sEkZkpxZk4u47c/go-libp2p-routing"
 	addrutil "gx/ipfs/QmVDnc2zvyQm8LhT72n22THcshvH7j3qPMnhvjerQER62T/go-addr-util"
-	floodsub "gx/ipfs/QmWiLbk7eE1jGePDAuS26E2A9bMK3e3PMH3dcSeRY3MEBR/floodsub"
 	metrics "gx/ipfs/QmX4j1JhubdEt4EB1JY1mMKTvJwPZSRzTv3uwh5zaDqyAi/go-libp2p-metrics"
 	pstore "gx/ipfs/QmXXCcQ7CLg5a81Ui9TTR35QcR4y7ZyihxwfjqaHfUVcVo/go-libp2p-peerstore"
 	discovery "gx/ipfs/QmZyBJGpRnbQ7oUstoGNZbhXC4HJuFUCgpp8pmsVTUwdS3/go-libp2p/p2p/discovery"

package.json

@@ -266,9 +266,9 @@
     },
     {
       "author": "whyrusleeping",
-      "hash": "QmWiLbk7eE1jGePDAuS26E2A9bMK3e3PMH3dcSeRY3MEBR",
+      "hash": "QmRJs5veT3gnuYpLAagC3NbzixbkgwjSdUXTKfh3hMo6XM",
       "name": "floodsub",
-      "version": "0.8.0"
+      "version": "0.8.1"
     },
     {
       "author": "whyrusleeping",