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

fix: replace WIthTopic to ToTopic

Browse Source
This commit is contained in:
Richard Ramos 2021-11-19 20:03:05 -04:00
parent 3ac06f01ce
commit 2b2f9b754e
No known key found for this signature in database
GPG Key ID: 80D4B01265FDFE8F
9 changed files with 21 additions and 21 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
examples/filter2/main.go
View File

@ -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().SubscribeWithTopic(ctx, pubsubTopic) sub, err := wakuNode.Relay().SubscribeToTopic(ctx, pubsubTopic)
if err != nil { if err != nil {
log.Error("Could not subscribe: ", err) log.Error("Could not subscribe: ", err)
return return

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().SubscribeWithTopic(ctx, nodeTopic) _, err := wakuNode.Relay().SubscribeToTopic(ctx, nodeTopic)
failOnErr(err, "Error subscring to topic") failOnErr(err, "Error subscring to topic")
} }
} }

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.SubscribeWithTopic(context.Background(), topic) sub, err := relay.SubscribeToTopic(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.PublishWithTopic(ctx, tests.CreateWakuMessage(testContentTopic, 0), testTopic) _, err = node2.PublishToTopic(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.PublishWithTopic(ctx, tests.CreateWakuMessage("TopicB", 1), testTopic) _, err = node2.PublishToTopic(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.PublishWithTopic(ctx, tests.CreateWakuMessage(testContentTopic, 2), testTopic) _, err = node2.PublishToTopic(ctx, tests.CreateWakuMessage(testContentTopic, 2), testTopic)
require.NoError(t, err) require.NoError(t, err)
wg.Wait() wg.Wait()
} }

6
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.PublishWithTopic(wakuLP.ctx, message, pubSubTopic) _, err := wakuLP.relay.PublishToTopic(wakuLP.ctx, message, pubSubTopic)
if err != nil { if err != nil {
response.IsSuccess = false response.IsSuccess = false
@ -181,7 +181,7 @@ func (wakuLP *WakuLightPush) Stop() {
wakuLP.h.RemoveStreamHandler(LightPushID_v20beta1) wakuLP.h.RemoveStreamHandler(LightPushID_v20beta1)
} }
func (wakuLP *WakuLightPush) PublishWithTopic(ctx context.Context, message *pb.WakuMessage, topic string, opts ...LightPushOption) ([]byte, error) { func (wakuLP *WakuLightPush) PublishToTopic(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")
} }
@ -204,5 +204,5 @@ func (wakuLP *WakuLightPush) PublishWithTopic(ctx context.Context, message *pb.W
} }
func (wakuLP *WakuLightPush) Publish(ctx context.Context, message *pb.WakuMessage, opts ...LightPushOption) ([]byte, error) { func (wakuLP *WakuLightPush) Publish(ctx context.Context, message *pb.WakuMessage, opts ...LightPushOption) ([]byte, error) {
return wakuLP.PublishWithTopic(ctx, message, relay.DefaultWakuTopic, opts...) return wakuLP.PublishToTopic(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.SubscribeWithTopic(context.Background(), topic) sub, err := relay.SubscribeToTopic(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.PublishWithTopic(ctx, msg2, testTopic) hash, err := client.PublishToTopic(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.PublishWithTopic(ctx, tests.CreateWakuMessage("test", float64(0)), testTopic) _, err = client.PublishToTopic(ctx, tests.CreateWakuMessage("test", float64(0)), testTopic)
require.Errorf(t, err, "no suitable remote peers") require.Errorf(t, err, "no suitable remote peers")
} }

8
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) PublishWithTopic(ctx context.Context, message *pb.WakuMessage, topic string) ([]byte, error) { func (w *WakuRelay) PublishToTopic(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")
@ -175,7 +175,7 @@ func (w *WakuRelay) PublishWithTopic(ctx context.Context, message *pb.WakuMessag
} }
func (w *WakuRelay) Publish(ctx context.Context, message *pb.WakuMessage) ([]byte, error) { func (w *WakuRelay) Publish(ctx context.Context, message *pb.WakuMessage) ([]byte, error) {
return w.PublishWithTopic(ctx, message, DefaultWakuTopic) return w.PublishToTopic(ctx, message, DefaultWakuTopic)
} }
func (w *WakuRelay) Stop() { func (w *WakuRelay) Stop() {
@ -191,7 +191,7 @@ func (w *WakuRelay) Stop() {
w.subscriptions = nil w.subscriptions = nil
} }
func (w *WakuRelay) SubscribeWithTopic(ctx context.Context, topic string) (*Subscription, error) { func (w *WakuRelay) SubscribeToTopic(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.
sub, err := w.subscribe(topic) sub, err := w.subscribe(topic)
@ -221,7 +221,7 @@ func (w *WakuRelay) SubscribeWithTopic(ctx context.Context, topic string) (*Subs
} }
func (w *WakuRelay) Subscribe(ctx context.Context) (*Subscription, error) { func (w *WakuRelay) Subscribe(ctx context.Context) (*Subscription, error) {
return w.SubscribeWithTopic(ctx, DefaultWakuTopic) return w.SubscribeToTopic(ctx, DefaultWakuTopic)
} }
func (w *WakuRelay) Unsubscribe(ctx context.Context, topic string) error { func (w *WakuRelay) Unsubscribe(ctx context.Context, topic string) error {

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.PublishWithTopic(context.Background(), msg, testTopic) _, err = relay.PublishToTopic(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().SubscribeWithTopic(context.Background(), testTopic) _, err = n.Relay().SubscribeToTopic(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.SubscribeWithTopic(context.Background(), testTopic) _, err = node.SubscribeToTopic(context.Background(), testTopic)
require.NoError(t, err) require.NoError(t, err)
_ = filter.NewWakuFilter(context.Background(), host, false) _ = filter.NewWakuFilter(context.Background(), host, false)

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

@ -81,7 +81,7 @@ func (r *RelayService) PostV1Message(req *http.Request, args *RelayMessageArgs,
if args.Topic == "" { if args.Topic == "" {
_, err = r.node.Relay().Publish(req.Context(), &args.Message) _, err = r.node.Relay().Publish(req.Context(), &args.Message)
} else { } else {
_, err = r.node.Relay().PublishWithTopic(req.Context(), &args.Message, args.Topic) _, err = r.node.Relay().PublishToTopic(req.Context(), &args.Message, args.Topic)
} }
if err != nil { if err != nil {
log.Error("Error publishing message:", err) log.Error("Error publishing message:", err)
@ -101,7 +101,7 @@ func (r *RelayService) PostV1Subscription(req *http.Request, args *TopicsArgs, r
_, err = r.node.Relay().Subscribe(ctx) _, err = r.node.Relay().Subscribe(ctx)
} else { } else {
_, err = r.node.Relay().SubscribeWithTopic(ctx, topic) _, err = r.node.Relay().SubscribeToTopic(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)