@cryptotaxi247 / kubo / commits / e45df729b

namesys/pubsub: publisher and resolver

Commits: namesys: pubsub Publisher and Resolver namesys/pubsub: pacify code climate. namesys/pubsub: timeout for rendezvous namesys/pubsub: filter self in bootstrap connections namesys/pubsub: Publish to the correct topic License: MIT Signed-off-by: vyzo <vyzo@hackzen.org> namesys/pubsub: unit test Commits: namesys/pubsub: test namesys/pubsub_test: pacify code climate namesys/pubsub: update test to use extant mock routing License: MIT Signed-off-by: vyzo <vyzo@hackzen.org> namesys/pubsub: integrate namesys pubsub namesys: integrate pubsub resolvers namesys/pubsub_test: tweak delays - trying to make travis happy. namesys/pubsub: fix duplicate bootstraps - subscription key is topic, not ipnskey. namesys/pubsub: no warning needed on cancellation namesys/pubsub: warning for receive errors - and more informative error messages at that. namesys/pubsub_test: smaller test - make it work with seemingly low fdlimits in travis/macosx. also, more informative test failures. namesys/pubsub: add delay to let pubsub perform handshake namesys/pubsub: update gx imports namesys/pubsub_test: preconnect publisher, reduce delays - preconnects the publisher to the receivers in order to avoid bootstrap flakiness with connectivity problems in travis. reduces sleeps to 1s for flood propagation (3s seems excessive with 5 hosts). namesys/pubsub: drop named return values in resolveOnce - per review comment. namesys/pubsub: check errors namesys/pubsub: store bytes in resolver datastore namesys/pubsub: resolver Cancel - for canceling subscriptions, pre whyrusleeping's request. namesys/pubsub: fix resolution without /ipns prefix - also improve the logging a bit. namesys/pubsub: don't resolve own keys through pubsub namesys/pubsub: signal ErrResolveFailed on resolution failure namesys/pubsub: use sync datastore, resolver lock only for subs namesys/pubsub_test: coverage for Cancel License: MIT Signed-off-by: vyzo <vyzo@hackzen.org> namesys/pubsub: parallelize dht and pubsub publishing Commits: namesys/pubsub: code cosmetics namesys: parallelize publishing with dht and pubsub namesys/pubsub: periodically reprovide topic rendezvous namesys/pubsub: cancelation for rendezvous goroutine namesys/pubsub: log ipns record seqno on publish License: MIT Signed-off-by: vyzo <vyzo@hackzen.org> namesys/pubsub: error checking License: MIT Signed-off-by: vyzo <vyzo@hackzen.org> namesys/pubsub: --enable-namesys-pubsub option and management Commits: package.json: update go-libp2p-blankhost namesys: fix stale package imports update go-testutil namesys/pubsub: reduce bootstrap provide period to 8hr namesys/pubsub: try to extract the key from id first option to enable ipns pubsub: --enable-namesys-pubsub ipfs name pubsub management subcommands corehttp/gateway_test: mockNamesys needs to implement GetResolver pacify code climate License: MIT Signed-off-by: vyzo <vyzo@hackzen.org> namesys/pubsub: pubsub sharness test test/sharness: test for ipns pubsub namesys/pubsub: return boolean indicator on Cancel package.json: remove duplicate entry for go-testutil update gx deps, testutil to 1.1.12 fix jenkins failure: use tabs in t0183-namesys-pubsub t0183: use 4 spaces for tabification License: MIT Signed-off-by: vyzo <vyzo@hackzen.org> namesys/pubsub: update for new command interface License: MIT Signed-off-by: vyzo <vyzo@hackzen.org> namesys/pubsub: fix sharness test for broken MacOS echo echo -n "" should print -n, but hey it's a mac. License: MIT Signed-off-by: vyzo <vyzo@hackzen.org>

vyzo committed Jul 7, 2017 at 18:32 UTC e45df729bea560b879e69a6200755be797d63c74
13 files changed +1008 -26
cmd/ipfs/daemon.go
+4
@@ -46,6 +46,7 @@ const (
46 unrestrictedApiAccessKwd = "unrestricted-api"
47 writableKwd = "writable"
48 enableFloodSubKwd = "enable-pubsub-experiment"
49 + enableIPNSPubSubKwd = "enable-namesys-pubsub"
50 enableMultiplexKwd = "enable-mplex-experiment"
51 // apiAddrKwd = "address-api"
52 // swarmAddrKwd = "address-swarm"
@@ -157,6 +158,7 @@ Headers.
158 cmdkit.BoolOption(offlineKwd, "Run offline. Do not connect to the rest of the network but provide local API."),
159 cmdkit.BoolOption(migrateKwd, "If true, assume yes at the migrate prompt. If false, assume no."),
160 cmdkit.BoolOption(enableFloodSubKwd, "Instantiate the ipfs daemon with the experimental pubsub feature enabled."),
161 + cmdkit.BoolOption(enableIPNSPubSubKwd, "Enable IPNS record distribution through pubsub; enables pubsub."),
162 cmdkit.BoolOption(enableMultiplexKwd, "Add the experimental 'go-multiplex' stream muxer to libp2p on construction.").WithDefault(true),
163
164 // TODO: add way to override addresses. tricky part: updating the config if also --init.
@@ -283,6 +285,7 @@ func daemonFunc(req cmds.Request, re cmds.ResponseEmitter) {
285
286 offline, _, _ := req.Option(offlineKwd).Bool()
287 pubsub, _, _ := req.Option(enableFloodSubKwd).Bool()
288 + ipnsps, _, _ := req.Option(enableIPNSPubSubKwd).Bool()
289 mplex, _, _ := req.Option(enableMultiplexKwd).Bool()
290
291 // Start assembling node config
@@ -292,6 +295,7 @@ func daemonFunc(req cmds.Request, re cmds.ResponseEmitter) {
295 Online: !offline,
296 ExtraOpts: map[string]bool{
297 "pubsub": pubsub,
298 + "ipnsps": ipnsps,
299 "mplex": mplex,
300 },
301 //TODO(Kubuxu): refactor Online vs Offline by adding Permanent vs Ephemeral
core/builder.go
+1 -1
@@ -210,7 +210,7 @@ func setupNode(ctx context.Context, n *IpfsNode, cfg *BuildCfg) error {
210
211 if cfg.Online {
212 do := setupDiscoveryOption(rcfg.Discovery)
213 - if err := n.startOnlineServices(ctx, cfg.Routing, cfg.Host, do, cfg.getOpt("pubsub"), cfg.getOpt("mplex")); err != nil {
213 + if err := n.startOnlineServices(ctx, cfg.Routing, cfg.Host, do, cfg.getOpt("pubsub"), cfg.getOpt("ipnsps"), cfg.getOpt("mplex")); err != nil {
214 return err
215 }
216 } else {
core/commands/ipnsps.go new
+163
@@ -0,0 +1,163 @@
1 +package commands
2 +
3 +import (
4 + "errors"
5 + "fmt"
6 + "io"
7 + "strings"
8 +
9 + cmds "github.com/ipfs/go-ipfs/commands"
10 + e "github.com/ipfs/go-ipfs/core/commands/e"
11 + ns "github.com/ipfs/go-ipfs/namesys"
12 +
13 + cmdkit "gx/ipfs/QmUyfy4QSr3NXym4etEiRyxBLqqAeKHJuRdi8AACxg63fZ/go-ipfs-cmdkit"
14 +)
15 +
16 +type ipnsPubsubState struct {
17 + Enabled bool
18 +}
19 +
20 +type ipnsPubsubCancel struct {
21 + Canceled bool
22 +}
23 +
24 +// IpnsPubsubCmd is the subcommand that allows us to manage the IPNS pubsub system
25 +var IpnsPubsubCmd = &cmds.Command{
26 + Helptext: cmdkit.HelpText{
27 + Tagline: "IPNS pubsub management",
28 + ShortDescription: `
29 +Manage and inspect the state of the IPNS pubsub resolver.
30 +
31 +Note: this command is experimental and subject to change as the system is refined
32 +`,
33 + },
34 + Subcommands: map[string]*cmds.Command{
35 + "state": ipnspsStateCmd,
36 + "subs": ipnspsSubsCmd,
37 + "cancel": ipnspsCancelCmd,
38 + },
39 +}
40 +
41 +var ipnspsStateCmd = &cmds.Command{
42 + Helptext: cmdkit.HelpText{
43 + Tagline: "Query the state of IPNS pubsub",
44 + },
45 + Run: func(req cmds.Request, res cmds.Response) {
46 + n, err := req.InvocContext().GetNode()
47 + if err != nil {
48 + res.SetError(err, cmdkit.ErrNormal)
49 + return
50 + }
51 +
52 + _, ok := n.Namesys.GetResolver("pubsub")
53 + res.SetOutput(&ipnsPubsubState{ok})
54 + },
55 + Type: ipnsPubsubState{},
56 + Marshalers: cmds.MarshalerMap{
57 + cmds.Text: func(res cmds.Response) (io.Reader, error) {
58 + v, err := unwrapOutput(res.Output())
59 + if err != nil {
60 + return nil, err
61 + }
62 +
63 + output, ok := v.(*ipnsPubsubState)
64 + if !ok {
65 + return nil, e.TypeErr(output, v)
66 + }
67 +
68 + var state string
69 + if output.Enabled {
70 + state = "enabled"
71 + } else {
72 + state = "disabled"
73 + }
74 +
75 + return strings.NewReader(state + "\n"), nil
76 + },
77 + },
78 +}
79 +
80 +var ipnspsSubsCmd = &cmds.Command{
81 + Helptext: cmdkit.HelpText{
82 + Tagline: "Show current name subscriptions",
83 + },
84 + Run: func(req cmds.Request, res cmds.Response) {
85 + n, err := req.InvocContext().GetNode()
86 + if err != nil {
87 + res.SetError(err, cmdkit.ErrNormal)
88 + return
89 + }
90 +
91 + r, ok := n.Namesys.GetResolver("pubsub")
92 + if !ok {
93 + res.SetError(errors.New("IPNS pubsub subsystem is not enabled"), cmdkit.ErrClient)
94 + return
95 + }
96 +
97 + psr, ok := r.(*ns.PubsubResolver)
98 + if !ok {
99 + res.SetError(fmt.Errorf("unexpected resolver type: %v", r), cmdkit.ErrNormal)
100 + return
101 + }
102 +
103 + res.SetOutput(&stringList{psr.GetSubscriptions()})
104 + },
105 + Type: stringList{},
106 + Marshalers: cmds.MarshalerMap{
107 + cmds.Text: stringListMarshaler,
108 + },
109 +}
110 +
111 +var ipnspsCancelCmd = &cmds.Command{
112 + Helptext: cmdkit.HelpText{
113 + Tagline: "Cancel a name subscription",
114 + },
115 + Run: func(req cmds.Request, res cmds.Response) {
116 + n, err := req.InvocContext().GetNode()
117 + if err != nil {
118 + res.SetError(err, cmdkit.ErrNormal)
119 + return
120 + }
121 +
122 + r, ok := n.Namesys.GetResolver("pubsub")
123 + if !ok {
124 + res.SetError(errors.New("IPNS pubsub subsystem is not enabled"), cmdkit.ErrClient)
125 + return
126 + }
127 +
128 + psr, ok := r.(*ns.PubsubResolver)
129 + if !ok {
130 + res.SetError(fmt.Errorf("unexpected resolver type: %v", r), cmdkit.ErrNormal)
131 + return
132 + }
133 +
134 + ok = psr.Cancel(req.Arguments()[0])
135 + res.SetOutput(&ipnsPubsubCancel{ok})
136 + },
137 + Arguments: []cmdkit.Argument{
138 + cmdkit.StringArg("name", true, false, "Name to cancel the subscription for."),
139 + },
140 + Type: ipnsPubsubCancel{},
141 + Marshalers: cmds.MarshalerMap{
142 + cmds.Text: func(res cmds.Response) (io.Reader, error) {
143 + v, err := unwrapOutput(res.Output())
144 + if err != nil {
145 + return nil, err
146 + }
147 +
148 + output, ok := v.(*ipnsPubsubCancel)
149 + if !ok {
150 + return nil, e.TypeErr(output, v)
151 + }
152 +
153 + var state string
154 + if output.Canceled {
155 + state = "canceled"
156 + } else {
157 + state = "no subscription"
158 + }
159 +
160 + return strings.NewReader(state + "\n"), nil
161 + },
162 + },
163 +}
core/commands/mount_windows.go
+1 -1
@@ -5,7 +5,7 @@ import (
5
6 cmds "github.com/ipfs/go-ipfs/commands"
7
8 - cmdkit "gx/ipfs/QmSNbH2A1evCCbJSDC6u3RV3GGDhgu6pRGbXHvrN89tMKf/go-ipfs-cmdkit"
8 + cmdkit "gx/ipfs/QmUyfy4QSr3NXym4etEiRyxBLqqAeKHJuRdi8AACxg63fZ/go-ipfs-cmdkit"
9 )
10
11 var MountCmd = &cmds.Command{
core/commands/name.go
+1
@@ -63,5 +63,6 @@ Resolve the value of a dnslink:
63 Subcommands: map[string]*cmds.Command{
64 "publish": PublishCmd,
65 "resolve": IpnsCmd,
66 + "pubsub": IpnsPubsubCmd,
67 },
68 }
core/core.go
+9 -2
@@ -152,7 +152,7 @@ type Mounts struct {
152 Ipns mount.Mount
153 }
154
155 -func (n *IpfsNode) startOnlineServices(ctx context.Context, routingOption RoutingOption, hostOption HostOption, do DiscoveryOption, pubsub, mplex bool) error {
155 +func (n *IpfsNode) startOnlineServices(ctx context.Context, routingOption RoutingOption, hostOption HostOption, do DiscoveryOption, pubsub, ipnsps, mplex bool) error {
156
157 if n.PeerHost != nil { // already online.
158 return errors.New("node already online")
@@ -249,10 +249,17 @@ func (n *IpfsNode) startOnlineServices(ctx context.Context, routingOption Routin
249 return err
250 }
251
252 - if pubsub {
252 + if pubsub || ipnsps {
253 n.Floodsub = floodsub.NewFloodSub(ctx, peerhost)
254 }
255
256 + if ipnsps {
257 + err = namesys.AddPubsubNameSystem(ctx, n.Namesys, n.PeerHost, n.Routing, n.Repo.Datastore(), n.Floodsub)
258 + if err != nil {
259 + return err
260 + }
261 + }
262 +
263 n.P2P = p2p.NewP2P(n.Identity, n.PeerHost, n.Peerstore)
264
265 // setup local discovery
core/corehttp/gateway_test.go
+4
@@ -48,6 +48,10 @@ func (m mockNamesys) PublishWithEOL(ctx context.Context, name ci.PrivKey, value
48 return errors.New("not implemented for mockNamesys")
49 }
50
51 +func (m mockNamesys) GetResolver(subs string) (namesys.Resolver, bool) {
52 + return nil, false
53 +}
54 +
55 func newNodeWithMockNamesys(ns mockNamesys) (*core.IpfsNode, error) {
56 c := config.Config{
57 Identity: config.Identity{
namesys/interface.go
+8
@@ -70,6 +70,7 @@ var ErrPublishFailed = errors.New("Could not publish name.")
70 type NameSystem interface {
71 Resolver
72 Publisher
73 + ResolverLookup
74 }
75
76 // Resolver is an object capable of resolving names.
@@ -112,3 +113,10 @@ type Publisher interface {
113 // call once the records spec is implemented
114 PublishWithEOL(ctx context.Context, name ci.PrivKey, value path.Path, eol time.Time) error
115 }
116 +
117 +// ResolverLookup is an object capable of finding resolvers for a subsystem
118 +type ResolverLookup interface {
119 +
120 + // GetResolver retrieves a resolver associated with a subsystem
121 + GetResolver(subs string) (Resolver, bool)
122 +}
namesys/namesys.go
+118 -20
@@ -2,14 +2,20 @@ package namesys
2
3 import (
4 "context"
5 + "errors"
6 "strings"
7 + "sync"
8 "time"
9
10 path "github.com/ipfs/go-ipfs/path"
11
12 routing "gx/ipfs/QmPR2JzfKd9poHx9XBhzoFeBBC31ZM3W5iUPKJZWyaoZZm/go-libp2p-routing"
13 + p2phost "gx/ipfs/QmRS46AyqtpJBsf1zmQdeizSDEzo1qkWR7rdEuPFAv8237/go-libp2p-host"
14 + mh "gx/ipfs/QmU9a9NV9RdPNwZQDYd5uKsm6N6LJLSvLbywDDYFbaaC6P/go-multihash"
15 + floodsub "gx/ipfs/QmVNv1WV6XxzQV4MBuiLX5729wMazaf8TNzm2Sq6ejyHh7/go-libp2p-floodsub"
16 ds "gx/ipfs/QmVSase1JP7cq9QkPT46oNwdp9pT6kBkG3oqS14y3QcZjG/go-datastore"
17 peer "gx/ipfs/QmXYjuNuxVzXKJCfWasQk1RqkhVLDM9jtUKhqc2WPQmFSB/go-libp2p-peer"
18 + isd "gx/ipfs/QmZmmuAXgX73UQmX1jRKjTGmjzq24Jinqkq8vzkBtno4uX/go-is-domain"
19 ci "gx/ipfs/QmaPbCnUMBohSGo3KnxEa2bHqyJVVeEEcwtqJAYxerieBo/go-libp2p-crypto"
20 )
21
@@ -36,11 +42,28 @@ func NewNameSystem(r routing.ValueStore, ds ds.Datastore, cachesize int) NameSys
42 "dht": NewRoutingResolver(r, cachesize),
43 },
44 publishers: map[string]Publisher{
39 - "/ipns/": NewRoutingPublisher(r, ds),
45 + "dht": NewRoutingPublisher(r, ds),
46 },
47 }
48 }
49
50 +// AddPubsubNameSystem adds the pubsub publisher and resolver to the namesystem
51 +func AddPubsubNameSystem(ctx context.Context, ns NameSystem, host p2phost.Host, r routing.IpfsRouting, ds ds.Datastore, ps *floodsub.PubSub) error {
52 + mpns, ok := ns.(*mpns)
53 + if !ok {
54 + return errors.New("unexpected NameSystem; not an mpns instance")
55 + }
56 +
57 + pkf, ok := r.(routing.PubKeyFetcher)
58 + if !ok {
59 + return errors.New("unexpected IpfsRouting; not a PubKeyFetcher instance")
60 + }
61 +
62 + mpns.resolvers["pubsub"] = NewPubsubResolver(ctx, host, r, pkf, ps)
63 + mpns.publishers["pubsub"] = NewPubsubPublisher(ctx, host, ds, r, ps)
64 + return nil
65 +}
66 +
67 const DefaultResolverCacheTTL = time.Minute
68
69 // Resolve implements Resolver.
@@ -72,38 +95,100 @@ func (ns *mpns) resolveOnce(ctx context.Context, name string) (path.Path, error)
95 return "", ErrResolveFailed
96 }
97
75 - for protocol, resolver := range ns.resolvers {
76 - log.Debugf("Attempting to resolve %s with %s", segments[2], protocol)
77 - p, err := resolver.resolveOnce(ctx, segments[2])
78 - if err == nil {
79 - if len(segments) > 3 {
80 - return path.FromSegments("", strings.TrimRight(p.String(), "/"), segments[3])
81 - } else {
82 - return p, err
98 + makePath := func(p path.Path) (path.Path, error) {
99 + if len(segments) > 3 {
100 + return path.FromSegments("", strings.TrimRight(p.String(), "/"), segments[3])
101 + } else {
102 + return p, nil
103 + }
104 + }
105 +
106 + // Resolver selection:
107 + // 1. if it is a multihash resolve through "pubsub" (if available),
108 + // with fallback to "dht"
109 + // 2. if it is a domain name, resolve through "dns"
110 + // 3. otherwise resolve through the "proquint" resolver
111 + key := segments[2]
112 +
113 + _, err := mh.FromB58String(key)
114 + if err == nil {
115 + res, ok := ns.resolvers["pubsub"]
116 + if ok {
117 + p, err := res.resolveOnce(ctx, key)
118 + if err == nil {
119 + return makePath(p)
120 + }
121 + }
122 +
123 + res, ok = ns.resolvers["dht"]
124 + if ok {
125 + p, err := res.resolveOnce(ctx, key)
126 + if err == nil {
127 + return makePath(p)
128 + }
129 + }
130 +
131 + return "", ErrResolveFailed
132 + }
133 +
134 + if isd.IsDomain(key) {
135 + res, ok := ns.resolvers["dns"]
136 + if ok {
137 + p, err := res.resolveOnce(ctx, key)
138 + if err == nil {
139 + return makePath(p)
140 }
141 }
142 +
143 + return "", ErrResolveFailed
144 }
145 +
146 + res, ok := ns.resolvers["proquint"]
147 + if ok {
148 + p, err := res.resolveOnce(ctx, key)
149 + if err == nil {
150 + return makePath(p)
151 + }
152 +
153 + return "", ErrResolveFailed
154 + }
155 +
156 log.Warningf("No resolver found for %s", name)
157 return "", ErrResolveFailed
158 }
159
160 // Publish implements Publisher
161 func (ns *mpns) Publish(ctx context.Context, name ci.PrivKey, value path.Path) error {
92 - err := ns.publishers["/ipns/"].Publish(ctx, name, value)
93 - if err != nil {
94 - return err
95 - }
96 - ns.addToDHTCache(name, value, time.Now().Add(DefaultRecordTTL))
97 - return nil
162 + return ns.PublishWithEOL(ctx, name, value, time.Now().Add(DefaultRecordTTL))
163 }
164
165 func (ns *mpns) PublishWithEOL(ctx context.Context, name ci.PrivKey, value path.Path, eol time.Time) error {
101 - err := ns.publishers["/ipns/"].PublishWithEOL(ctx, name, value, eol)
102 - if err != nil {
103 - return err
166 + var dhtErr error
167 +
168 + wg := &sync.WaitGroup{}
169 + wg.Add(1)
170 + go func() {
171 + dhtErr = ns.publishers["dht"].PublishWithEOL(ctx, name, value, eol)
172 + if dhtErr == nil {
173 + ns.addToDHTCache(name, value, eol)
174 + }
175 + wg.Done()
176 + }()
177 +
178 + pub, ok := ns.publishers["pubsub"]
179 + if ok {
180 + wg.Add(1)
181 + go func() {
182 + err := pub.PublishWithEOL(ctx, name, value, eol)
183 + if err != nil {
184 + log.Warningf("error publishing %s with pubsub: %s", name, err.Error())
185 + }
186 + wg.Done()
187 + }()
188 }
105 - ns.addToDHTCache(name, value, eol)
106 - return nil
189 +
190 + wg.Wait()
191 + return dhtErr
192 }
193
194 func (ns *mpns) addToDHTCache(key ci.PrivKey, value path.Path, eol time.Time) {
@@ -138,3 +223,16 @@ func (ns *mpns) addToDHTCache(key ci.PrivKey, value path.Path, eol time.Time) {
223 eol: eol,
224 })
225 }
226 +
227 +// GetResolver implements ResolverLookup
228 +func (ns *mpns) GetResolver(subs string) (Resolver, bool) {
229 + res, ok := ns.resolvers[subs]
230 + if ok {
231 + ires, ok := res.(Resolver)
232 + if ok {
233 + return ires, true
234 + }
235 + }
236 +
237 + return nil, false
238 +}
namesys/namesys_test.go
+2 -2
@@ -58,8 +58,8 @@ func mockResolverTwo() *mockResolver {
58 func TestNamesysResolution(t *testing.T) {
59 r := &mpns{
60 resolvers: map[string]resolver{
61 - "one": mockResolverOne(),
62 - "two": mockResolverTwo(),
61 + "dht": mockResolverOne(),
62 + "dns": mockResolverTwo(),
63 },
64 }
65
namesys/pubsub.go new
+430
@@ -0,0 +1,430 @@
1 +package namesys
2 +
3 +import (
4 + "context"
5 + "errors"
6 + "fmt"
7 + "strings"
8 + "sync"
9 + "time"
10 +
11 + pb "github.com/ipfs/go-ipfs/namesys/pb"
12 + path "github.com/ipfs/go-ipfs/path"
13 + dshelp "github.com/ipfs/go-ipfs/thirdparty/ds-help"
14 +
15 + cid "gx/ipfs/QmNp85zy9RLrQ5oQD4hPyS39ezrrXpcaa7R4Y9kxdWQLLQ/go-cid"
16 + routing "gx/ipfs/QmPR2JzfKd9poHx9XBhzoFeBBC31ZM3W5iUPKJZWyaoZZm/go-libp2p-routing"
17 + pstore "gx/ipfs/QmPgDWmTmuzvP7QE5zwo1TmjbJme9pmZHNujB2453jkCTr/go-libp2p-peerstore"
18 + p2phost "gx/ipfs/QmRS46AyqtpJBsf1zmQdeizSDEzo1qkWR7rdEuPFAv8237/go-libp2p-host"
19 + u "gx/ipfs/QmSU6eubNdhXjFBJBSksTp8kv8YRub8mGAPv8tVJHmL2EU/go-ipfs-util"
20 + mh "gx/ipfs/QmU9a9NV9RdPNwZQDYd5uKsm6N6LJLSvLbywDDYFbaaC6P/go-multihash"
21 + floodsub "gx/ipfs/QmVNv1WV6XxzQV4MBuiLX5729wMazaf8TNzm2Sq6ejyHh7/go-libp2p-floodsub"
22 + ds "gx/ipfs/QmVSase1JP7cq9QkPT46oNwdp9pT6kBkG3oqS14y3QcZjG/go-datastore"
23 + dssync "gx/ipfs/QmVSase1JP7cq9QkPT46oNwdp9pT6kBkG3oqS14y3QcZjG/go-datastore/sync"
24 + peer "gx/ipfs/QmXYjuNuxVzXKJCfWasQk1RqkhVLDM9jtUKhqc2WPQmFSB/go-libp2p-peer"
25 + proto "gx/ipfs/QmZ4Qi3GaRbjcx28Sme5eMH7RQjGkt8wHxt2a65oLaeFEV/gogo-protobuf/proto"
26 + ci "gx/ipfs/QmaPbCnUMBohSGo3KnxEa2bHqyJVVeEEcwtqJAYxerieBo/go-libp2p-crypto"
27 + record "gx/ipfs/QmbxkgUceEcuSZ4ZdBA3x74VUDSSYjHYmmeEqkjxbtZ6Jg/go-libp2p-record"
28 + dhtpb "gx/ipfs/QmbxkgUceEcuSZ4ZdBA3x74VUDSSYjHYmmeEqkjxbtZ6Jg/go-libp2p-record/pb"
29 +)
30 +
31 +// PubsubPublisher is a publisher that distributes IPNS records through pubsub
32 +type PubsubPublisher struct {
33 + ctx context.Context
34 + ds ds.Datastore
35 + host p2phost.Host
36 + cr routing.ContentRouting
37 + ps *floodsub.PubSub
38 +
39 + mx sync.Mutex
40 + subs map[string]struct{}
41 +}
42 +
43 +// PubsubResolver is a resolver that receives IPNS records through pubsub
44 +type PubsubResolver struct {
45 + ctx context.Context
46 + ds ds.Datastore
47 + host p2phost.Host
48 + cr routing.ContentRouting
49 + pkf routing.PubKeyFetcher
50 + ps *floodsub.PubSub
51 +
52 + mx sync.Mutex
53 + subs map[string]*floodsub.Subscription
54 +}
55 +
56 +// NewPubsubPublisher constructs a new Publisher that publishes IPNS records through pubsub.
57 +// The constructor interface is complicated by the need to bootstrap the pubsub topic.
58 +// This could be greatly simplified if the pubsub implementation handled bootstrap itself
59 +func NewPubsubPublisher(ctx context.Context, host p2phost.Host, ds ds.Datastore, cr routing.ContentRouting, ps *floodsub.PubSub) *PubsubPublisher {
60 + return &PubsubPublisher{
61 + ctx: ctx,
62 + ds: ds,
63 + host: host, // needed for pubsub bootstrap
64 + cr: cr, // needed for pubsub bootstrap
65 + ps: ps,
66 + subs: make(map[string]struct{}),
67 + }
68 +}
69 +
70 +// NewPubsubResolver constructs a new Resolver that resolves IPNS records through pubsub.
71 +// same as above for pubsub bootstrap dependencies
72 +func NewPubsubResolver(ctx context.Context, host p2phost.Host, cr routing.ContentRouting, pkf routing.PubKeyFetcher, ps *floodsub.PubSub) *PubsubResolver {
73 + return &PubsubResolver{
74 + ctx: ctx,
75 + ds: dssync.MutexWrap(ds.NewMapDatastore()),
76 + host: host, // needed for pubsub bootstrap
77 + cr: cr, // needed for pubsub bootstrap
78 + pkf: pkf,
79 + ps: ps,
80 + subs: make(map[string]*floodsub.Subscription),
81 + }
82 +}
83 +
84 +// Publish publishes an IPNS record through pubsub with default TTL
85 +func (p *PubsubPublisher) Publish(ctx context.Context, k ci.PrivKey, value path.Path) error {
86 + return p.PublishWithEOL(ctx, k, value, time.Now().Add(DefaultRecordTTL))
87 +}
88 +
89 +// PublishWithEOL publishes an IPNS record through pubsub
90 +func (p *PubsubPublisher) PublishWithEOL(ctx context.Context, k ci.PrivKey, value path.Path, eol time.Time) error {
91 + id, err := peer.IDFromPrivateKey(k)
92 + if err != nil {
93 + return err
94 + }
95 +
96 + _, ipnskey := IpnsKeysForID(id)
97 +
98 + seqno, err := p.getPreviousSeqNo(ctx, ipnskey)
99 + if err != nil {
100 + return err
101 + }
102 +
103 + seqno++
104 +
105 + return p.publishRecord(ctx, k, value, seqno, eol, ipnskey, id)
106 +}
107 +
108 +func (p *PubsubPublisher) getPreviousSeqNo(ctx context.Context, ipnskey string) (uint64, error) {
109 + // the datastore is shared with the routing publisher to properly increment and persist
110 + // ipns record sequence numbers.
111 + prevrec, err := p.ds.Get(dshelp.NewKeyFromBinary([]byte(ipnskey)))
112 + if err != nil {
113 + if err == ds.ErrNotFound {
114 + // None found, lets start at zero!
115 + return 0, nil
116 + }
117 + return 0, err
118 + }
119 +
120 + prbytes, ok := prevrec.([]byte)
121 + if !ok {
122 + return 0, fmt.Errorf("unexpected type returned from datastore: %#v", prevrec)
123 + }
124 +
125 + var dsrec dhtpb.Record
126 + err = proto.Unmarshal(prbytes, &dsrec)
127 + if err != nil {
128 + return 0, err
129 + }
130 +
131 + var entry pb.IpnsEntry
132 + err = proto.Unmarshal(dsrec.GetValue(), &entry)
133 + if err != nil {
134 + return 0, err
135 + }
136 +
137 + return entry.GetSequence(), nil
138 +}
139 +
140 +func (p *PubsubPublisher) publishRecord(ctx context.Context, k ci.PrivKey, value path.Path, seqno uint64, eol time.Time, ipnskey string, ID peer.ID) error {
141 + entry, err := CreateRoutingEntryData(k, value, seqno, eol)
142 + if err != nil {
143 + return err
144 + }
145 +
146 + data, err := proto.Marshal(entry)
147 + if err != nil {
148 + return err
149 + }
150 +
151 + // the datastore is shared with the routing publisher to properly increment and persist
152 + // ipns record sequence numbers; so we need to Record our new entry in the datastore
153 + dsrec, err := record.MakePutRecord(k, ipnskey, data, true)
154 + if err != nil {
155 + return err
156 + }
157 +
158 + dsdata, err := proto.Marshal(dsrec)
159 + if err != nil {
160 + return err
161 + }
162 +
163 + err = p.ds.Put(dshelp.NewKeyFromBinary([]byte(ipnskey)), dsdata)
164 + if err != nil {
165 + return err
166 + }
167 +
168 + // now we publish, but we also need to bootstrap pubsub for our messages to propagate
169 + topic := "/ipns/" + ID.Pretty()
170 +
171 + p.mx.Lock()
172 + _, ok := p.subs[topic]
173 +
174 + if !ok {
175 + p.subs[topic] = struct{}{}
176 + p.mx.Unlock()
177 +
178 + bootstrapPubsub(p.ctx, p.cr, p.host, topic)
179 + } else {
180 + p.mx.Unlock()
181 + }
182 +
183 + log.Debugf("PubsubPublish: publish IPNS record for %s (%d)", topic, seqno)
184 + return p.ps.Publish(topic, data)
185 +}
186 +
187 +// Resolve resolves a name through pubsub and default depth limit
188 +func (r *PubsubResolver) Resolve(ctx context.Context, name string) (path.Path, error) {
189 + return r.ResolveN(ctx, name, DefaultDepthLimit)
190 +}
191 +
192 +// ResolveN resolves a name through pubsub with the specified depth limit
193 +func (r *PubsubResolver) ResolveN(ctx context.Context, name string, depth int) (path.Path, error) {
194 + return resolve(ctx, r, name, depth, "/ipns/")
195 +}
196 +
197 +func (r *PubsubResolver) resolveOnce(ctx context.Context, name string) (path.Path, error) {
198 + log.Debugf("PubsubResolve: resolve '%s'", name)
199 +
200 + // retrieve the public key once (for verifying messages)
201 + xname := strings.TrimPrefix(name, "/ipns/")
202 + hash, err := mh.FromB58String(xname)
203 + if err != nil {
204 + log.Warningf("PubsubResolve: bad input hash: [%s]", xname)
205 + return "", err
206 + }
207 +
208 + id := peer.ID(hash)
209 + if r.host.Peerstore().PrivKey(id) != nil {
210 + return "", errors.New("Cannot resolve own name through pubsub")
211 + }
212 +
213 + pubk := id.ExtractPublicKey()
214 + if pubk == nil {
215 + pubk, err = r.pkf.GetPublicKey(ctx, id)
216 + if err != nil {
217 + log.Warningf("PubsubResolve: error fetching public key: %s [%s]", err.Error(), xname)
218 + return "", err
219 + }
220 + }
221 +
222 + // the topic is /ipns/Qmhash
223 + if !strings.HasPrefix(name, "/ipns/") {
224 + name = "/ipns/" + name
225 + }
226 +
227 + r.mx.Lock()
228 + // see if we already have a pubsub subscription; if not, subscribe
229 + sub, ok := r.subs[name]
230 + if !ok {
231 + sub, err = r.ps.Subscribe(name)
232 + if err != nil {
233 + r.mx.Unlock()
234 + return "", err
235 + }
236 +
237 + log.Debugf("PubsubResolve: subscribed to %s", name)
238 +
239 + r.subs[name] = sub
240 +
241 + ctx, cancel := context.WithCancel(r.ctx)
242 + go r.handleSubscription(sub, name, pubk, cancel)
243 + go bootstrapPubsub(ctx, r.cr, r.host, name)
244 + }
245 + r.mx.Unlock()
246 +
247 + // resolve to what we may already have in the datastore
248 + dsval, err := r.ds.Get(dshelp.NewKeyFromBinary([]byte(name)))
249 + if err != nil {
250 + if err == ds.ErrNotFound {
251 + return "", ErrResolveFailed
252 + }
253 + return "", err
254 + }
255 +
256 + data := dsval.([]byte)
257 + entry := new(pb.IpnsEntry)
258 +
259 + err = proto.Unmarshal(data, entry)
260 + if err != nil {
261 + return "", err
262 + }
263 +
264 + // check EOL; if the entry has expired, delete from datastore and return ds.ErrNotFound
265 + eol, ok := checkEOL(entry)
266 + if ok && eol.Before(time.Now()) {
267 + err = r.ds.Delete(dshelp.NewKeyFromBinary([]byte(name)))
268 + if err != nil {
269 + log.Warningf("PubsubResolve: error deleting stale value for %s: %s", name, err.Error())
270 + }
271 +
272 + return "", ErrResolveFailed
273 + }
274 +
275 + value, err := path.ParsePath(string(entry.GetValue()))
276 + return value, err
277 +}
278 +
279 +// GetSubscriptions retrieves a list of active topic subscriptions
280 +func (r *PubsubResolver) GetSubscriptions() []string {
281 + r.mx.Lock()
282 + defer r.mx.Unlock()
283 +
284 + var res []string
285 + for sub := range r.subs {
286 + res = append(res, sub)
287 + }
288 +
289 + return res
290 +}
291 +
292 +// Cancel cancels a topic subscription; returns true if an active
293 +// subscription was canceled
294 +func (r *PubsubResolver) Cancel(name string) bool {
295 + r.mx.Lock()
296 + defer r.mx.Unlock()
297 +
298 + sub, ok := r.subs[name]
299 + if ok {
300 + sub.Cancel()
301 + delete(r.subs, name)
302 + }
303 +
304 + return ok
305 +}
306 +
307 +func (r *PubsubResolver) handleSubscription(sub *floodsub.Subscription, name string, pubk ci.PubKey, cancel func()) {
308 + defer sub.Cancel()
309 + defer cancel()
310 +
311 + for {
312 + msg, err := sub.Next(r.ctx)
313 + if err != nil {
314 + if err != context.Canceled {
315 + log.Warningf("PubsubResolve: subscription error in %s: %s", name, err.Error())
316 + }
317 + return
318 + }
319 +
320 + err = r.receive(msg, name, pubk)
321 + if err != nil {
322 + log.Warningf("PubsubResolve: error proessing update for %s: %s", name, err.Error())
323 + }
324 + }
325 +}
326 +
327 +func (r *PubsubResolver) receive(msg *floodsub.Message, name string, pubk ci.PubKey) error {
328 + data := msg.GetData()
329 + if data == nil {
330 + return errors.New("empty message")
331 + }
332 +
333 + entry := new(pb.IpnsEntry)
334 + err := proto.Unmarshal(data, entry)
335 + if err != nil {
336 + return err
337 + }
338 +
339 + ok, err := pubk.Verify(ipnsEntryDataForSig(entry), entry.GetSignature())
340 + if err != nil || !ok {
341 + return errors.New("signature verification failed")
342 + }
343 +
344 + _, err = path.ParsePath(string(entry.GetValue()))
345 + if err != nil {
346 + return err
347 + }
348 +
349 + eol, ok := checkEOL(entry)
350 + if ok && eol.Before(time.Now()) {
351 + return errors.New("stale update; EOL exceeded")
352 + }
353 +
354 + // check the sequence number against what we may already have in our datastore
355 + oval, err := r.ds.Get(dshelp.NewKeyFromBinary([]byte(name)))
356 + if err == nil {
357 + odata := oval.([]byte)
358 + oentry := new(pb.IpnsEntry)
359 +
360 + err = proto.Unmarshal(odata, oentry)
361 + if err != nil {
362 + return err
363 + }
364 +
365 + if entry.GetSequence() <= oentry.GetSequence() {
366 + return errors.New("stale update; sequence number too small")
367 + }
368 + }
369 +
370 + log.Debugf("PubsubResolve: receive IPNS record for %s", name)
371 +
372 + return r.ds.Put(dshelp.NewKeyFromBinary([]byte(name)), data)
373 +}
374 +
375 +// rendezvous with peers in the name topic through provider records
376 +// Note: rendezbous/boostrap should really be handled by the pubsub implementation itself!
377 +func bootstrapPubsub(ctx context.Context, cr routing.ContentRouting, host p2phost.Host, name string) {
378 + topic := "floodsub:" + name
379 + hash := u.Hash([]byte(topic))
380 + rz := cid.NewCidV1(cid.Raw, hash)
381 +
382 + err := cr.Provide(ctx, rz, true)
383 + if err != nil {
384 + log.Warningf("bootstrapPubsub: error providing rendezvous for %s: %s", topic, err.Error())
385 + }
386 +
387 + go func() {
388 + for {
389 + select {
390 + case <-time.After(8 * time.Hour):
391 + err := cr.Provide(ctx, rz, true)
392 + if err != nil {
393 + log.Warningf("bootstrapPubsub: error providing rendezvous for %s: %s", topic, err.Error())
394 + }
395 + case <-ctx.Done():
396 + return
397 + }
398 + }
399 + }()
400 +
401 + rzctx, cancel := context.WithTimeout(ctx, time.Second*10)
402 + defer cancel()
403 +
404 + wg := &sync.WaitGroup{}
405 + for pi := range cr.FindProvidersAsync(rzctx, rz, 10) {
406 + if pi.ID == host.ID() {
407 + continue
408 + }
409 + wg.Add(1)
410 + go func(pi pstore.PeerInfo) {
411 + defer wg.Done()
412 +
413 + ctx, cancel := context.WithTimeout(ctx, time.Second*10)
414 + defer cancel()
415 +
416 + err := host.Connect(ctx, pi)
417 + if err != nil {
418 + log.Debugf("Error connecting to pubsub peer %s: %s", pi.ID, err.Error())
419 + return
420 + }
421 +
422 + // delay to let pubsub perform its handshake
423 + time.Sleep(time.Millisecond * 250)
424 +
425 + log.Debugf("Connected to pubsub peer %s", pi.ID)
426 + }(pi)
427 + }
428 +
429 + wg.Wait()
430 +}
namesys/pubsub_test.go new
+187
@@ -0,0 +1,187 @@
1 +package namesys
2 +
3 +import (
4 + "context"
5 + "sync"
6 + "testing"
7 + "time"
8 +
9 + path "github.com/ipfs/go-ipfs/path"
10 + mockrouting "github.com/ipfs/go-ipfs/routing/mock"
11 +
12 + routing "gx/ipfs/QmPR2JzfKd9poHx9XBhzoFeBBC31ZM3W5iUPKJZWyaoZZm/go-libp2p-routing"
13 + pstore "gx/ipfs/QmPgDWmTmuzvP7QE5zwo1TmjbJme9pmZHNujB2453jkCTr/go-libp2p-peerstore"
14 + testutil "gx/ipfs/QmQgLZP9haZheimMHqqAjJh2LhRmNfEoZDfbtkpeMhi9xK/go-testutil"
15 + p2phost "gx/ipfs/QmRS46AyqtpJBsf1zmQdeizSDEzo1qkWR7rdEuPFAv8237/go-libp2p-host"
16 + netutil "gx/ipfs/QmUUNDRYXgfqdjxTg79ogkciczU5y4WY1tKMU2vEX9CRN7/go-libp2p-netutil"
17 + floodsub "gx/ipfs/QmVNv1WV6XxzQV4MBuiLX5729wMazaf8TNzm2Sq6ejyHh7/go-libp2p-floodsub"
18 + ds "gx/ipfs/QmVSase1JP7cq9QkPT46oNwdp9pT6kBkG3oqS14y3QcZjG/go-datastore"
19 + peer "gx/ipfs/QmXYjuNuxVzXKJCfWasQk1RqkhVLDM9jtUKhqc2WPQmFSB/go-libp2p-peer"
20 + ci "gx/ipfs/QmaPbCnUMBohSGo3KnxEa2bHqyJVVeEEcwtqJAYxerieBo/go-libp2p-crypto"
21 + bhost "gx/ipfs/Qmb37wDRoh9VZMZXmmZktN35szvj9GeBYDtA9giDmXwwd7/go-libp2p-blankhost"
22 +)
23 +
24 +func newNetHost(ctx context.Context, t *testing.T) p2phost.Host {
25 + netw := netutil.GenSwarmNetwork(t, ctx)
26 + return bhost.NewBlankHost(netw)
27 +}
28 +
29 +func newNetHosts(ctx context.Context, t *testing.T, n int) []p2phost.Host {
30 + var out []p2phost.Host
31 +
32 + for i := 0; i < n; i++ {
33 + h := newNetHost(ctx, t)
34 + out = append(out, h)
35 + }
36 +
37 + return out
38 +}
39 +
40 +// PubKeyFetcher implementation with a global key store
41 +type mockKeyStore struct {
42 + keys map[peer.ID]ci.PubKey
43 + mx sync.Mutex
44 +}
45 +
46 +func (m *mockKeyStore) addPubKey(id peer.ID, pkey ci.PubKey) {
47 + m.mx.Lock()
48 + defer m.mx.Unlock()
49 + m.keys[id] = pkey
50 +}
51 +
52 +func (m *mockKeyStore) getPubKey(id peer.ID) (ci.PubKey, error) {
53 + m.mx.Lock()
54 + defer m.mx.Unlock()
55 + pkey, ok := m.keys[id]
56 + if ok {
57 + return pkey, nil
58 + }
59 +
60 + return nil, routing.ErrNotFound
61 +}
62 +
63 +func (m *mockKeyStore) GetPublicKey(ctx context.Context, id peer.ID) (ci.PubKey, error) {
64 + return m.getPubKey(id)
65 +}
66 +
67 +func newMockKeyStore() *mockKeyStore {
68 + return &mockKeyStore{
69 + keys: make(map[peer.ID]ci.PubKey),
70 + }
71 +}
72 +
73 +// ConentRouting mock
74 +func newMockRouting(ms mockrouting.Server, ks *mockKeyStore, host p2phost.Host) routing.ContentRouting {
75 + id := host.ID()
76 +
77 + privk := host.Peerstore().PrivKey(id)
78 + pubk := host.Peerstore().PubKey(id)
79 + pi := host.Peerstore().PeerInfo(id)
80 +
81 + ks.addPubKey(id, pubk)
82 + return ms.Client(testutil.NewIdentity(id, pi.Addrs[0], privk, pubk))
83 +}
84 +
85 +func newMockRoutingForHosts(ms mockrouting.Server, ks *mockKeyStore, hosts []p2phost.Host) []routing.ContentRouting {
86 + rs := make([]routing.ContentRouting, len(hosts))
87 + for i := 0; i < len(hosts); i++ {
88 + rs[i] = newMockRouting(ms, ks, hosts[i])
89 + }
90 + return rs
91 +}
92 +
93 +// tests
94 +func TestPubsubPublishSubscribe(t *testing.T) {
95 + ctx, cancel := context.WithCancel(context.Background())
96 + defer cancel()
97 +
98 + ms := mockrouting.NewServer()
99 + ks := newMockKeyStore()
100 +
101 + pubhost := newNetHost(ctx, t)
102 + pubmr := newMockRouting(ms, ks, pubhost)
103 + pub := NewPubsubPublisher(ctx, pubhost, ds.NewMapDatastore(), pubmr, floodsub.NewFloodSub(ctx, pubhost))
104 + privk := pubhost.Peerstore().PrivKey(pubhost.ID())
105 + pubpinfo := pstore.PeerInfo{ID: pubhost.ID(), Addrs: pubhost.Addrs()}
106 +
107 + name := "/ipns/" + pubhost.ID().Pretty()
108 +
109 + reshosts := newNetHosts(ctx, t, 5)
110 + resmrs := newMockRoutingForHosts(ms, ks, reshosts)
111 + res := make([]*PubsubResolver, len(reshosts))
112 + for i := 0; i < len(res); i++ {
113 + res[i] = NewPubsubResolver(ctx, reshosts[i], resmrs[i], ks, floodsub.NewFloodSub(ctx, reshosts[i]))
114 + if err := reshosts[i].Connect(ctx, pubpinfo); err != nil {
115 + t.Fatal(err)
116 + }
117 + }
118 +
119 + time.Sleep(time.Millisecond * 100)
120 + for i := 0; i < len(res); i++ {
121 + checkResolveNotFound(ctx, t, i, res[i], name)
122 + // delay to avoid connection storms
123 + time.Sleep(time.Millisecond * 100)
124 + }
125 +
126 + // let the bootstrap finish
127 + time.Sleep(time.Second * 1)
128 +
129 + val := path.Path("/ipfs/QmP1DfoUjiWH2ZBo1PBH6FupdBucbDepx3HpWmEY6JMUpY")
130 + err := pub.Publish(ctx, privk, val)
131 + if err != nil {
132 + t.Fatal(err)
133 + }
134 +
135 + // let the flood propagate
136 + time.Sleep(time.Second * 1)
137 + for i := 0; i < len(res); i++ {
138 + checkResolve(ctx, t, i, res[i], name, val)
139 + }
140 +
141 + val = path.Path("/ipfs/QmP1wMAqk6aZYRZirbaAwmrNeqFRgQrwBt3orUtvSa1UYD")
142 + err = pub.Publish(ctx, privk, val)
143 + if err != nil {
144 + t.Fatal(err)
145 + }
146 +
147 + // let the flood propagate
148 + time.Sleep(time.Second * 1)
149 + for i := 0; i < len(res); i++ {
150 + checkResolve(ctx, t, i, res[i], name, val)
151 + }
152 +
153 + // cancel subscriptions
154 + for i := 0; i < len(res); i++ {
155 + res[i].Cancel(name)
156 + }
157 + time.Sleep(time.Millisecond * 100)
158 +
159 + nval := path.Path("/ipfs/QmPgDWmTmuzvP7QE5zwo1TmjbJme9pmZHNujB2453jkCTr")
160 + err = pub.Publish(ctx, privk, nval)
161 + if err != nil {
162 + t.Fatal(err)
163 + }
164 +
165 + // check we still have the old value in the resolver
166 + time.Sleep(time.Second * 1)
167 + for i := 0; i < len(res); i++ {
168 + checkResolve(ctx, t, i, res[i], name, val)
169 + }
170 +}
171 +
172 +func checkResolveNotFound(ctx context.Context, t *testing.T, i int, resolver Resolver, name string) {
173 + _, err := resolver.Resolve(ctx, name)
174 + if err != ErrResolveFailed {
175 + t.Fatalf("[resolver %d] unexpected error: %s", i, err.Error())
176 + }
177 +}
178 +
179 +func checkResolve(ctx context.Context, t *testing.T, i int, resolver Resolver, name string, val path.Path) {
180 + xval, err := resolver.Resolve(ctx, name)
181 + if err != nil {
182 + t.Fatalf("[resolver %d] resolve failed: %s", i, err.Error())
183 + }
184 + if xval != val {
185 + t.Fatalf("[resolver %d] unexpected value: %s %s", i, val, xval)
186 + }
187 +}
test/sharness/t0183-namesys-pubsub.sh new
+80
@@ -0,0 +1,80 @@
1 +#!/bin/sh
2 +
3 +test_description="Test IPNS pubsub"
4 +
5 +. lib/test-lib.sh
6 +
7 +# start iptb + wait for peering
8 +NUM_NODES=5
9 +test_expect_success 'init iptb' '
10 + iptb init -n $NUM_NODES --bootstrap=none --port=0
11 +'
12 +
13 +startup_cluster $NUM_NODES --enable-namesys-pubsub
14 +
15 +test_expect_success 'peer ids' '
16 + PEERID_0=$(iptb get id 0)
17 +'
18 +
19 +test_expect_success 'check namesys pubsub state' '
20 + echo enabled > expected &&
21 + ipfsi 0 name pubsub state > state0 &&
22 + ipfsi 1 name pubsub state > state1 &&
23 + ipfsi 2 name pubsub state > state2 &&
24 + test_cmp expected state0 &&
25 + test_cmp expected state1 &&
26 + test_cmp expected state2
27 +'
28 +
29 +test_expect_success 'subscribe nodes to the publisher topic' '
30 + ipfsi 1 name resolve /ipns/$PEERID_0 &&
31 + ipfsi 2 name resolve /ipns/$PEERID_0
32 +'
33 +
34 +test_expect_success 'check subscriptions' '
35 + echo /ipns/$PEERID_0 > expected &&
36 + ipfsi 1 name pubsub subs > subs1 &&
37 + ipfsi 2 name pubsub subs > subs2 &&
38 + test_cmp expected subs1 &&
39 + test_cmp expected subs2
40 +'
41 +
42 +test_expect_success 'add an obect on publisher node' '
43 + echo "ipns is super fun" > file &&
44 + HASH_FILE=$(ipfsi 0 add -q file)
45 +'
46 +
47 +test_expect_success 'publish that object as an ipns entry' '
48 + ipfsi 0 name publish $HASH_FILE
49 +'
50 +
51 +test_expect_success 'wait for the flood' '
52 + sleep 1
53 +'
54 +
55 +test_expect_success 'resolve name in subscriber nodes' '
56 + echo "/ipfs/$HASH_FILE" > expected &&
57 + ipfsi 1 name resolve /ipns/$PEERID_0 > name1 &&
58 + ipfsi 2 name resolve /ipns/$PEERID_0 > name2 &&
59 + test_cmp expected name1 &&
60 + test_cmp expected name2
61 +'
62 +
63 +test_expect_success 'cancel subscriptions to the publisher topic' '
64 + ipfsi 1 name pubsub cancel /ipns/$PEERID_0 &&
65 + ipfsi 2 name pubsub cancel /ipns/$PEERID_0
66 +'
67 +
68 +test_expect_success 'check subscriptions' '
69 + rm -f expected && touch expected &&
70 + ipfsi 1 name pubsub subs > subs1 &&
71 + ipfsi 2 name pubsub subs > subs2 &&
72 + test_cmp expected subs1 &&
73 + test_cmp expected subs2
74 +'
75 +
76 +test_expect_success "shut down iptb" '
77 + iptb stop
78 +'
79 +
80 +test_done