feat: pubsub http rpc with multibase (#151)
* feat: pubsub http rpc with multibase This updates HTTP RPC wire format to one from https://github.com/ipfs/go-ipfs/pull/8183 * chore: use updated go-ipfs * chore: switch ci to go-ipfs master This commit was moved from ipfs/go-ipfs-http-client@c832fc0ce1d8235c830fecd0dd44dda04a9fba37
Marcin Rataj committed
Nov 29, 2021 at 22:24 UTC
5322cf89e45197afb74cc36104ee581ca179c1d6
2 files changed
+64
-20
client/httpapi/README.md
+3
-3
@@ -1,8 +1,8 @@
1
# go-ipfs-http-api
2
3
-[](http://ipn.io)
4
-[](http://ipfs.io/)
5
-[](http://webchat.freenode.net/?channels=%23ipfs)
3
+[](https://protocol.ai)
4
+[](https://ipfs.io/)
5
+[](https://app.element.io/#/room/#ipfs:matrix.org)
6
[](https://github.com/RichardLitt/standard-readme)
7
[](https://godoc.org/github.com/ipfs/go-ipfs-http-api)
8
client/httpapi/pubsub.go
+61
-17
@@ -9,6 +9,7 @@ import (
9
iface "github.com/ipfs/interface-go-ipfs-core"
10
caopts "github.com/ipfs/interface-go-ipfs-core/options"
11
"github.com/libp2p/go-libp2p-core/peer"
12
+ mbase "github.com/multiformats/go-multibase"
13
)
14
15
type PubsubAPI HttpApi
@@ -21,8 +22,15 @@ func (api *PubsubAPI) Ls(ctx context.Context) ([]string, error) {
22
if err := api.core().Request("pubsub/ls").Exec(ctx, &out); err != nil {
23
return nil, err
24
}
24
-
25
- return out.Strings, nil
25
+ topics := make([]string, len(out.Strings))
26
+ for n, mb := range out.Strings {
27
+ _, topic, err := mbase.Decode(mb)
28
+ if err != nil {
29
+ return nil, err
30
+ }
31
+ topics[n] = string(topic)
32
+ }
33
+ return topics, nil
34
}
35
36
func (api *PubsubAPI) Peers(ctx context.Context, opts ...caopts.PubSubPeersOption) ([]peer.ID, error) {
@@ -35,7 +43,11 @@ func (api *PubsubAPI) Peers(ctx context.Context, opts ...caopts.PubSubPeersOptio
43
Strings []string
44
}
45
38
- if err := api.core().Request("pubsub/peers", options.Topic).Exec(ctx, &out); err != nil {
46
+ var optionalTopic string
47
+ if len(options.Topic) > 0 {
48
+ optionalTopic = toMultibase([]byte(options.Topic))
49
+ }
50
+ if err := api.core().Request("pubsub/peers", optionalTopic).Exec(ctx, &out); err != nil {
51
return nil, err
52
}
53
@@ -51,7 +63,7 @@ func (api *PubsubAPI) Peers(ctx context.Context, opts ...caopts.PubSubPeersOptio
63
}
64
65
func (api *PubsubAPI) Publish(ctx context.Context, topic string, message []byte) error {
54
- return api.core().Request("pubsub/pub", topic).
66
+ return api.core().Request("pubsub/pub", toMultibase([]byte(topic))).
67
FileBody(bytes.NewReader(message)).
68
Exec(ctx, nil)
69
}
@@ -64,13 +76,18 @@ type pubsubSub struct {
76
}
77
78
type pubsubMessage struct {
67
- JFrom []byte `json:"from,omitempty"`
68
- JData []byte `json:"data,omitempty"`
69
- JSeqno []byte `json:"seqno,omitempty"`
79
+ JFrom string `json:"from,omitempty"`
80
+ JData string `json:"data,omitempty"`
81
+ JSeqno string `json:"seqno,omitempty"`
82
JTopicIDs []string `json:"topicIDs,omitempty"`
83
72
- from peer.ID
73
- err error
84
+ // real values after unpacking from text/multibase envelopes
85
+ from peer.ID
86
+ data []byte
87
+ seqno []byte
88
+ topics []string
89
+
90
+ err error
91
}
92
93
func (msg *pubsubMessage) From() peer.ID {
@@ -78,15 +95,17 @@ func (msg *pubsubMessage) From() peer.ID {
95
}
96
97
func (msg *pubsubMessage) Data() []byte {
81
- return msg.JData
98
+ return msg.data
99
}
100
101
func (msg *pubsubMessage) Seq() []byte {
85
- return msg.JSeqno
102
+ return msg.seqno
103
}
104
105
+// TODO: do we want to keep this interface as []string,
106
+// or change to more correct [][]byte?
107
func (msg *pubsubMessage) Topics() []string {
89
- return msg.JTopicIDs
108
+ return msg.topics
109
}
110
111
func (s *pubsubSub) Next(ctx context.Context) (iface.PubSubMessage, error) {
@@ -98,22 +117,41 @@ func (s *pubsubSub) Next(ctx context.Context) (iface.PubSubMessage, error) {
117
if msg.err != nil {
118
return nil, msg.err
119
}
120
+ // unpack values from text/multibase envelopes
121
var err error
102
- msg.from, err = peer.IDFromBytes(msg.JFrom)
103
- return &msg, err
122
+ msg.from, err = peer.Decode(msg.JFrom)
123
+ if err != nil {
124
+ return nil, err
125
+ }
126
+ _, msg.data, err = mbase.Decode(msg.JData)
127
+ if err != nil {
128
+ return nil, err
129
+ }
130
+ _, msg.seqno, err = mbase.Decode(msg.JSeqno)
131
+ if err != nil {
132
+ return nil, err
133
+ }
134
+ for _, mbt := range msg.JTopicIDs {
135
+ _, topic, err := mbase.Decode(mbt)
136
+ if err != nil {
137
+ return nil, err
138
+ }
139
+ msg.topics = append(msg.topics, string(topic))
140
+ }
141
+ return &msg, nil
142
case <-ctx.Done():
143
return nil, ctx.Err()
144
}
145
}
146
147
func (api *PubsubAPI) Subscribe(ctx context.Context, topic string, opts ...caopts.PubSubSubscribeOption) (iface.PubSubSubscription, error) {
148
+ /* right now we have no options (discover got deprecated)
149
options, err := caopts.PubSubSubscribeOptions(opts...)
150
if err != nil {
151
return nil, err
152
}
114
-
115
- resp, err := api.core().Request("pubsub/sub", topic).
116
- Option("discover", options.Discover).Send(ctx)
153
+ */
154
+ resp, err := api.core().Request("pubsub/sub", toMultibase([]byte(topic))).Send(ctx)
155
156
if err != nil {
157
return nil, err
@@ -168,3 +206,9 @@ func (s *pubsubSub) Close() error {
206
func (api *PubsubAPI) core() *HttpApi {
207
return (*HttpApi)(api)
208
}
209
+
210
+// Encodes bytes into URL-safe multibase that can be sent over HTTP RPC (URL or body)
211
+func toMultibase(data []byte) string {
212
+ mb, _ := mbase.Encode(mbase.Base64url, data)
213
+ return mb
214
+}