go-waku/library/relay.go

109 lines
2.6 KiB
Go
Raw Normal View History

2023-08-10 13:30:38 +00:00
package library
import (
"context"
"time"
"github.com/ethereum/go-ethereum/common/hexutil"
"github.com/waku-org/go-waku/waku/v2/protocol"
"github.com/waku-org/go-waku/waku/v2/protocol/pb"
"github.com/waku-org/go-waku/waku/v2/protocol/relay"
)
2023-08-10 13:30:38 +00:00
// RelayEnoughPeers determines if there are enough peers to publish a message on a topic
func RelayEnoughPeers(topic string) (bool, error) {
if wakuState.node == nil {
2023-08-10 13:30:38 +00:00
return false, errWakuNodeNotReady
}
topicToCheck := protocol.DefaultPubsubTopic{}.String()
if topic != "" {
topicToCheck = topic
}
2023-08-10 13:30:38 +00:00
return wakuState.node.Relay().EnoughPeersToPublishToTopic(topicToCheck), nil
}
func relayPublish(msg *pb.WakuMessage, pubsubTopic string, ms int) (string, error) {
if wakuState.node == nil {
return "", errWakuNodeNotReady
}
var ctx context.Context
var cancel context.CancelFunc
if ms > 0 {
ctx, cancel = context.WithTimeout(context.Background(), time.Duration(int(ms))*time.Millisecond)
defer cancel()
} else {
ctx = context.Background()
}
hash, err := wakuState.node.Relay().Publish(ctx, msg, relay.WithPubSubTopic(pubsubTopic))
return hexutil.Encode(hash), err
}
2023-08-10 13:30:38 +00:00
// RelayPublish publishes a message using waku relay and returns the message ID
func RelayPublish(messageJSON string, topic string, ms int) (string, error) {
msg, err := wakuMessage(messageJSON)
if err != nil {
2023-08-10 13:30:38 +00:00
return "", err
}
2023-10-28 23:37:53 +00:00
return relayPublish(msg, topic, int(ms))
}
2023-10-28 23:37:53 +00:00
func relaySubscribe(filterJSON string) error {
cf, err := toContentFilter(filterJSON)
if err != nil {
return err
}
2023-10-28 23:37:53 +00:00
subscriptions, err := wakuState.node.Relay().Subscribe(context.Background(), cf)
if err != nil {
return err
}
2023-10-28 23:37:53 +00:00
for _, sub := range subscriptions {
go func(subscription *relay.Subscription) {
for envelope := range subscription.Ch {
send("message", toSubscriptionMessage(envelope))
}
}(sub)
}
return nil
}
2023-08-10 13:30:38 +00:00
// RelaySubscribe subscribes to a WakuRelay topic.
2023-10-28 23:37:53 +00:00
func RelaySubscribe(contentFilterJSON string) error {
if wakuState.node == nil {
2023-08-10 13:30:38 +00:00
return errWakuNodeNotReady
}
2023-10-28 23:37:53 +00:00
return relaySubscribe(contentFilterJSON)
}
2023-08-10 13:30:38 +00:00
// RelayTopics returns a list of pubsub topics the node is subscribed to in WakuRelay
func RelayTopics() (string, error) {
2023-03-29 19:20:31 +00:00
if wakuState.node == nil {
2023-08-10 13:30:38 +00:00
return "", errWakuNodeNotReady
2023-03-29 19:20:31 +00:00
}
2023-08-10 13:30:38 +00:00
return marshalJSON(wakuState.node.Relay().Topics())
2023-03-29 19:20:31 +00:00
}
2023-08-10 13:30:38 +00:00
// RelayUnsubscribe closes the pubsub subscription to a pubsub topic
2023-10-28 23:37:53 +00:00
func RelayUnsubscribe(contentFilterJSON string) error {
cf, err := toContentFilter(contentFilterJSON)
if err != nil {
return err
}
2023-10-28 23:37:53 +00:00
if wakuState.node == nil {
return errWakuNodeNotReady
}
2023-10-28 23:37:53 +00:00
return wakuState.node.Relay().Unsubscribe(context.Background(), cf)
}