floodsub: add api for pub/sub
License: MIT Signed-off-by: Jeromy <why@ipfs.io>
Jeromy committed
Sep 10, 2016 at 06:48 UTC
4b096c4bba60ffac1789cd5b58a42d44eb597f3b
4 files changed
+175
-1
core/commands/pubsub.go
new
+162
@@ -0,0 +1,162 @@
1
+package commands
2
+
3
+import (
4
+ "bytes"
5
+ "encoding/binary"
6
+ "io"
7
+
8
+ cmds "github.com/ipfs/go-ipfs/commands"
9
+
10
+ floodsub "gx/ipfs/QmQriRMW5cCJyLrzDnXi7fZ5mVbetiEZjPjbqoJhuSL94m/floodsub"
11
+ u "gx/ipfs/QmZNVWh8LLjAavuQ2JXuFmuYH3C11xo988vSgp7UQrTRj1/go-ipfs-util"
12
+)
13
+
14
+var PubsubCmd = &cmds.Command{
15
+ Helptext: cmds.HelpText{
16
+ Tagline: "An experimental publish-subscribe system on ipfs.",
17
+ ShortDescription: `
18
+ipfs pubsub allows you to publish messages to a given topic, and also to
19
+subscribe to new messages on a given topic.
20
+
21
+This is an experimental feature. It is not intended in its current state
22
+to be used in a production environment.
23
+`,
24
+ },
25
+ Subcommands: map[string]*cmds.Command{
26
+ "pub": PubsubPubCmd,
27
+ "sub": PubsubSubCmd,
28
+ },
29
+}
30
+
31
+var PubsubSubCmd = &cmds.Command{
32
+ Helptext: cmds.HelpText{
33
+ Tagline: "Subscribe to messages on a given topic.",
34
+ ShortDescription: `
35
+ipfs pubsub sub subscribes to messages on a given topic.
36
+
37
+This is an experimental feature. It is not intended in its current state
38
+to be used in a production environment.
39
+`,
40
+ },
41
+ Arguments: []cmds.Argument{
42
+ cmds.StringArg("topic", true, false, "String name of topic to subscribe to."),
43
+ },
44
+ Run: func(req cmds.Request, res cmds.Response) {
45
+ n, err := req.InvocContext().GetNode()
46
+ if err != nil {
47
+ res.SetError(err, cmds.ErrNormal)
48
+ return
49
+ }
50
+
51
+ // Must be online!
52
+ if !n.OnlineMode() {
53
+ res.SetError(errNotOnline, cmds.ErrClient)
54
+ return
55
+ }
56
+
57
+ topic := req.Arguments()[0]
58
+ msgs, err := n.Floodsub.Subscribe(topic)
59
+ if err != nil {
60
+ res.SetError(err, cmds.ErrNormal)
61
+ return
62
+ }
63
+
64
+ out := make(chan interface{})
65
+ res.SetOutput((<-chan interface{})(out))
66
+
67
+ ctx := req.Context()
68
+ go func() {
69
+ defer close(out)
70
+ for {
71
+ select {
72
+ case msg, ok := <-msgs:
73
+ if !ok {
74
+ return
75
+ }
76
+ out <- msg
77
+ case <-ctx.Done():
78
+ n.Floodsub.Unsub(topic)
79
+ }
80
+ }
81
+ }()
82
+ },
83
+ Marshalers: cmds.MarshalerMap{
84
+ cmds.Text: getPsMsgMarshaler(func(m *floodsub.Message) (io.Reader, error) {
85
+ log.Error("FROM: ", m.GetFrom())
86
+ return bytes.NewReader(m.Data), nil
87
+ }),
88
+ "ndpayload": getPsMsgMarshaler(func(m *floodsub.Message) (io.Reader, error) {
89
+ m.Data = append(m.Data, '\n')
90
+ return bytes.NewReader(m.Data), nil
91
+ }),
92
+ "lenpayload": getPsMsgMarshaler(func(m *floodsub.Message) (io.Reader, error) {
93
+ buf := make([]byte, 8)
94
+ n := binary.PutUvarint(buf, uint64(len(m.Data)))
95
+ return io.MultiReader(bytes.NewReader(buf[:n]), bytes.NewReader(m.Data)), nil
96
+ }),
97
+ },
98
+ Type: floodsub.Message{},
99
+}
100
+
101
+func getPsMsgMarshaler(f func(m *floodsub.Message) (io.Reader, error)) func(cmds.Response) (io.Reader, error) {
102
+ return func(res cmds.Response) (io.Reader, error) {
103
+ outChan, ok := res.Output().(<-chan interface{})
104
+ if !ok {
105
+ return nil, u.ErrCast()
106
+ }
107
+
108
+ marshal := func(v interface{}) (io.Reader, error) {
109
+ obj, ok := v.(*floodsub.Message)
110
+ if !ok {
111
+ return nil, u.ErrCast()
112
+ }
113
+
114
+ return f(obj)
115
+ }
116
+
117
+ return &cmds.ChannelMarshaler{
118
+ Channel: outChan,
119
+ Marshaler: marshal,
120
+ Res: res,
121
+ }, nil
122
+ }
123
+}
124
+
125
+var PubsubPubCmd = &cmds.Command{
126
+ Helptext: cmds.HelpText{
127
+ Tagline: "Publish a message to a given pubsub topic.",
128
+ ShortDescription: `
129
+ipfs pubsub pub publishes a message to a specified topic.
130
+
131
+This is an experimental feature. It is not intended in its current state
132
+to be used in a production environment.
133
+`,
134
+ },
135
+ Arguments: []cmds.Argument{
136
+ cmds.StringArg("topic", true, false, "Topic to publish to."),
137
+ cmds.StringArg("data", true, true, "Payload of message to publish.").EnableStdin(),
138
+ },
139
+ Options: []cmds.Option{},
140
+ Run: func(req cmds.Request, res cmds.Response) {
141
+ n, err := req.InvocContext().GetNode()
142
+ if err != nil {
143
+ res.SetError(err, cmds.ErrNormal)
144
+ return
145
+ }
146
+
147
+ // Must be online!
148
+ if !n.OnlineMode() {
149
+ res.SetError(errNotOnline, cmds.ErrClient)
150
+ return
151
+ }
152
+
153
+ topic := req.Arguments()[0]
154
+
155
+ for _, data := range req.Arguments()[1:] {
156
+ if err := n.Floodsub.Publish(topic, []byte(data)); err != nil {
157
+ res.SetError(err, cmds.ErrNormal)
158
+ return
159
+ }
160
+ }
161
+ },
162
+}
core/commands/root.go
+1
@@ -105,6 +105,7 @@ var rootSubcommands = map[string]*cmds.Command{
105
"stats": StatsCmd,
106
"swarm": SwarmCmd,
107
"tar": TarCmd,
108
+ "pubsub": PubsubCmd,
109
"tour": tourCmd,
110
"file": unixfs.UnixFSCmd,
111
"update": ExternalBinary(),
core/core.go
+6
-1
@@ -17,7 +17,8 @@ import (
17
"time"
18
19
diag "github.com/ipfs/go-ipfs/diagnostics"
20
- goprocess "gx/ipfs/QmQopLATEYMNg7dVqZRNDfeE2S1yKy8zrRh5xnYiuqeZBn/goprocess"
20
+ floodsub "gx/ipfs/QmQriRMW5cCJyLrzDnXi7fZ5mVbetiEZjPjbqoJhuSL94m/floodsub"
21
+ goprocess "gx/ipfs/QmSF8fPo3jgVBAy8fpdjjYqgG87dkJgUprRBHRd2tmfgpP/goprocess"
22
mamask "gx/ipfs/QmSMZwvs3n4GBikZ7hKzT17c3bk65FmyZo2JqtJ16swqCv/multiaddr-filter"
23
logging "gx/ipfs/QmSpJByNKFX1sCsHBEp3R73FL4NF6FnQTEGyNAXHm2GS52/go-log"
24
b58 "gx/ipfs/QmT8rehPR3F6bmwL6zjUN8XpiDBFFpMP2myPdC6ApsWfJf/go-base58"
@@ -112,6 +113,8 @@ type IpfsNode struct {
113
Reprovider *rp.Reprovider // the value reprovider system
114
IpnsRepub *ipnsrp.Republisher
115
116
+ Floodsub *floodsub.PubSub
117
+
118
proc goprocess.Process
119
ctx context.Context
120
@@ -184,6 +187,8 @@ func (n *IpfsNode) startOnlineServices(ctx context.Context, routingOption Routin
187
go n.Reprovider.ProvideEvery(ctx, interval)
188
}
189
190
+ n.Floodsub = floodsub.NewFloodSub(ctx, peerhost)
191
+
192
// setup local discovery
193
if do != nil {
194
service, err := do(ctx, n.PeerHost)
package.json
+6
@@ -263,6 +263,12 @@
263
"hash": "QmdCL8M8DXJdSRnwhpDhukX5r8ydjxnzPJpaKrFudDA8yn",
264
"name": "hang-fds",
265
"version": "0.0.0"
266
+ },
267
+ {
268
+ "author": "whyrusleeping",
269
+ "hash": "QmQriRMW5cCJyLrzDnXi7fZ5mVbetiEZjPjbqoJhuSL94m",
270
+ "name": "floodsub",
271
+ "version": "0.3.0"
272
}
273
],
274
"gxVersion": "0.4.0",