go-waku/waku/v2/node/wakunode2.go

396 lines
8.6 KiB
Go
Raw Normal View History

2021-03-11 20:27:12 +00:00
package node
import (
"context"
2021-03-15 16:07:23 +00:00
"crypto/ecdsa"
"crypto/sha256"
2021-03-11 20:27:12 +00:00
"errors"
"fmt"
"net"
2021-03-15 16:07:23 +00:00
"sync"
2021-03-15 23:59:18 +00:00
"time"
2021-03-11 20:27:12 +00:00
2021-03-15 16:07:23 +00:00
proto "github.com/golang/protobuf/proto"
2021-03-22 16:45:13 +00:00
logging "github.com/ipfs/go-log"
2021-03-11 20:27:12 +00:00
"github.com/libp2p/go-libp2p"
2021-03-22 16:45:13 +00:00
connmgr "github.com/libp2p/go-libp2p-connmgr"
2021-03-11 20:27:12 +00:00
"github.com/libp2p/go-libp2p-core/crypto"
"github.com/libp2p/go-libp2p-core/host"
2021-03-18 16:40:47 +00:00
"github.com/libp2p/go-libp2p-core/peer"
2021-03-11 20:27:12 +00:00
ma "github.com/multiformats/go-multiaddr"
manet "github.com/multiformats/go-multiaddr-net"
common "github.com/status-im/go-waku/waku/common"
2021-03-11 20:27:12 +00:00
"github.com/status-im/go-waku/waku/v2/protocol"
2021-03-18 16:40:47 +00:00
store "github.com/status-im/go-waku/waku/v2/protocol/waku_store"
wakurelay "github.com/status-im/go-wakurelay-pubsub"
2021-03-11 20:27:12 +00:00
)
2021-03-22 16:45:13 +00:00
var log = logging.Logger("wakunode")
2021-03-11 20:27:12 +00:00
// Default clientId
const clientId string = "Go Waku v2 node"
type Topic string
2021-03-15 16:07:23 +00:00
const DefaultWakuTopic Topic = "/waku/2/default-waku/proto"
2021-03-11 20:27:12 +00:00
type Message []byte
type WakuInfo struct {
// NOTE One for simplicity, can extend later as needed
listenStr string
multiaddrStrings []byte
}
type MessagePair struct {
a *Topic
b *protocol.WakuMessage
}
type Subscription struct {
C chan *common.Envelope
closed bool
mutex sync.Mutex
pubSubscription *wakurelay.Subscription
quit chan struct{}
}
2021-03-11 20:27:12 +00:00
type WakuNode struct {
2021-03-15 16:07:23 +00:00
host host.Host
pubsub *wakurelay.PubSub
2021-03-18 16:40:47 +00:00
store *store.WakuStore
2021-03-15 16:07:23 +00:00
topics map[Topic]*wakurelay.Topic
2021-03-22 16:45:13 +00:00
topicsMutex sync.Mutex
2021-03-15 16:07:23 +00:00
2021-03-22 16:45:13 +00:00
subscriptions []*Subscription
subscriptionsMutex sync.Mutex
2021-03-15 16:07:23 +00:00
2021-03-22 16:45:13 +00:00
ctx context.Context
cancel context.CancelFunc
privKey crypto.PrivKey
2021-03-11 20:27:12 +00:00
}
func New(ctx context.Context, privKey *ecdsa.PrivateKey, hostAddr []net.Addr, opts ...libp2p.Option) (*WakuNode, error) {
2021-03-11 20:27:12 +00:00
// Creates a Waku Node.
if hostAddr == nil {
2021-03-22 16:45:13 +00:00
return nil, errors.New("host address cannot be null")
2021-03-11 20:27:12 +00:00
}
var multiAddresses []ma.Multiaddr
for _, addr := range hostAddr {
hostAddrMA, err := manet.FromNetAddr(addr)
2021-03-11 20:27:12 +00:00
if err != nil {
return nil, err
}
multiAddresses = append(multiAddresses, hostAddrMA)
2021-03-11 20:27:12 +00:00
}
2021-03-15 16:07:23 +00:00
nodeKey := crypto.PrivKey((*crypto.Secp256k1PrivateKey)(privKey))
2021-03-22 16:45:13 +00:00
opts = append(opts,
2021-03-11 20:27:12 +00:00
libp2p.ListenAddrs(multiAddresses...),
libp2p.Identity(nodeKey),
2021-03-22 16:45:13 +00:00
libp2p.DefaultTransports,
libp2p.NATPortMap(), // Attempt to open ports using uPNP for NATed hosts.
libp2p.EnableNATService(), // TODO: is this needed?)
libp2p.ConnectionManager(connmgr.NewConnManager(200, 300, 0)), // ?
)
2021-03-11 20:27:12 +00:00
2021-03-15 16:07:23 +00:00
ctx, cancel := context.WithCancel(ctx)
_ = cancel
2021-03-11 20:27:12 +00:00
host, err := libp2p.New(ctx, opts...)
if err != nil {
return nil, err
}
w := new(WakuNode)
2021-03-15 16:07:23 +00:00
w.pubsub = nil
w.host = host
w.cancel = cancel
w.privKey = nodeKey
w.ctx = ctx
w.topics = make(map[Topic]*wakurelay.Topic)
2021-03-11 20:27:12 +00:00
hostInfo, _ := ma.NewMultiaddr(fmt.Sprintf("/p2p/%s", host.ID().Pretty()))
for _, addr := range host.Addrs() {
fullAddr := addr.Encapsulate(hostInfo)
2021-03-22 16:45:13 +00:00
log.Info("Listening on", fullAddr)
2021-03-11 20:27:12 +00:00
}
return w, nil
}
2021-03-22 16:45:13 +00:00
func (w *WakuNode) Stop() {
w.subscriptionsMutex.Lock()
defer w.subscriptionsMutex.Unlock()
2021-03-15 16:07:23 +00:00
defer w.cancel()
2021-03-22 16:45:13 +00:00
for _, sub := range w.subscriptions {
sub.Unsubscribe()
}
w.subscriptions = nil
2021-03-15 16:07:23 +00:00
}
func (w *WakuNode) Host() host.Host {
return w.host
}
func (w *WakuNode) ID() string {
return w.host.ID().Pretty()
}
func (w *WakuNode) PubSub() *wakurelay.PubSub {
2021-03-15 16:07:23 +00:00
return w.pubsub
}
func (w *WakuNode) SetPubSub(pubSub *wakurelay.PubSub) {
2021-03-15 16:07:23 +00:00
w.pubsub = pubSub
}
func (w *WakuNode) MountRelay(opts ...wakurelay.Option) error {
ps, err := wakurelay.NewWakuRelaySub(w.ctx, w.host, opts...)
2021-03-15 16:07:23 +00:00
if err != nil {
return err
}
w.pubsub = ps
// TODO: filters
// TODO: rlnRelay
return nil
}
2021-03-22 16:45:13 +00:00
func (w *WakuNode) MountStore(s store.MessageProvider) error {
2021-03-18 16:40:47 +00:00
sub, err := w.Subscribe(nil)
if err != nil {
return err
}
2021-03-22 16:45:13 +00:00
w.store = store.NewWakuStore(w.ctx, w.host, sub.C, s)
2021-03-18 16:40:47 +00:00
return nil
}
func (w *WakuNode) StartStore() error {
if w.store == nil {
return errors.New("WakuStore is not set")
}
w.store.Start()
return nil
}
2021-03-18 16:40:47 +00:00
func (w *WakuNode) AddStorePeer(address string) error {
if w.store == nil {
return errors.New("WakuStore is not set")
}
storePeer, err := ma.NewMultiaddr(address)
if err != nil {
return err
}
// Extract the peer ID from the multiaddr.
info, err := peer.AddrInfoFromP2pAddr(storePeer)
if err != nil {
return err
}
return w.store.AddPeer(info.ID, info.Addrs)
}
func (w *WakuNode) Query(contentTopic uint32, asc bool, pageSize int64) (*protocol.HistoryResponse, error) {
if w.store == nil {
return nil, errors.New("WakuStore is not set")
}
query := new(protocol.HistoryQuery)
query.Topics = append(query.Topics, contentTopic)
query.PagingInfo = new(protocol.PagingInfo)
if asc {
query.PagingInfo.Direction = protocol.PagingInfo_FORWARD
} else {
query.PagingInfo.Direction = protocol.PagingInfo_BACKWARD
}
query.PagingInfo.PageSize = pageSize
result, err := w.store.Query(query)
if err != nil {
return nil, err
}
return result, nil
}
func (node *WakuNode) Subscribe(topic *Topic) (*Subscription, error) {
2021-03-18 16:40:47 +00:00
// Subscribes to a PubSub topic.
2021-03-15 16:07:23 +00:00
// NOTE The data field SHOULD be decoded as a WakuMessage.
if node.pubsub == nil {
return nil, errors.New("PubSub hasn't been set. Execute mountWakuRelay() or setPubSub() first")
}
pubSubTopic, err := node.upsertTopic(topic)
if err != nil {
return nil, err
}
sub, err := pubSubTopic.Subscribe()
if err != nil {
return nil, err
}
subscription := new(Subscription)
subscription.closed = false
subscription.pubSubscription = sub
subscription.C = make(chan *common.Envelope)
subscription.quit = make(chan struct{})
2021-03-15 16:07:23 +00:00
go func(ctx context.Context, sub *wakurelay.Subscription) {
2021-03-15 23:59:18 +00:00
nextMsgTicker := time.NewTicker(time.Millisecond * 10)
defer nextMsgTicker.Stop()
2021-03-15 16:07:23 +00:00
for {
select {
case <-subscription.quit:
subscription.mutex.Lock()
defer subscription.mutex.Unlock()
close(subscription.C)
subscription.closed = true
2021-03-15 16:07:23 +00:00
return
2021-03-15 23:59:18 +00:00
case <-nextMsgTicker.C:
msg, err := sub.Next(ctx)
if err != nil {
2021-03-22 16:45:13 +00:00
log.Error("error receiving message", err)
2021-03-18 16:40:47 +00:00
close(subscription.quit)
return
}
2021-03-15 16:07:23 +00:00
wakuMessage := &protocol.WakuMessage{}
if err := proto.Unmarshal(msg.Data, wakuMessage); err != nil {
log.Error("could not decode message", err)
return
}
2021-03-15 16:07:23 +00:00
envelope := common.NewEnvelope(wakuMessage, len(msg.Data), sha256.Sum256(msg.Data))
subscription.C <- envelope
}
2021-03-15 16:07:23 +00:00
}
}(node.ctx, sub)
2021-03-22 16:45:13 +00:00
node.subscriptionsMutex.Lock()
defer node.subscriptionsMutex.Unlock()
node.subscriptions = append(node.subscriptions, subscription)
return subscription, nil
2021-03-15 16:07:23 +00:00
}
func (subs *Subscription) Unsubscribe() {
2021-03-15 16:07:23 +00:00
// Unsubscribes a handler from a PubSub topic.
subs.mutex.Lock()
defer subs.mutex.Unlock()
if !subs.closed {
close(subs.quit)
}
2021-03-15 16:07:23 +00:00
}
func (subs *Subscription) IsClosed() bool {
subs.mutex.Lock()
defer subs.mutex.Unlock()
return subs.closed
}
func (node *WakuNode) upsertTopic(topic *Topic) (*wakurelay.Topic, error) {
2021-03-22 16:45:13 +00:00
defer node.topicsMutex.Unlock()
node.topicsMutex.Lock()
2021-03-15 16:07:23 +00:00
var t Topic = DefaultWakuTopic
if topic != nil {
t = *topic
}
pubSubTopic, ok := node.topics[t]
if !ok { // Joins topic if node hasn't joined yet
newTopic, err := node.pubsub.Join(string(t))
if err != nil {
return nil, err
}
node.topics[t] = newTopic
pubSubTopic = newTopic
}
return pubSubTopic, nil
}
func (node *WakuNode) Publish(message *protocol.WakuMessage, topic *Topic) error {
// Publish a `WakuMessage` to a PubSub topic. `WakuMessage` should contain a
// `contentTopic` field for light node functionality. This field may be also
// be omitted.
if node.pubsub == nil {
return errors.New("PubSub hasn't been set. Execute mountWakuRelay() or setPubSub() first")
}
if message == nil {
2021-03-22 16:45:13 +00:00
return errors.New("message can't be null")
2021-03-15 16:07:23 +00:00
}
pubSubTopic, err := node.upsertTopic(topic)
2021-03-22 16:45:13 +00:00
2021-03-15 16:07:23 +00:00
if err != nil {
return err
}
out, err := proto.Marshal(message)
if err != nil {
return err
}
err = pubSubTopic.Publish(node.ctx, out)
2021-03-22 16:45:13 +00:00
2021-03-15 16:07:23 +00:00
if err != nil {
return err
}
2021-03-11 20:27:12 +00:00
2021-03-15 16:07:23 +00:00
return nil
2021-03-11 20:27:12 +00:00
}
func (w *WakuNode) DialPeer(address string) error {
p, err := ma.NewMultiaddr(address)
if err != nil {
return err
}
// Extract the peer ID from the multiaddr.
info, err := peer.AddrInfoFromP2pAddr(p)
if err != nil {
return err
}
w.host.Connect(w.ctx, *info)
return nil
}
func (w *WakuNode) ClosePeerByAddress(address string) error {
p, err := ma.NewMultiaddr(address)
if err != nil {
return err
}
// Extract the peer ID from the multiaddr.
info, err := peer.AddrInfoFromP2pAddr(p)
if err != nil {
return err
}
return w.ClosePeerById(info.ID)
}
func (w *WakuNode) ClosePeerById(id peer.ID) error {
return w.host.Network().ClosePeer(id)
}
func (w *WakuNode) PeerCount() int {
return len(w.host.Network().Peers())
}