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

444 lines
10 KiB
Go
Raw Normal View History

2021-03-11 20:27:12 +00:00
package node
import (
"context"
"fmt"
2021-03-15 23:59:18 +00:00
"time"
2021-03-11 20:27:12 +00:00
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-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-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"
pubsub "github.com/libp2p/go-libp2p-pubsub"
"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"
2021-04-22 00:09:37 +00:00
2021-10-05 02:13:54 +00:00
rendezvous "github.com/status-im/go-waku-rendezvous"
2021-11-01 14:42:55 +00:00
v2 "github.com/status-im/go-waku/waku/v2"
"github.com/status-im/go-waku/waku/v2/metrics"
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"
"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"
"github.com/status-im/go-waku/waku/v2/utils"
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
type Message []byte
type WakuNode struct {
host host.Host
opts *WakuNodeParameters
relay *relay.WakuRelay
filter *filter.WakuFilter
lightPush *lightpush.WakuLightPush
2021-10-01 17:49:50 +00:00
rendezvous *rendezvous.RendezvousService
ping *ping.PingService
store *store.WakuStore
2021-11-01 14:42:55 +00:00
bcaster v2.Broadcaster
connectionNotif ConnectionNotifier
2021-06-16 10:14:22 +00:00
protocolEventSub event.Subscription
identificationEventSub event.Subscription
ctx context.Context
cancel context.CancelFunc
quit chan struct{}
2021-06-16 10:14:22 +00:00
// Channel passed to WakuNode constructor
// receiving connection status notifications
connStatusChan chan ConnStatus
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)
params.libP2POpts = DefaultLibP2POptions
for _, opt := range opts {
err := opt(params)
2021-03-11 20:27:12 +00:00
if err != nil {
2021-10-01 17:49:50 +00:00
cancel()
2021-03-11 20:27:12 +00:00
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, params.Identity())
}
2021-03-15 16:07:23 +00:00
2021-10-15 02:15:02 +00:00
if params.addressFactory != nil {
params.libP2POpts = append(params.libP2POpts, libp2p.AddrsFactory(params.addressFactory))
}
host, err := libp2p.New(ctx, params.libP2POpts...)
2021-03-11 20:27:12 +00:00
if err != nil {
2021-10-01 17:49:50 +00:00
cancel()
2021-03-11 20:27:12 +00:00
return nil, err
}
w := new(WakuNode)
2021-11-01 14:42:55 +00:00
w.bcaster = v2.NewBroadcaster(1024)
2021-03-15 16:07:23 +00:00
w.host = host
w.cancel = cancel
w.ctx = ctx
w.opts = params
w.quit = make(chan struct{})
2021-06-16 10:14:22 +00:00
if w.protocolEventSub, err = host.EventBus().Subscribe(new(event.EvtPeerProtocolsUpdated)); err != nil {
return nil, err
}
2021-06-16 10:14:22 +00:00
if w.identificationEventSub, err = host.EventBus().Subscribe(new(event.EvtPeerIdentificationCompleted)); err != nil {
return nil, err
}
2021-06-16 10:14:22 +00:00
if params.connStatusC != nil {
w.connStatusChan = params.connStatusC
2021-06-28 14:14:28 +00:00
}
2021-10-16 21:50:49 +00:00
w.connectionNotif = NewConnectionNotifier(ctx, host)
w.host.Network().Notify(w.connectionNotif)
2021-06-16 10:14:22 +00:00
go w.connectednessListener()
2021-06-28 14:14:28 +00:00
2021-10-05 02:13:54 +00:00
if w.opts.keepAliveInterval > time.Duration(0) {
w.startKeepAlive(w.opts.keepAliveInterval)
}
for _, addr := range w.ListenAddresses() {
log.Info("Listening on ", addr)
}
return w, nil
}
func (w *WakuNode) Start() error {
w.store = store.NewWakuStore(w.opts.messageProvider, w.opts.maxMessages, w.opts.maxDuration)
2021-10-05 02:13:54 +00:00
if w.opts.enableStore {
2021-06-16 10:14:22 +00:00
w.startStore()
}
2021-10-05 02:13:54 +00:00
if w.opts.enableFilter {
w.filter = filter.NewWakuFilter(w.ctx, w.host, w.opts.isFilterFullNode)
2021-06-10 12:59:51 +00:00
}
2021-10-05 02:13:54 +00:00
if w.opts.enableRendezvous {
rendezvous := rendezvous.NewRendezvousDiscovery(w.host)
w.opts.wOpts = append(w.opts.wOpts, pubsub.WithDiscovery(rendezvous, w.opts.rendezvousOpts...))
2021-10-01 17:49:50 +00:00
}
2021-11-01 14:42:55 +00:00
err := w.mountRelay(w.opts.wOpts...)
if err != nil {
return err
2021-06-16 10:14:22 +00:00
}
2021-11-01 12:38:03 +00:00
w.lightPush = lightpush.NewWakuLightPush(w.ctx, w.host, w.relay)
2021-10-05 02:13:54 +00:00
if w.opts.enableLightPush {
2021-11-01 12:38:03 +00:00
if err := w.lightPush.Start(); err != nil {
return err
}
}
2021-10-05 02:13:54 +00:00
if w.opts.enableRendezvousServer {
2021-10-01 17:49:50 +00:00
err := w.mountRendezvous()
if err != nil {
2021-10-05 02:13:54 +00:00
return err
2021-10-01 17:49:50 +00:00
}
}
2021-11-01 14:42:55 +00:00
// Subscribe store to topic
if w.opts.storeMsgs {
log.Info("Subscribing store to broadcaster")
w.bcaster.Register(w.store.MsgC)
}
if w.filter != nil {
log.Info("Subscribing filter to broadcaster")
w.bcaster.Register(w.filter.MsgC)
}
2021-10-05 02:13:54 +00:00
return nil
2021-03-11 20:27:12 +00:00
}
2021-03-22 16:45:13 +00:00
func (w *WakuNode) Stop() {
2021-03-15 16:07:23 +00:00
defer w.cancel()
2021-03-22 16:45:13 +00:00
close(w.quit)
2021-11-01 14:42:55 +00:00
w.bcaster.Close()
defer w.connectionNotif.Close()
2021-06-16 10:14:22 +00:00
defer w.protocolEventSub.Close()
defer w.identificationEventSub.Close()
if w.rendezvous != nil {
w.rendezvous.Stop()
}
if w.filter != nil {
w.filter.Stop()
2021-03-22 16:45:13 +00:00
}
2021-11-01 14:42:55 +00:00
w.relay.Stop()
2021-11-01 12:38:03 +00:00
w.lightPush.Stop()
w.store.Stop()
w.host.Close()
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) 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-11-01 12:38:03 +00:00
func (w *WakuNode) Store() *store.WakuStore {
return w.store
}
2021-06-10 12:59:51 +00:00
func (w *WakuNode) Filter() *filter.WakuFilter {
return w.filter
}
2021-11-01 12:38:03 +00:00
func (w *WakuNode) Lightpush() *lightpush.WakuLightPush {
return w.lightPush
}
func (w *WakuNode) mountRelay(opts ...pubsub.Option) error {
var err error
2021-11-01 14:42:55 +00:00
w.relay, err = relay.NewWakuRelay(w.ctx, w.host, w.bcaster, opts...)
if err != nil {
return err
}
2021-03-15 16:07:23 +00:00
2021-11-01 14:42:55 +00:00
if w.opts.enableRelay {
_, err = w.relay.Subscribe(w.ctx, nil)
if err != nil {
return err
}
2021-06-28 14:14:28 +00:00
}
2021-03-15 16:07:23 +00:00
// TODO: rlnRelay
return err
2021-03-18 16:40:47 +00:00
}
2021-10-01 17:49:50 +00:00
func (w *WakuNode) mountRendezvous() error {
w.rendezvous = rendezvous.NewRendezvousService(w.host, w.opts.rendevousStorage)
if err := w.rendezvous.Start(); err != nil {
2021-10-01 17:49:50 +00:00
return err
}
2021-10-01 17:49:50 +00:00
log.Info("Rendezvous service started")
return nil
}
2021-06-28 14:14:28 +00:00
func (w *WakuNode) startStore() {
w.store.Start(w.ctx, w.host)
if w.opts.shouldResume {
// TODO: extract this to a function and run it when you go offline
// TODO: determine if a store is listening to a topic
go func() {
for {
t := time.NewTicker(time.Second)
peerVerif:
for {
select {
case <-w.quit:
return
case <-t.C:
_, err := utils.SelectPeer(w.host, string(store.StoreID_v20beta3))
if err == nil {
break peerVerif
}
}
}
ctxWithTimeout, ctxCancel := context.WithTimeout(w.ctx, 20*time.Second)
defer ctxCancel()
2021-11-01 12:38:03 +00:00
if _, err := w.store.Resume(ctxWithTimeout, string(relay.DefaultWakuTopic), nil); err != nil {
log.Info("Retrying in 10s...")
time.Sleep(10 * time.Second)
} else {
break
}
}
}()
2021-08-13 11:56:09 +00:00
}
}
func (w *WakuNode) addPeer(info *peer.AddrInfo, protocolID p2pproto.ID) error {
log.Info(fmt.Sprintf("Adding peer %s to peerstore", info.ID.Pretty()))
w.host.Peerstore().AddAddrs(info.ID, info.Addrs, peerstore.PermanentAddrTTL)
2021-10-16 21:50:49 +00:00
err := w.host.Peerstore().AddProtocols(info.ID, string(protocolID))
if err != nil {
return err
}
2021-06-10 12:59:51 +00:00
2021-10-16 21:50:49 +00:00
return nil
2021-08-31 18:19:49 +00:00
}
2021-06-10 12:59:51 +00:00
func (w *WakuNode) AddPeer(address ma.Multiaddr, protocolID p2pproto.ID) (*peer.ID, error) {
info, err := peer.AddrInfoFromP2pAddr(address)
if err != nil {
return nil, err
}
2021-06-10 12:59:51 +00:00
return &info.ID, w.addPeer(info, protocolID)
2021-06-28 14:14:28 +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
}
2021-10-16 21:50:49 +00:00
stats.Record(ctx, metrics.Dials.M(1))
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
}
2021-09-06 13:34:58 +00:00
return nil
}
func (w *WakuNode) PeerCount() int {
return len(w.host.Network().Peers())
}
func (w *WakuNode) Peers() PeerStats {
p := make(PeerStats)
for _, peerID := range w.host.Network().Peers() {
protocols, err := w.host.Peerstore().GetProtocols(peerID)
if err != nil {
continue
}
p[peerID] = protocols
}
return p
}
2021-10-29 15:24:31 +00:00
// startKeepAlive creates a go routine that periodically pings connected peers.
// This is necessary because TCP connections are automatically closed due to inactivity,
// and doing a ping will avoid this (with a small bandwidth cost)
func (w *WakuNode) startKeepAlive(t time.Duration) {
log.Info("Setting up ping protocol with duration of ", t)
w.ping = ping.NewPingService(w.host)
ticker := time.NewTicker(t)
2021-10-16 23:09:04 +00:00
go func() {
for {
select {
case <-ticker.C:
// Compared to Network's peers collection,
// Peerstore contains all peers ever connected to,
// thus if a host goes down and back again,
// pinging a peer will trigger identification process,
// which is not possible when iterating
// through Network's peer collection, as it will be empty
for _, p := range w.host.Peerstore().Peers() {
2021-10-29 15:24:31 +00:00
go pingPeer(w.ctx, w.ping, p)
}
case <-w.quit:
ticker.Stop()
return
}
}
}()
2021-08-31 18:19:49 +00:00
}
2021-10-29 15:24:31 +00:00
func pingPeer(ctx context.Context, pingService *ping.PingService, peer peer.ID) {
ctx, cancel := context.WithTimeout(ctx, 3*time.Second)
defer cancel()
log.Debug("Pinging ", peer)
pr := pingService.Ping(ctx, peer)
select {
case res := <-pr:
if res.Error != nil {
log.Error(fmt.Sprintf("Could not ping %s: %s", peer, res.Error.Error()))
}
case <-ctx.Done():
log.Error(fmt.Sprintf("Could not ping %s: %s", peer, ctx.Err()))
}
}