logos-messaging/logos-messaging-go
1
0
Fork 0
mirror of https://github.com/logos-messaging/logos-messaging-go.git synced 2026-05-21 18:59:29 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor: create separate functions for subscriptions and publishing

Browse Source
This commit is contained in:
Richard Ramos 2021-11-19 16:01:52 -04:00
parent 84b1f995bc
commit 3ac06f01ce
No known key found for this signature in database
GPG Key ID: 80D4B01265FDFE8F
13 changed files with 54 additions and 40 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
examples/basic2/main.go
View File

@ -103,7 +103,7 @@ func write(ctx context.Context, wakuNode *node.WakuNode, msgContent string) {
Timestamp: timestamp, Timestamp: timestamp,
} }
_, err = wakuNode.Relay().Publish(ctx, msg, nil) _, err = wakuNode.Relay().Publish(ctx, msg)
if err != nil { if err != nil {
log.Error("Error sending a message: ", err) log.Error("Error sending a message: ", err)
} }
@ -117,7 +117,7 @@ func writeLoop(ctx context.Context, wakuNode *node.WakuNode) {
} }
func readLoop(ctx context.Context, wakuNode *node.WakuNode) { func readLoop(ctx context.Context, wakuNode *node.WakuNode) {
sub, err := wakuNode.Relay().Subscribe(ctx, nil) sub, err := wakuNode.Relay().Subscribe(ctx)
if err != nil { if err != nil {
log.Error("Could not subscribe: ", err) log.Error("Could not subscribe: ", err)
return return

8
examples/chat2/chat.go
View File

@ -51,7 +51,7 @@ func NewChat(ctx context.Context, n *node.WakuNode, selfID peer.ID, contentTopic
if useLightPush { if useLightPush {
cf := filter.ContentFilter{ cf := filter.ContentFilter{
Topic: string(relay.GetTopic(nil)), Topic: relay.DefaultWakuTopic,
ContentTopics: []string{contentTopic}, ContentTopics: []string{contentTopic},
} }
var err error var err error
@ -61,7 +61,7 @@ func NewChat(ctx context.Context, n *node.WakuNode, selfID peer.ID, contentTopic
return nil, err return nil, err
} }
} else { } else {
sub, err := n.Relay().Subscribe(ctx, nil) sub, err := n.Relay().Subscribe(ctx)
if err != nil { if err != nil {
return nil, err return nil, err
} }
@ -124,10 +124,10 @@ func (cr *Chat) Publish(ctx context.Context, message string) error {
} }
if cr.useLightPush { if cr.useLightPush {
_, err = cr.node.Lightpush().Publish(ctx, wakuMsg, nil) _, err = cr.node.Lightpush().Publish(ctx, wakuMsg)
} else { } else {
_, err = cr.node.Relay().Publish(ctx, wakuMsg, nil) _, err = cr.node.Relay().Publish(ctx, wakuMsg)
} }

4
examples/filter2/main.go
View File

@ -158,7 +158,7 @@ func write(ctx context.Context, wakuNode *node.WakuNode, msgContent string) {
Timestamp: timestamp, Timestamp: timestamp,
} }
_, err := wakuNode.Relay().Publish(ctx, msg, nil) _, err := wakuNode.Relay().Publish(ctx, msg)
if err != nil { if err != nil {
log.Error("Error sending a message: ", err) log.Error("Error sending a message: ", err)
} }
@ -173,7 +173,7 @@ func writeLoop(ctx context.Context, wakuNode *node.WakuNode) {
func readLoop(ctx context.Context, wakuNode *node.WakuNode) { func readLoop(ctx context.Context, wakuNode *node.WakuNode) {
pubsubTopic := pubSubTopic.String() pubsubTopic := pubSubTopic.String()
sub, err := wakuNode.Relay().Subscribe(ctx, &pubsubTopic) sub, err := wakuNode.Relay().SubscribeWithTopic(ctx, pubsubTopic)
if err != nil { if err != nil {
log.Error("Could not subscribe: ", err) log.Error("Could not subscribe: ", err)
return return

4
tests/connection_test.go
View File

@ -38,7 +38,7 @@ func TestBasicSendingReceiving(t *testing.T) {
require.NoError(t, write(ctx, wakuNode, "test")) require.NoError(t, write(ctx, wakuNode, "test"))
sub, err := wakuNode.Relay().Subscribe(ctx, nil) sub, err := wakuNode.Relay().Subscribe(ctx)
require.NoError(t, err) require.NoError(t, err)
value := <-sub.C value := <-sub.C
@ -69,6 +69,6 @@ func write(ctx context.Context, wakuNode *node.WakuNode, msgContent string) erro
Timestamp: timestamp, Timestamp: timestamp,
} }
_, err = wakuNode.Relay().Publish(ctx, msg, nil) _, err = wakuNode.Relay().Publish(ctx, msg)
return err return err
} }

2
waku/node.go
View File

@ -232,7 +232,7 @@ func Execute(options Options) {
if !options.Relay.Disable { if !options.Relay.Disable {
for _, nodeTopic := range options.Relay.Topics { for _, nodeTopic := range options.Relay.Topics {
_, err := wakuNode.Relay().Subscribe(ctx, &nodeTopic) _, err := wakuNode.Relay().SubscribeWithTopic(ctx, nodeTopic)
failOnErr(err, "Error subscring to topic") failOnErr(err, "Error subscring to topic")
} }
} }

2
waku/v2/node/wakunode2.go
View File

@ -362,7 +362,7 @@ func (w *WakuNode) mountRelay(opts ...pubsub.Option) error {
} }
if w.opts.enableRelay { if w.opts.enableRelay {
_, err = w.relay.Subscribe(w.ctx, nil) _, err = w.relay.Subscribe(w.ctx)
if err != nil { if err != nil {
return err return err
} }

8
waku/v2/protocol/filter/waku_filter_test.go
View File

@ -25,7 +25,7 @@ func makeWakuRelay(t *testing.T, topic string, broadcaster v2.Broadcaster) (*rel
relay, err := relay.NewWakuRelay(context.Background(), host, broadcaster) relay, err := relay.NewWakuRelay(context.Background(), host, broadcaster)
require.NoError(t, err) require.NoError(t, err)
sub, err := relay.Subscribe(context.Background(), &topic) sub, err := relay.SubscribeWithTopic(context.Background(), topic)
require.NoError(t, err) require.NoError(t, err)
return relay, sub, host return relay, sub, host
@ -95,7 +95,7 @@ func TestWakuFilter(t *testing.T) {
require.Equal(t, contentFilter.ContentTopics[0], env.Message().GetContentTopic()) require.Equal(t, contentFilter.ContentTopics[0], env.Message().GetContentTopic())
}() }()
_, err = node2.Publish(ctx, tests.CreateWakuMessage(testContentTopic, 0), &testTopic) _, err = node2.PublishWithTopic(ctx, tests.CreateWakuMessage(testContentTopic, 0), testTopic)
require.NoError(t, err) require.NoError(t, err)
wg.Wait() wg.Wait()
@ -112,7 +112,7 @@ func TestWakuFilter(t *testing.T) {
} }
}() }()
_, err = node2.Publish(ctx, tests.CreateWakuMessage("TopicB", 1), &testTopic) _, err = node2.PublishWithTopic(ctx, tests.CreateWakuMessage("TopicB", 1), testTopic)
require.NoError(t, err) require.NoError(t, err)
wg.Wait() wg.Wait()
@ -134,7 +134,7 @@ func TestWakuFilter(t *testing.T) {
time.Sleep(1 * time.Second) time.Sleep(1 * time.Second)
_, err = node2.Publish(ctx, tests.CreateWakuMessage(testContentTopic, 2), &testTopic) _, err = node2.PublishWithTopic(ctx, tests.CreateWakuMessage(testContentTopic, 2), testTopic)
require.NoError(t, err) require.NoError(t, err)
wg.Wait() wg.Wait()
} }

10
waku/v2/protocol/lightpush/waku_lightpush.go
View File

@ -83,7 +83,7 @@ func (wakuLP *WakuLightPush) onRequest(s network.Stream) {
// TODO: Assumes success, should probably be extended to check for network, peers, etc // TODO: Assumes success, should probably be extended to check for network, peers, etc
// It might make sense to use WithReadiness option here? // It might make sense to use WithReadiness option here?
_, err := wakuLP.relay.Publish(wakuLP.ctx, message, &pubSubTopic) _, err := wakuLP.relay.PublishWithTopic(wakuLP.ctx, message, pubSubTopic)
if err != nil { if err != nil {
response.IsSuccess = false response.IsSuccess = false
@ -181,14 +181,14 @@ func (wakuLP *WakuLightPush) Stop() {
wakuLP.h.RemoveStreamHandler(LightPushID_v20beta1) wakuLP.h.RemoveStreamHandler(LightPushID_v20beta1)
} }
func (wakuLP *WakuLightPush) Publish(ctx context.Context, message *pb.WakuMessage, topic *string, opts ...LightPushOption) ([]byte, error) { func (wakuLP *WakuLightPush) PublishWithTopic(ctx context.Context, message *pb.WakuMessage, topic string, opts ...LightPushOption) ([]byte, error) {
if message == nil { if message == nil {
return nil, errors.New("message can't be null") return nil, errors.New("message can't be null")
} }
req := new(pb.PushRequest) req := new(pb.PushRequest)
req.Message = message req.Message = message
req.PubsubTopic = string(relay.GetTopic(topic)) req.PubsubTopic = topic
response, err := wakuLP.request(ctx, req, opts...) response, err := wakuLP.request(ctx, req, opts...)
if err != nil { if err != nil {
@ -202,3 +202,7 @@ func (wakuLP *WakuLightPush) Publish(ctx context.Context, message *pb.WakuMessag
return nil, errors.New(response.Info) return nil, errors.New(response.Info)
} }
} }
func (wakuLP *WakuLightPush) Publish(ctx context.Context, message *pb.WakuMessage, opts ...LightPushOption) ([]byte, error) {
return wakuLP.PublishWithTopic(ctx, message, relay.DefaultWakuTopic, opts...)
}

6
waku/v2/protocol/lightpush/waku_lightpush_test.go
View File

@ -27,7 +27,7 @@ func makeWakuRelay(t *testing.T, topic string) (*relay.WakuRelay, *relay.Subscri
relay, err := relay.NewWakuRelay(context.Background(), host, v2.NewBroadcaster(10)) relay, err := relay.NewWakuRelay(context.Background(), host, v2.NewBroadcaster(10))
require.NoError(t, err) require.NoError(t, err)
sub, err := relay.Subscribe(context.Background(), &topic) sub, err := relay.SubscribeWithTopic(context.Background(), topic)
require.NoError(t, err) require.NoError(t, err)
return relay, sub, host return relay, sub, host
@ -109,7 +109,7 @@ func TestWakuLightPush(t *testing.T) {
require.True(t, resp.IsSuccess) require.True(t, resp.IsSuccess)
// Checking that msg hash is correct // Checking that msg hash is correct
hash, err := client.Publish(ctx, msg2, &testTopic) hash, err := client.PublishWithTopic(ctx, msg2, testTopic)
require.NoError(t, err) require.NoError(t, err)
require.Equal(t, protocol.NewEnvelope(msg2, string(testTopic)).Hash(), hash) require.Equal(t, protocol.NewEnvelope(msg2, string(testTopic)).Hash(), hash)
wg.Wait() wg.Wait()
@ -137,6 +137,6 @@ func TestWakuLightPushNoPeers(t *testing.T) {
require.NoError(t, err) require.NoError(t, err)
client := NewWakuLightPush(ctx, clientHost, nil) client := NewWakuLightPush(ctx, clientHost, nil)
_, err = client.Publish(ctx, tests.CreateWakuMessage("test", float64(0)), &testTopic) _, err = client.PublishWithTopic(ctx, tests.CreateWakuMessage("test", float64(0)), testTopic)
require.Errorf(t, err, "no suitable remote peers") require.Errorf(t, err, "no suitable remote peers")
} }

25
waku/v2/protocol/relay/waku_relay.go
View File

@ -143,7 +143,7 @@ func (w *WakuRelay) subscribe(topic string) (subs *pubsub.Subscription, err erro
return sub, nil return sub, nil
} }
func (w *WakuRelay) Publish(ctx context.Context, message *pb.WakuMessage, topic *string) ([]byte, error) { func (w *WakuRelay) PublishWithTopic(ctx context.Context, message *pb.WakuMessage, topic string) ([]byte, error) {
// Publish a `WakuMessage` to a PubSub topic. // Publish a `WakuMessage` to a PubSub topic.
if w.pubsub == nil { if w.pubsub == nil {
return nil, errors.New("PubSub hasn't been set") return nil, errors.New("PubSub hasn't been set")
@ -153,7 +153,7 @@ func (w *WakuRelay) Publish(ctx context.Context, message *pb.WakuMessage, topic
return nil, errors.New("message can't be null") return nil, errors.New("message can't be null")
} }
pubSubTopic, err := w.upsertTopic(GetTopic(topic)) pubSubTopic, err := w.upsertTopic(topic)
if err != nil { if err != nil {
return nil, err return nil, err
@ -174,12 +174,8 @@ func (w *WakuRelay) Publish(ctx context.Context, message *pb.WakuMessage, topic
return hash, nil return hash, nil
} }
func GetTopic(topic *string) string { func (w *WakuRelay) Publish(ctx context.Context, message *pb.WakuMessage) ([]byte, error) {
t := DefaultWakuTopic return w.PublishWithTopic(ctx, message, DefaultWakuTopic)
if topic != nil {
t = *topic
}
return t
} }
func (w *WakuRelay) Stop() { func (w *WakuRelay) Stop() {
@ -195,11 +191,10 @@ func (w *WakuRelay) Stop() {
w.subscriptions = nil w.subscriptions = nil
} }
func (w *WakuRelay) Subscribe(ctx context.Context, topic *string) (*Subscription, error) { func (w *WakuRelay) SubscribeWithTopic(ctx context.Context, topic string) (*Subscription, error) {
// Subscribes to a PubSub topic. // Subscribes to a PubSub topic.
// NOTE The data field SHOULD be decoded as a WakuMessage. // NOTE The data field SHOULD be decoded as a WakuMessage.
t := GetTopic(topic) sub, err := w.subscribe(topic)
sub, err := w.subscribe(t)
if err != nil { if err != nil {
return nil, err return nil, err
@ -214,17 +209,21 @@ func (w *WakuRelay) Subscribe(ctx context.Context, topic *string) (*Subscription
w.subscriptionsMutex.Lock() w.subscriptionsMutex.Lock()
defer w.subscriptionsMutex.Unlock() defer w.subscriptionsMutex.Unlock()
w.subscriptions[t] = append(w.subscriptions[t], subscription) w.subscriptions[topic] = append(w.subscriptions[topic], subscription)
if w.bcaster != nil { if w.bcaster != nil {
w.bcaster.Register(subscription.C) w.bcaster.Register(subscription.C)
} }
go w.subscribeToTopic(t, subscription, sub) go w.subscribeToTopic(topic, subscription, sub)
return subscription, nil return subscription, nil
} }
func (w *WakuRelay) Subscribe(ctx context.Context) (*Subscription, error) {
return w.SubscribeWithTopic(ctx, DefaultWakuTopic)
}
func (w *WakuRelay) Unsubscribe(ctx context.Context, topic string) error { func (w *WakuRelay) Unsubscribe(ctx context.Context, topic string) error {
if _, ok := w.relaySubs[topic]; !ok { if _, ok := w.relaySubs[topic]; !ok {
return fmt.Errorf("topics %s is not subscribed", (string)(topic)) return fmt.Errorf("topics %s is not subscribed", (string)(topic))

2
waku/v2/protocol/relay/waku_relay_test.go
View File

@ -45,7 +45,7 @@ func TestWakuRelay(t *testing.T) {
ContentTopic: "test", ContentTopic: "test",
Timestamp: 0, Timestamp: 0,
} }
_, err = relay.Publish(context.Background(), msg, &testTopic) _, err = relay.PublishWithTopic(context.Background(), msg, testTopic)
require.NoError(t, err) require.NoError(t, err)
<-ctx.Done() <-ctx.Done()

4
waku/v2/rpc/filter_test.go
View File

@ -24,7 +24,7 @@ func makeFilterService(t *testing.T) *FilterService {
err = n.Start() err = n.Start()
require.NoError(t, err) require.NoError(t, err)
_, err = n.Relay().Subscribe(context.Background(), &testTopic) _, err = n.Relay().SubscribeWithTopic(context.Background(), testTopic)
require.NoError(t, err) require.NoError(t, err)
return &FilterService{n} return &FilterService{n}
@ -40,7 +40,7 @@ func TestFilterSubscription(t *testing.T) {
node, err := relay.NewWakuRelay(context.Background(), host, v2.NewBroadcaster(10)) node, err := relay.NewWakuRelay(context.Background(), host, v2.NewBroadcaster(10))
require.NoError(t, err) require.NoError(t, err)
_, err = node.Subscribe(context.Background(), &testTopic) _, err = node.SubscribeWithTopic(context.Background(), testTopic)
require.NoError(t, err) require.NoError(t, err)
_ = filter.NewWakuFilter(context.Background(), host, false) _ = filter.NewWakuFilter(context.Background(), host, false)

15
waku/v2/rpc/relay.go
View File

@ -77,7 +77,12 @@ func (r *RelayService) Stop() {
} }
func (r *RelayService) PostV1Message(req *http.Request, args *RelayMessageArgs, reply *SuccessReply) error { func (r *RelayService) PostV1Message(req *http.Request, args *RelayMessageArgs, reply *SuccessReply) error {
_, err := r.node.Relay().Publish(req.Context(), &args.Message, &args.Topic) var err error
if args.Topic == "" {
_, err = r.node.Relay().Publish(req.Context(), &args.Message)
} else {
_, err = r.node.Relay().PublishWithTopic(req.Context(), &args.Message, args.Topic)
}
if err != nil { if err != nil {
log.Error("Error publishing message:", err) log.Error("Error publishing message:", err)
reply.Success = false reply.Success = false
@ -91,7 +96,13 @@ func (r *RelayService) PostV1Message(req *http.Request, args *RelayMessageArgs,
func (r *RelayService) PostV1Subscription(req *http.Request, args *TopicsArgs, reply *SuccessReply) error { func (r *RelayService) PostV1Subscription(req *http.Request, args *TopicsArgs, reply *SuccessReply) error {
ctx := req.Context() ctx := req.Context()
for _, topic := range args.Topics { for _, topic := range args.Topics {
_, err := r.node.Relay().Subscribe(ctx, &topic) var err error
if topic == "" {
_, err = r.node.Relay().Subscribe(ctx)
} else {
_, err = r.node.Relay().SubscribeWithTopic(ctx, topic)
}
if err != nil { if err != nil {
log.Error("Error subscribing to topic:", topic, "err:", err) log.Error("Error subscribing to topic:", topic, "err:", err)
reply.Success = false reply.Success = false