coreapi: implement pubsub api
License: MIT Signed-off-by: Łukasz Magiera <magik6k@gmail.com>
Łukasz Magiera committed
Mar 10, 2018 at 19:28 UTC
51bb9d6843f12507db23d809caab163862ad2cc8
6 files changed
+123
-16
core/coreapi/coreapi.go
+5
@@ -72,3 +72,8 @@ func (api *CoreAPI) Dht() coreiface.DhtAPI {
72
func (api *CoreAPI) Swarm() coreiface.SwarmAPI {
73
return (*SwarmAPI)(api)
74
}
75
+
76
+// PubSub returns the PubSubAPI interface implementation backed by the go-ipfs node
77
+func (api *CoreAPI) PubSub() coreiface.PubSubAPI {
78
+ return (*PubSubAPI)(api)
79
+}
core/coreapi/interface/coreapi.go
+3
@@ -37,6 +37,9 @@ type CoreAPI interface {
37
// Swarm returns an implementation of Swarm API
38
Swarm() SwarmAPI
39
40
+ // PubSub returns an implementation of PubSub API
41
+ PubSub() PubSubAPI
42
+
43
// ResolvePath resolves the path using Unixfs resolver
44
ResolvePath(context.Context, Path) (ResolvedPath, error)
45
core/coreapi/interface/errors.go
+1
-1
@@ -4,5 +4,5 @@ import "errors"
4
5
var (
6
ErrIsDir = errors.New("object is a directory")
7
- ErrOffline = errors.New("can't resolve, ipfs node is offline")
7
+ ErrOffline = errors.New("this action must be run in online mode, try running 'ipfs daemon' first")
8
)
core/coreapi/interface/options/pubsub.go
+5
-3
@@ -39,16 +39,18 @@ func PubSubSubscribeOptions(opts ...PubSubSubscribeOption) (*PubSubSubscribeSett
39
return options, nil
40
}
41
42
-type PubSubOptions struct{}
42
+type pubsubOpts struct{}
43
44
-func (api *PubSubOptions) WithTopic(topic string) PubSubPeersOption {
44
+var PubBub nameOpts
45
+
46
+func (pubsubOpts) Topic(topic string) PubSubPeersOption {
47
return func(settings *PubSubPeersSettings) error {
48
settings.Topic = topic
49
return nil
50
}
51
}
52
51
-func (api *PubSubOptions) WithDiscover(discover bool) PubSubSubscribeOption {
53
+func (pubsubOpts) Discover(discover bool) PubSubSubscribeOption {
54
return func(settings *PubSubSubscribeSettings) error {
55
settings.Discover = discover
56
return nil
core/coreapi/interface/pubsub.go
+4
-12
@@ -6,15 +6,15 @@ import (
6
7
options "github.com/ipfs/go-ipfs/core/coreapi/interface/options"
8
9
- peer "gx/ipfs/QmZoWKhxUmZ2seW4BzX6fJkNR8hh9PsGModr7q171yq2SS/go-libp2p-peer"
9
+ peer "gx/ipfs/QmQsErDt8Qgw1XrsXf2BpEzDgGWtB1YLsTAARBup5b6B9W/go-libp2p-peer"
10
)
11
12
// PubSubSubscription is an active PubSub subscription
13
type PubSubSubscription interface {
14
io.Closer
15
16
- // Chan return incoming message channel
17
- Chan(context.Context) <-chan PubSubMessage
16
+ // Next return the next incoming message
17
+ Next(context.Context) (PubSubMessage, error)
18
}
19
20
// PubSubMessage is a single PubSub message
@@ -35,17 +35,9 @@ type PubSubAPI interface {
35
// TODO: WithTopic
36
Peers(context.Context, ...options.PubSubPeersOption) ([]peer.ID, error)
37
38
- // WithTopic is an option for peers which specifies a topic filter for the
39
- // function
40
- WithTopic(topic string) options.PubSubPeersOption
41
-
38
// Publish a message to a given pubsub topic
39
Publish(context.Context, string, []byte) error
40
41
// Subscribe to messages on a given topic
46
- Subscribe(context.Context, string) (PubSubSubscription, error)
47
-
48
- // WithDiscover is an option for Subscribe which specifies whether to try to
49
- // discover other peers subscribed to the same topic
50
- WithDiscover(discover bool) options.PubSubSubscribeOption
42
+ Subscribe(context.Context, string, ...options.PubSubSubscribeOption) (PubSubSubscription, error)
43
}
core/coreapi/pubsub.go
new
+105
@@ -0,0 +1,105 @@
1
+package coreapi
2
+
3
+import (
4
+ "context"
5
+ "errors"
6
+
7
+ coreiface "github.com/ipfs/go-ipfs/core/coreapi/interface"
8
+ caopts "github.com/ipfs/go-ipfs/core/coreapi/interface/options"
9
+
10
+ peer "gx/ipfs/QmQsErDt8Qgw1XrsXf2BpEzDgGWtB1YLsTAARBup5b6B9W/go-libp2p-peer"
11
+ floodsub "gx/ipfs/QmY1L5krVk8dv8d74uESmJTXGpoigVYqBVxXXz1aS8aFSb/go-libp2p-floodsub"
12
+)
13
+
14
+type PubSubAPI CoreAPI
15
+
16
+type pubSubSubscription struct {
17
+ subscription *floodsub.Subscription
18
+}
19
+
20
+type pubSubMessage struct {
21
+ msg *floodsub.Message
22
+}
23
+
24
+func (api *PubSubAPI) Ls(ctx context.Context) ([]string, error) {
25
+ if err := api.checkNode(); err != nil {
26
+ return nil, err
27
+ }
28
+
29
+ return api.node.Floodsub.GetTopics(), nil
30
+}
31
+
32
+func (api *PubSubAPI) Peers(ctx context.Context, opts ...caopts.PubSubPeersOption) ([]peer.ID, error) {
33
+ if err := api.checkNode(); err != nil {
34
+ return nil, err
35
+ }
36
+
37
+ settings, err := caopts.PubSubPeersOptions(opts...)
38
+ if err != nil {
39
+ return nil, err
40
+ }
41
+
42
+ peers := api.node.Floodsub.ListPeers(settings.Topic)
43
+ out := make([]peer.ID, len(peers))
44
+
45
+ for i, peer := range peers {
46
+ out[i] = peer
47
+ }
48
+
49
+ return out, nil
50
+}
51
+
52
+func (api *PubSubAPI) Publish(ctx context.Context, topic string, data []byte) error {
53
+ if err := api.checkNode(); err != nil {
54
+ return err
55
+ }
56
+
57
+ return api.node.Floodsub.Publish(topic, data)
58
+}
59
+
60
+func (api *PubSubAPI) Subscribe(ctx context.Context, topic string, opts ...caopts.PubSubSubscribeOption) (coreiface.PubSubSubscription, error) {
61
+ if err := api.checkNode(); err != nil {
62
+ return nil, err
63
+ }
64
+
65
+ sub, err := api.node.Floodsub.Subscribe(topic)
66
+ if err != nil {
67
+ return nil, err
68
+ }
69
+
70
+ return &pubSubSubscription{sub}, nil
71
+}
72
+
73
+func (api *PubSubAPI) checkNode() error {
74
+ if !api.node.OnlineMode() {
75
+ return coreiface.ErrOffline
76
+ }
77
+
78
+ if api.node.Floodsub == nil {
79
+ return errors.New("experimental pubsub feature not enabled. Run daemon with --enable-pubsub-experiment to use.")
80
+ }
81
+
82
+ return nil
83
+}
84
+
85
+func (sub *pubSubSubscription) Close() error {
86
+ sub.subscription.Cancel()
87
+ return nil
88
+}
89
+
90
+func (sub *pubSubSubscription) Next(ctx context.Context) (coreiface.PubSubMessage, error) {
91
+ msg, err := sub.subscription.Next(ctx)
92
+ if err != nil {
93
+ return nil, err
94
+ }
95
+
96
+ return &pubSubMessage{msg}, nil
97
+}
98
+
99
+func (msg *pubSubMessage) From() peer.ID {
100
+ return peer.ID(msg.msg.From)
101
+}
102
+
103
+func (msg *pubSubMessage) Data() []byte {
104
+ return msg.msg.Data
105
+}