go-waku/waku/v2/rpc/relay.go

136 lines
3.1 KiB
Go
Raw Normal View History

2021-11-06 10:49:47 +00:00
package rpc
import (
"fmt"
2021-11-06 10:49:47 +00:00
"net/http"
2021-11-18 14:20:58 +00:00
"sync"
2021-11-06 10:49:47 +00:00
"github.com/status-im/go-waku/waku/v2/node"
2021-11-18 14:20:58 +00:00
"github.com/status-im/go-waku/waku/v2/protocol"
2021-11-06 10:49:47 +00:00
"github.com/status-im/go-waku/waku/v2/protocol/pb"
"go.uber.org/zap"
2021-11-06 10:49:47 +00:00
)
type RelayService struct {
node *node.WakuNode
2021-11-18 14:20:58 +00:00
log *zap.SugaredLogger
2021-11-18 14:20:58 +00:00
messages map[string][]*pb.WakuMessage
messagesMutex sync.RWMutex
2021-11-22 14:48:32 +00:00
runner *runnerService
2021-11-06 10:49:47 +00:00
}
type RelayMessageArgs struct {
Topic string `json:"topic,omitempty"`
Message pb.WakuMessage `json:"message,omitempty"`
}
type TopicsArgs struct {
Topics []string `json:"topics,omitempty"`
}
2021-11-18 14:20:58 +00:00
type TopicArgs struct {
Topic string `json:"topic,omitempty"`
}
func NewRelayService(node *node.WakuNode, log *zap.SugaredLogger) *RelayService {
2021-11-22 14:48:32 +00:00
s := &RelayService{
2021-11-18 14:20:58 +00:00
node: node,
log: log.Named("relay"),
2021-11-18 14:20:58 +00:00
messages: make(map[string][]*pb.WakuMessage),
}
2021-11-22 14:48:32 +00:00
s.runner = newRunnerService(node.Broadcaster(), s.addEnvelope)
return s
2021-11-18 14:20:58 +00:00
}
func (r *RelayService) addEnvelope(envelope *protocol.Envelope) {
r.messagesMutex.Lock()
defer r.messagesMutex.Unlock()
if _, ok := r.messages[envelope.PubsubTopic()]; !ok {
return
}
r.messages[envelope.PubsubTopic()] = append(r.messages[envelope.PubsubTopic()], envelope.Message())
}
func (r *RelayService) Start() {
2021-11-22 14:48:32 +00:00
r.runner.Start()
2021-11-18 14:20:58 +00:00
}
func (r *RelayService) Stop() {
2021-11-22 14:48:32 +00:00
r.runner.Stop()
2021-11-18 14:20:58 +00:00
}
2021-11-06 10:49:47 +00:00
func (r *RelayService) PostV1Message(req *http.Request, args *RelayMessageArgs, reply *SuccessReply) error {
var err error
if args.Topic == "" {
_, err = r.node.Relay().Publish(req.Context(), &args.Message)
} else {
2021-11-20 00:03:05 +00:00
_, err = r.node.Relay().PublishToTopic(req.Context(), &args.Message, args.Topic)
}
2021-11-06 10:49:47 +00:00
if err != nil {
r.log.Error("Error publishing message: ", err)
2021-11-06 10:49:47 +00:00
reply.Success = false
reply.Error = err.Error()
} else {
reply.Success = true
}
return nil
}
func (r *RelayService) PostV1Subscription(req *http.Request, args *TopicsArgs, reply *SuccessReply) error {
ctx := req.Context()
for _, topic := range args.Topics {
var err error
if topic == "" {
_, err = r.node.Relay().Subscribe(ctx)
} else {
2021-11-20 00:03:05 +00:00
_, err = r.node.Relay().SubscribeToTopic(ctx, topic)
}
2021-11-06 10:49:47 +00:00
if err != nil {
r.log.Error("Error subscribing to topic:", topic, "err:", err)
2021-11-06 10:49:47 +00:00
reply.Success = false
reply.Error = err.Error()
return nil
}
2021-11-18 14:20:58 +00:00
r.messages[topic] = make([]*pb.WakuMessage, 0)
2021-11-06 10:49:47 +00:00
}
reply.Success = true
return nil
}
func (r *RelayService) DeleteV1Subscription(req *http.Request, args *TopicsArgs, reply *SuccessReply) error {
ctx := req.Context()
for _, topic := range args.Topics {
2021-11-19 16:19:48 +00:00
err := r.node.Relay().Unsubscribe(ctx, topic)
2021-11-06 10:49:47 +00:00
if err != nil {
r.log.Error("Error unsubscribing from topic", topic, "err:", err)
2021-11-06 10:49:47 +00:00
reply.Success = false
reply.Error = err.Error()
return nil
}
2021-11-18 14:20:58 +00:00
delete(r.messages, topic)
2021-11-06 10:49:47 +00:00
}
reply.Success = true
return nil
}
2021-11-18 14:20:58 +00:00
func (r *RelayService) GetV1Messages(req *http.Request, args *TopicArgs, reply *MessagesReply) error {
r.messagesMutex.Lock()
defer r.messagesMutex.Unlock()
if _, ok := r.messages[args.Topic]; !ok {
return fmt.Errorf("topic %s not subscribed", args.Topic)
}
reply.Messages = r.messages[args.Topic]
r.messages[args.Topic] = make([]*pb.WakuMessage, 0)
return nil
}