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

887 lines
22 KiB
Go
Raw Normal View History

2021-03-11 20:27:12 +00:00
package node
import (
"context"
"errors"
"fmt"
2021-09-11 11:36:54 +00:00
//"log/syslog"
//"strconv"
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-09-11 11:36:54 +00:00
"github.com/libp2p/go-libp2p/p2p/protocol/identify"
2021-08-31 18:19:49 +00:00
2021-06-16 10:14:22 +00:00
"github.com/libp2p/go-libp2p-core/event"
2021-03-11 20:27:12 +00:00
"github.com/libp2p/go-libp2p-core/host"
2021-06-16 10:14:22 +00:00
"github.com/libp2p/go-libp2p-core/network"
2021-03-18 16:40:47 +00:00
"github.com/libp2p/go-libp2p-core/peer"
2021-08-31 18:19:49 +00:00
"github.com/libp2p/go-libp2p-core/peerstore"
p2pproto "github.com/libp2p/go-libp2p-core/protocol"
"github.com/libp2p/go-libp2p/p2p/protocol/ping"
2021-03-11 20:27:12 +00:00
ma "github.com/multiformats/go-multiaddr"
"go.opencensus.io/stats"
"go.opencensus.io/tag"
2021-04-22 00:09:37 +00:00
"github.com/status-im/go-waku/waku/v2/metrics"
2021-03-11 20:27:12 +00:00
"github.com/status-im/go-waku/waku/v2/protocol"
2021-06-10 12:59:51 +00:00
"github.com/status-im/go-waku/waku/v2/protocol/filter"
"github.com/status-im/go-waku/waku/v2/protocol/lightpush"
2021-04-22 00:09:37 +00:00
"github.com/status-im/go-waku/waku/v2/protocol/pb"
"github.com/status-im/go-waku/waku/v2/protocol/relay"
2021-04-22 00:09:37 +00:00
"github.com/status-im/go-waku/waku/v2/protocol/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-09-11 11:36:54 +00:00
//var logwriter, _ = syslog.New(syslog.LOG_ERR|syslog.LOG_LOCAL0, "WAKU")
2021-03-11 20:27:12 +00:00
type Message []byte
2021-06-16 10:14:22 +00:00
// A map of peer IDs to supported protocols
type PeerStats map[peer.ID][]string
type ConnStatus struct {
IsOnline bool
HasHistory bool
Peers PeerStats
2021-06-16 10:14:22 +00:00
}
2021-03-11 20:27:12 +00:00
type WakuNode struct {
2021-09-11 11:36:54 +00:00
host host.Host
idService *identify.IDService
opts *WakuNodeParameters
2021-03-15 16:07:23 +00:00
relay *relay.WakuRelay
2021-06-10 12:59:51 +00:00
filter *filter.WakuFilter
lightPush *lightpush.WakuLightPush
ping *ping.PingService
subscriptions map[relay.Topic][]*Subscription
2021-03-22 16:45:13 +00:00
subscriptionsMutex sync.Mutex
2021-03-15 16:07:23 +00:00
bcaster Broadcaster
2021-06-10 12:59:51 +00:00
filters filter.Filters
2021-06-16 10:14:22 +00:00
connectednessEventSub event.Subscription
protocolEventSub event.Subscription
identificationEventSub event.Subscription
ctx context.Context
cancel context.CancelFunc
quit chan struct{}
2021-06-16 10:14:22 +00:00
// Map of peers and their supported protocols
peers PeerStats
peersMutex sync.Mutex
2021-06-16 10:14:22 +00:00
// Internal protocol implementations that wish
// to listen to peer added/removed events (e.g. Filter)
peerListeners []chan *event.EvtPeerConnectednessChanged
// Channel passed to WakuNode constructor
// receiving connection status notifications
connStatusChan chan ConnStatus
2021-08-10 14:23:49 +00:00
pingEventsChan chan interface{}
2021-06-16 10:14:22 +00:00
}
2021-08-10 14:23:49 +00:00
func (w *WakuNode) handleConnectednessChanged(ev event.EvtPeerConnectednessChanged) {
log.Debug("### EvtPeerConnectednessChanged ", w.Host().ID(), " to ", ev.Peer, " : ", ev.Connectedness)
w.peersMutex.Lock()
defer w.peersMutex.Unlock()
2021-08-10 14:23:49 +00:00
if ev.Connectedness == network.Connected {
_, ok := w.peers[ev.Peer]
if !ok {
peerProtocols, _ := w.host.Peerstore().GetProtocols(ev.Peer)
log.Debug("protocols found for peer: ", ev.Peer, ", protocols: ", peerProtocols)
2021-08-10 14:23:49 +00:00
w.peers[ev.Peer] = peerProtocols
} else {
log.Debug("### Peer already exists")
2021-08-10 14:23:49 +00:00
}
} else if ev.Connectedness == network.NotConnected {
log.Debug("Peer down: ", ev.Peer)
2021-08-10 14:23:49 +00:00
delete(w.peers, ev.Peer)
// for _, pl := range w.peerListeners {
// pl <- &ev
// }
// TODO
// There seems to be no proper way to
// remove a dropped peer from Host's Peerstore
// https://github.com/libp2p/go-libp2p-host/issues/13
//w.Host().Network().ClosePeer(ev.Peer)
}
2021-06-16 10:14:22 +00:00
2021-08-10 14:23:49 +00:00
}
func (w *WakuNode) handleProtocolsUpdated(ev event.EvtPeerProtocolsUpdated) {
log.Debug("### EvtPeerProtocolsUpdated ", w.Host().ID(), " to ", ev.Peer, " added: ", ev.Added, ", removed: ", ev.Removed)
w.peersMutex.Lock()
defer w.peersMutex.Unlock()
2021-08-10 14:23:49 +00:00
_, ok := w.peers[ev.Peer]
if ok {
peerProtocols, _ := w.host.Peerstore().GetProtocols(ev.Peer)
log.Debug("updated protocols found for peer: ", ev.Peer, ", protocols: ", peerProtocols)
2021-08-10 14:23:49 +00:00
w.peers[ev.Peer] = peerProtocols
}
2021-06-16 10:14:22 +00:00
2021-08-10 14:23:49 +00:00
}
func (w *WakuNode) handlePeerIdentificationCompleted(ev event.EvtPeerIdentificationCompleted) {
log.Debug("### EvtPeerIdentificationCompleted ", w.Host().ID(), " to ", ev.Peer)
w.peersMutex.Lock()
defer w.peersMutex.Unlock()
2021-08-10 14:23:49 +00:00
peerProtocols, _ := w.host.Peerstore().GetProtocols(ev.Peer)
log.Debug("identified protocols found for peer: ", ev.Peer, ", protocols: ", peerProtocols)
2021-09-11 11:36:54 +00:00
w.peers[ev.Peer] = peerProtocols
2021-08-10 14:23:49 +00:00
}
func (w *WakuNode) processHostEvent(e interface{}) {
if e == nil {
log.Debug("processHostEvent nil event")
2021-08-10 14:23:49 +00:00
return
}
isOnline := w.IsOnline()
hasHistory := w.HasHistory()
switch e := e.(type) {
2021-08-10 14:23:49 +00:00
case event.EvtPeerConnectednessChanged:
w.handleConnectednessChanged(e)
2021-08-10 14:23:49 +00:00
case event.EvtPeerProtocolsUpdated:
w.handleProtocolsUpdated(e)
2021-08-10 14:23:49 +00:00
case event.EvtPeerIdentificationCompleted:
w.handlePeerIdentificationCompleted(e)
2021-08-10 14:23:49 +00:00
}
log.Debug("###processHostEvent before isOnline()")
2021-08-10 14:23:49 +00:00
newIsOnline := w.IsOnline()
log.Debug("###processHostEvent before hasHistory()")
2021-08-10 14:23:49 +00:00
newHasHistory := w.HasHistory()
log.Debug("###ConnStatus isOnline: ", isOnline, "/", newIsOnline, " hasHistory: ",
2021-08-10 14:23:49 +00:00
hasHistory, "/", newHasHistory)
if w.connStatusChan != nil {
connStatus := ConnStatus{IsOnline: newIsOnline, HasHistory: newHasHistory, Peers: w.Peers()}
log.Debug("New ConnStatus: ", connStatus)
w.connStatusChan <- connStatus
2021-08-10 14:23:49 +00:00
}
}
func (w *WakuNode) connectednessListener() {
for {
var e interface{}
log.Debug("connectednessListener before select")
2021-08-10 14:23:49 +00:00
select {
case e = <-w.connectednessEventSub.Out():
log.Debug("connectednessListener connectednessEvent")
2021-08-10 14:23:49 +00:00
case e = <-w.protocolEventSub.Out():
log.Debug("connectednessListener protocolEvent")
2021-08-10 14:23:49 +00:00
case e = <-w.identificationEventSub.Out():
log.Debug("connectednessListener identificationEvent")
2021-08-10 14:23:49 +00:00
case e = <-w.pingEventsChan:
log.Debug("connectednessListener pingEvent")
2021-06-16 10:14:22 +00:00
}
log.Debug("connectednessListener after select")
2021-08-10 14:23:49 +00:00
w.processHostEvent(e)
log.Debug("connectednessListener after processHostEvent")
2021-06-16 10:14:22 +00:00
}
2021-03-11 20:27:12 +00:00
}
func New(ctx context.Context, opts ...WakuNodeOption) (*WakuNode, error) {
params := new(WakuNodeParameters)
2021-03-11 20:27:12 +00:00
ctx, cancel := context.WithCancel(ctx)
_ = cancel
params.libP2POpts = DefaultLibP2POptions
for _, opt := range opts {
err := opt(params)
2021-03-11 20:27:12 +00:00
if err != nil {
return nil, err
}
}
if len(params.multiAddr) > 0 {
params.libP2POpts = append(params.libP2POpts, libp2p.ListenAddrs(params.multiAddr...))
}
2021-03-11 20:27:12 +00:00
if params.privKey != nil {
params.libP2POpts = append(params.libP2POpts, libp2p.Identity(*params.privKey))
}
2021-03-15 16:07:23 +00:00
host, err := libp2p.New(ctx, params.libP2POpts...)
2021-03-11 20:27:12 +00:00
if err != nil {
return nil, err
}
w := new(WakuNode)
w.bcaster = NewBroadcaster(1024)
2021-03-15 16:07:23 +00:00
w.host = host
2021-09-11 11:36:54 +00:00
w.idService = identify.NewIDService(host)
2021-03-15 16:07:23 +00:00
w.cancel = cancel
w.ctx = ctx
w.subscriptions = make(map[relay.Topic][]*Subscription)
w.opts = params
w.quit = make(chan struct{})
2021-06-16 10:14:22 +00:00
w.peers = make(PeerStats)
2021-06-16 10:14:22 +00:00
// Subscribe to Connectedness events
connectednessEventSub, _ := host.EventBus().Subscribe(new(event.EvtPeerConnectednessChanged))
w.connectednessEventSub = connectednessEventSub
protocolEventSub, _ := host.EventBus().Subscribe(new(event.EvtPeerProtocolsUpdated))
w.protocolEventSub = protocolEventSub
identificationEventSub, _ := host.EventBus().Subscribe(new(event.EvtPeerIdentificationCompleted))
w.identificationEventSub = identificationEventSub
if params.connStatusChan != nil {
w.connStatusChan = params.connStatusChan
2021-06-28 14:14:28 +00:00
}
2021-08-10 14:23:49 +00:00
w.pingEventsChan = make(chan interface{})
2021-06-16 10:14:22 +00:00
go w.connectednessListener()
2021-06-28 14:14:28 +00:00
2021-06-16 10:14:22 +00:00
if params.enableStore {
w.startStore()
}
2021-06-10 12:59:51 +00:00
if params.enableFilter {
w.filters = make(filter.Filters)
err := w.mountFilter()
if err != nil {
return nil, err
}
}
2021-06-16 10:14:22 +00:00
err = w.mountRelay(params.enableRelay, params.wOpts...)
if err != nil {
return nil, err
}
if params.enableLightPush {
w.mountLightPush()
}
if params.keepAliveInterval > time.Duration(0) {
w.startKeepAlive(params.keepAliveInterval)
}
2021-04-04 17:05:33 +00:00
for _, addr := range w.ListenAddresses() {
2021-04-04 19:33:21 +00:00
log.Info("Listening on ", addr)
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
close(w.quit)
2021-06-16 10:14:22 +00:00
defer w.connectednessEventSub.Close()
defer w.protocolEventSub.Close()
defer w.identificationEventSub.Close()
for _, topic := range w.relay.Topics() {
for _, sub := range w.subscriptions[topic] {
sub.Unsubscribe()
}
2021-03-22 16:45:13 +00:00
}
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()
}
2021-06-16 10:14:22 +00:00
func (w *WakuNode) IsOnline() bool {
w.peersMutex.Lock()
defer w.peersMutex.Unlock()
2021-06-16 10:14:22 +00:00
hasRelay := false
hasLightPush := false
hasStore := false
hasFilter := false
for _, v := range w.peers {
for _, protocol := range v {
if !hasRelay && protocol == string(wakurelay.WakuRelayID_v200) {
2021-06-16 10:14:22 +00:00
hasRelay = true
}
if !hasLightPush && protocol == string(lightpush.WakuLightPushProtocolId) {
hasLightPush = true
}
if !hasStore && protocol == string(store.WakuStoreProtocolId) {
hasStore = true
}
if !hasFilter && protocol == string(filter.WakuFilterProtocolId) {
hasFilter = true
}
if hasRelay || hasLightPush && (hasStore || hasFilter) {
return true
}
}
}
return false
}
func (w *WakuNode) HasHistory() bool {
w.peersMutex.Lock()
defer w.peersMutex.Unlock()
2021-06-16 10:14:22 +00:00
for _, v := range w.peers {
for _, protocol := range v {
if protocol == string(store.WakuStoreProtocolId) {
return true
}
}
}
return false
}
func (w *WakuNode) ListenAddresses() []ma.Multiaddr {
2021-04-04 17:05:33 +00:00
hostInfo, _ := ma.NewMultiaddr(fmt.Sprintf("/p2p/%s", w.host.ID().Pretty()))
var result []ma.Multiaddr
2021-04-04 17:05:33 +00:00
for _, addr := range w.host.Addrs() {
result = append(result, addr.Encapsulate(hostInfo))
2021-04-04 17:05:33 +00:00
}
return result
}
func (w *WakuNode) Relay() *relay.WakuRelay {
return w.relay
2021-03-15 16:07:23 +00:00
}
2021-06-10 12:59:51 +00:00
func (w *WakuNode) Filter() *filter.WakuFilter {
return w.filter
}
2021-06-28 14:14:28 +00:00
func (w *WakuNode) mountRelay(shouldRelayMessages bool, opts ...wakurelay.Option) error {
var err error
w.relay, err = relay.NewWakuRelay(w.ctx, w.host, opts...)
2021-03-15 16:07:23 +00:00
2021-06-28 14:14:28 +00:00
if shouldRelayMessages {
_, err := w.Subscribe(nil)
if err != nil {
return err
}
}
2021-03-15 16:07:23 +00:00
// TODO: rlnRelay
return err
2021-03-18 16:40:47 +00:00
}
2021-06-10 12:59:51 +00:00
func (w *WakuNode) mountFilter() error {
filterHandler := func(requestId string, msg pb.MessagePush) {
for _, message := range msg.Messages {
w.filters.Notify(message, requestId) // Trigger filter handlers on a light node
}
}
2021-06-16 10:14:22 +00:00
peerChan := make(chan *event.EvtPeerConnectednessChanged)
w.filter = filter.NewWakuFilter(w.ctx, w.host, filterHandler, peerChan)
w.peerListeners = append(w.peerListeners, peerChan)
2021-06-10 12:59:51 +00:00
return nil
}
func (w *WakuNode) mountLightPush() {
w.lightPush = lightpush.NewWakuLightPush(w.ctx, w.host, w.relay)
}
2021-08-31 18:19:49 +00:00
func (w *WakuNode) AddPeer(info *peer.AddrInfo, protocolId string) error {
log.Info(fmt.Sprintf("adding peer %s with protocol %s", info.ID.Pretty(), protocolId))
2021-06-16 10:14:22 +00:00
2021-08-31 18:19:49 +00:00
w.host.Peerstore().AddAddrs(info.ID, info.Addrs, peerstore.PermanentAddrTTL)
2021-06-16 10:14:22 +00:00
2021-08-31 18:19:49 +00:00
return w.host.Peerstore().AddProtocols(info.ID, protocolId)
2021-06-16 10:14:22 +00:00
}
2021-06-28 14:14:28 +00:00
func (w *WakuNode) startStore() {
2021-06-16 10:14:22 +00:00
peerChan := make(chan *event.EvtPeerConnectednessChanged)
w.opts.store.Start(w.ctx, w.host, peerChan)
w.peerListeners = append(w.peerListeners, peerChan)
if w.opts.shouldResume {
if _, err := w.opts.store.Resume(string(relay.GetTopic(nil)), nil); err != nil {
log.Error("failed to resume", err)
}
2021-08-13 11:56:09 +00:00
}
}
2021-08-31 18:19:49 +00:00
func (w *WakuNode) addPeerWithProtocol(address string, proto p2pproto.ID) (*peer.ID, error) {
info, err := addrInfoFromMultiaddrString(address)
2021-03-18 16:40:47 +00:00
if err != nil {
return nil, err
2021-03-18 16:40:47 +00:00
}
2021-08-31 18:19:49 +00:00
return &info.ID, w.AddPeer(info, string(proto))
2021-03-18 16:40:47 +00:00
}
2021-08-31 18:19:49 +00:00
func (w *WakuNode) AddStorePeer(address string) (*peer.ID, error) {
return w.addPeerWithProtocol(address, store.WakuStoreProtocolId)
}
2021-06-10 12:59:51 +00:00
2021-08-31 18:19:49 +00:00
func (w *WakuNode) AddRelayPeer(address string) (*peer.ID, error) {
return w.addPeerWithProtocol(address, wakurelay.WakuRelayID_v200)
}
2021-06-10 12:59:51 +00:00
2021-08-31 18:19:49 +00:00
func (w *WakuNode) AddFilterPeer(address string) (*peer.ID, error) {
return w.addPeerWithProtocol(address, filter.WakuFilterProtocolId)
2021-06-10 12:59:51 +00:00
}
2021-06-28 14:14:28 +00:00
func (w *WakuNode) AddLightPushPeer(address string) (*peer.ID, error) {
2021-08-31 18:19:49 +00:00
return w.addPeerWithProtocol(address, lightpush.WakuLightPushProtocolId)
2021-06-28 14:14:28 +00:00
}
2021-04-22 00:09:37 +00:00
func (w *WakuNode) Query(ctx context.Context, contentTopics []string, startTime float64, endTime float64, opts ...store.HistoryRequestOption) (*pb.HistoryResponse, error) {
if w.opts.store == nil {
return nil, errors.New("WakuStore is not set")
}
2021-04-22 00:09:37 +00:00
query := new(pb.HistoryQuery)
2021-04-20 21:46:35 +00:00
for _, ct := range contentTopics {
2021-04-22 00:09:37 +00:00
query.ContentFilters = append(query.ContentFilters, &pb.ContentFilter{ContentTopic: ct})
2021-04-20 21:46:35 +00:00
}
query.StartTime = startTime
query.EndTime = endTime
2021-04-22 00:09:37 +00:00
query.PagingInfo = new(pb.PagingInfo)
result, err := w.opts.store.Query(ctx, query, opts...)
if err != nil {
return nil, err
}
return result, nil
}
2021-06-10 13:00:06 +00:00
func (w *WakuNode) Resume(ctx context.Context, peerList []peer.ID) error {
if w.opts.store == nil {
return errors.New("WakuStore is not set")
}
result, err := w.opts.store.Resume(string(relay.DefaultWakuTopic), peerList)
2021-06-10 13:00:06 +00:00
if err != nil {
return err
}
log.Info("the number of retrieved messages since the last online time: ", result)
return nil
}
func (node *WakuNode) Subscribe(topic *relay.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.relay == nil {
2021-06-28 14:14:28 +00:00
return nil, errors.New("WakuRelay hasn't been set")
2021-03-15 16:07:23 +00:00
}
t := relay.GetTopic(topic)
sub, isNew, err := node.relay.Subscribe(t)
// Subscribe store to topic
if isNew && node.opts.store != nil && node.opts.storeMsgs {
log.Info("Subscribing store to topic ", t)
node.bcaster.Register(node.opts.store.MsgC)
}
2021-03-15 16:07:23 +00:00
2021-06-10 12:59:51 +00:00
// Subscribe filter
if isNew && node.filter != nil {
log.Info("Subscribing filter to topic ", t)
node.bcaster.Register(node.filter.MsgC)
}
2021-03-15 16:07:23 +00:00
if err != nil {
return nil, err
}
// Create client subscription
subscription := new(Subscription)
subscription.closed = false
2021-04-22 00:09:37 +00:00
subscription.C = make(chan *protocol.Envelope, 1024) // To avoid blocking
subscription.quit = make(chan struct{})
2021-03-15 16:07:23 +00:00
node.subscriptionsMutex.Lock()
defer node.subscriptionsMutex.Unlock()
node.subscriptions[t] = append(node.subscriptions[t], subscription)
node.bcaster.Register(subscription.C)
go func(t relay.Topic) {
2021-03-15 23:59:18 +00:00
nextMsgTicker := time.NewTicker(time.Millisecond * 10)
defer nextMsgTicker.Stop()
ctx, err := tag.New(node.ctx, tag.Insert(metrics.KeyType, "relay"))
if err != nil {
log.Error(err)
return
}
2021-03-15 16:07:23 +00:00
for {
select {
case <-subscription.quit:
subscription.mutex.Lock()
node.bcaster.Unregister(subscription.C) // Remove from broadcast list
close(subscription.C)
subscription.mutex.Unlock()
2021-03-15 23:59:18 +00:00
case <-nextMsgTicker.C:
msg, err := sub.Next(node.ctx)
if err != nil {
2021-04-04 17:05:33 +00:00
subscription.mutex.Lock()
for _, subscription := range node.subscriptions[t] {
subscription.Unsubscribe()
2021-04-04 17:05:33 +00:00
}
subscription.mutex.Unlock()
2021-03-18 16:40:47 +00:00
return
}
2021-03-15 16:07:23 +00:00
stats.Record(ctx, metrics.Messages.M(1))
2021-04-22 00:09:37 +00:00
wakuMessage := &pb.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 := protocol.NewEnvelope(wakuMessage, string(t))
node.bcaster.Submit(envelope)
}
2021-03-15 16:07:23 +00:00
}
}(t)
2021-03-22 16:45:13 +00:00
return subscription, nil
2021-03-15 16:07:23 +00:00
}
2021-06-10 12:59:51 +00:00
// Wrapper around WakuFilter.Subscribe
// that adds a Filter object to node.filters
2021-06-28 14:14:28 +00:00
func (node *WakuNode) SubscribeFilter(ctx context.Context, request pb.FilterRequest, ch filter.ContentFilterChan) error {
2021-06-10 12:59:51 +00:00
// Registers for messages that match a specific filter. Triggers the handler whenever a message is received.
// ContentFilterChan takes MessagePush structs
// Status: Implemented.
// Sanity check for well-formed subscribe FilterRequest
//doAssert(request.subscribe, "invalid subscribe request")
2021-06-14 12:54:50 +00:00
log.Info("SubscribeFilter, request: ", request)
2021-06-10 12:59:51 +00:00
var id string
2021-06-28 14:14:28 +00:00
var err error
2021-06-10 12:59:51 +00:00
2021-06-28 14:14:28 +00:00
if node.filter == nil {
return errors.New("WakuFilter is not set")
}
2021-06-10 12:59:51 +00:00
2021-06-28 14:14:28 +00:00
id, err = node.filter.Subscribe(ctx, request)
if id == "" || err != nil {
// Failed to subscribe
log.Error("remote subscription to filter failed", request)
//waku_node_errors.inc(labelValues = ["subscribe_filter_failure"])
return err
2021-06-10 12:59:51 +00:00
}
// Register handler for filter, whether remote subscription succeeded or not
2021-06-28 14:14:28 +00:00
node.filters[id] = filter.Filter{
Topic: request.Topic,
ContentFilters: request.ContentFilters,
Chan: ch,
}
return nil
2021-06-14 12:54:50 +00:00
}
func (node *WakuNode) UnsubscribeFilter(ctx context.Context, request pb.FilterRequest) {
log.Info("UnsubscribeFilter, request: ", request)
// Send message to full node in order to unsubscribe
node.filter.Unsubscribe(ctx, request)
// Remove local filter
var idsToRemove []string
for id, f := range node.filters {
// Iterate filter entries to remove matching content topics
// make sure we delete the content filter
// if no more topics are left
for _, cfToDelete := range request.ContentFilters {
for i, cf := range f.ContentFilters {
if cf.ContentTopic == cfToDelete.ContentTopic {
l := len(f.ContentFilters) - 1
f.ContentFilters[l], f.ContentFilters[i] = f.ContentFilters[i], f.ContentFilters[l]
f.ContentFilters = f.ContentFilters[:l]
break
}
}
if len(f.ContentFilters) == 0 {
idsToRemove = append(idsToRemove, id)
}
}
}
for _, rId := range idsToRemove {
for id := range node.filters {
2021-06-14 12:54:50 +00:00
if id == rId {
delete(node.filters, id)
break
}
}
}
2021-06-10 12:59:51 +00:00
}
func (node *WakuNode) Publish(ctx context.Context, message *pb.WakuMessage, topic *relay.Topic) ([]byte, error) {
if node.relay == nil {
return nil, errors.New("WakuRelay hasn't been set")
2021-03-15 16:07:23 +00:00
}
if message == nil {
return nil, errors.New("message can't be null")
}
2021-06-28 14:14:28 +00:00
if node.lightPush != nil {
return node.LightPush(ctx, message, topic)
}
hash, err := node.relay.Publish(ctx, message, topic)
if err != nil {
return nil, err
}
return hash, nil
}
func (node *WakuNode) LightPush(ctx context.Context, message *pb.WakuMessage, topic *relay.Topic, opts ...lightpush.LightPushOption) ([]byte, error) {
if node.lightPush == nil {
return nil, errors.New("WakuLightPush hasn't been set")
2021-03-15 16:07:23 +00:00
}
if message == nil {
2021-04-11 23:43:59 +00:00
return nil, errors.New("message can't be null")
2021-03-15 16:07:23 +00:00
}
req := new(pb.PushRequest)
req.Message = message
req.PubsubTopic = string(relay.GetTopic(topic))
2021-03-22 16:45:13 +00:00
response, err := node.lightPush.Request(ctx, req, opts...)
2021-03-15 16:07:23 +00:00
if err != nil {
2021-04-11 23:43:59 +00:00
return nil, err
2021-03-15 16:07:23 +00:00
}
if response.IsSuccess {
hash, _ := message.Hash()
return hash, nil
} else {
return nil, errors.New(response.Info)
2021-03-15 16:07:23 +00:00
}
2021-03-11 20:27:12 +00:00
}
func (w *WakuNode) DialPeerWithMultiAddress(ctx context.Context, address ma.Multiaddr) error {
info, err := peer.AddrInfoFromP2pAddr(address)
if err != nil {
return err
}
return w.connect(ctx, *info)
}
func (w *WakuNode) DialPeer(ctx context.Context, address string) error {
p, err := ma.NewMultiaddr(address)
if err != nil {
return err
}
info, err := peer.AddrInfoFromP2pAddr(p)
if err != nil {
return err
}
return w.connect(ctx, *info)
}
func (w *WakuNode) connect(ctx context.Context, info peer.AddrInfo) error {
err := w.host.Connect(ctx, info)
if err != nil {
return err
}
w.processHostEvent(event.EvtPeerConnectednessChanged{
Peer: info.ID,
Connectedness: network.Connected,
})
return nil
}
func (w *WakuNode) DialPeerByID(ctx context.Context, peerID peer.ID) error {
2021-08-31 19:17:56 +00:00
info := w.host.Peerstore().PeerInfo(peerID)
return w.connect(ctx, info)
2021-08-31 19:17:56 +00:00
}
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 {
err := w.host.Network().ClosePeer(id)
if err != nil {
return err
}
w.processHostEvent(event.EvtPeerConnectednessChanged{
Peer: id,
Connectedness: network.NotConnected,
})
2021-09-06 13:34:58 +00:00
return nil
}
func (w *WakuNode) PeerCount() int {
w.peersMutex.Lock()
defer w.peersMutex.Unlock()
return len(w.peers)
}
func (w *WakuNode) Peers() PeerStats {
w.peersMutex.Lock()
defer w.peersMutex.Unlock()
p := make(PeerStats)
for k, v := range w.peers {
p[k] = v
}
return p
}
func (w *WakuNode) startKeepAlive(t time.Duration) {
2021-09-11 11:36:54 +00:00
log.Info("Setting up ping protocol with duration of ", t)
w.ping = ping.NewPingService(w.host)
ticker := time.NewTicker(t)
go func() {
2021-08-10 14:23:49 +00:00
2021-09-11 11:36:54 +00:00
// This map contains peers that we're
// waiting for the ping response from
2021-08-10 14:23:49 +00:00
peerMap := make(map[peer.ID]<-chan ping.Result)
var mu sync.Mutex
for {
select {
case <-ticker.C:
2021-09-18 14:19:19 +00:00
for _, p := range w.host.Peerstore().Peers() {
if p == w.host.ID() {
log.Info("###PING skip ", p)
continue
}
2021-08-10 14:23:49 +00:00
mu.Lock()
2021-09-11 11:36:54 +00:00
_, ok := peerMap[p]
2021-08-10 14:23:49 +00:00
mu.Unlock()
2021-09-11 11:36:54 +00:00
var s = p.Pretty()
s = s[len(s)-4:]
2021-08-10 14:23:49 +00:00
if !ok {
2021-09-11 11:36:54 +00:00
log.Info("###PING ", s)
result := w.ping.Ping(w.ctx, p)
2021-08-10 14:23:49 +00:00
mu.Lock()
2021-09-11 11:36:54 +00:00
peerMap[p] = result
2021-08-10 14:23:49 +00:00
mu.Unlock()
go func(peerID peer.ID) {
2021-08-10 14:23:49 +00:00
peerFound := false
w.peersMutex.Lock()
for p := range w.peers {
if p == peerID {
2021-08-10 14:23:49 +00:00
peerFound = true
break
}
}
defer w.peersMutex.Unlock()
log.Debug("###PING before fetching result")
2021-08-10 14:23:49 +00:00
pingTicker := time.NewTicker(time.Duration(1) * time.Second)
isError := false
select {
case resVal := <-result:
isError = resVal.Error != nil
case <-pingTicker.C:
isError = true
}
pingTicker.Stop()
if !peerFound && !isError {
//EventBus Emitter doesn't seem to work when there's no connection
w.pingEventsChan <- event.EvtPeerConnectednessChanged{
Peer: peerID,
2021-08-10 14:23:49 +00:00
Connectedness: network.Connected,
}
peerConns := w.host.Network().ConnsToPeer(peerID)
2021-09-11 11:36:54 +00:00
if len(peerConns) > 0 {
// log.Info("###PING " + s + " IdentifyWait")
// logwriter.Write([]byte("###PING " + s + " IdentifyWait"))
//w.idService.IdentifyWait(peerConns[0])
} else {
go func(peerID peer.ID) {
ctx, cancel := context.WithTimeout(w.ctx, time.Duration(5)*time.Second)
defer cancel()
if err := w.DialPeerByID(ctx, peerID); err != nil {
log.Warn("could not dial peer ", peerID)
}
}(peerID)
2021-09-11 11:36:54 +00:00
}
2021-08-10 14:23:49 +00:00
} else if peerFound && isError {
w.pingEventsChan <- event.EvtPeerConnectednessChanged{
Peer: peerID,
2021-08-10 14:23:49 +00:00
Connectedness: network.NotConnected,
}
}
mu.Lock()
delete(peerMap, peerID)
2021-08-10 14:23:49 +00:00
mu.Unlock()
2021-09-11 11:36:54 +00:00
}(p)
2021-08-10 14:23:49 +00:00
} else {
2021-09-11 11:36:54 +00:00
log.Info("###PING " + s + " already pinged")
// logwriter.Write([]byte("###PING " + s + " already pinged"))
2021-08-10 14:23:49 +00:00
}
}
case <-w.quit:
ticker.Stop()
return
}
}
}()
2021-08-31 18:19:49 +00:00
}
func addrInfoFromMultiaddrString(address string) (*peer.AddrInfo, error) {
ma, err := ma.NewMultiaddr(address)
if err != nil {
return nil, err
}
2021-08-31 18:19:49 +00:00
return peer.AddrInfoFromP2pAddr(ma)
}