bump go-waku (#2400)
This commit is contained in:
Richard Ramos
GitHub
parent
75aebd94ab
commit
a8d661ee71
4
go.mod
4
go.mod
|
|
@ -47,9 +47,9 @@ require (
|
|||
github.com/russolsen/same v0.0.0-20160222130632-f089df61f51d // indirect
|
||||
github.com/russolsen/transit v0.0.0-20180705123435-0794b4c4505a
|
||||
github.com/status-im/doubleratchet v3.0.0+incompatible
|
||||
github.com/status-im/go-waku v0.0.0-20211005021624-79e21dbb993c
|
||||
github.com/status-im/go-waku v0.0.0-20211006192541-761ae88bbdea
|
||||
github.com/status-im/go-waku-rendezvous v0.0.0-20211005020656-b53661c58574
|
||||
github.com/status-im/go-wakurelay-pubsub v0.4.3-0.20210729162817-adc68830282a
|
||||
github.com/status-im/go-wakurelay-pubsub v0.4.3-0.20211006151908-91851dda6375
|
||||
github.com/status-im/markdown v0.0.0-20210405121740-32e5a5055fb6
|
||||
github.com/status-im/migrate/v4 v4.6.2-status.2
|
||||
github.com/status-im/rendezvous v1.3.1-0.20210824184947-7c79c858170c
|
||||
|
|
|
|||
8
go.sum
8
go.sum
|
|
@ -1076,12 +1076,12 @@ github.com/status-im/go-ethereum v1.10.4-status.3 h1:RF618iSCvqJtXu3ZSg7XNg6MJaS
|
|||
github.com/status-im/go-ethereum v1.10.4-status.3/go.mod h1:GvIhpdCOgMHI6i5xVPEZOrv/qSMeOFHbZh77AoyZUoE=
|
||||
github.com/status-im/go-multiaddr-ethv4 v1.2.0 h1:OT84UsUzTCwguqCpJqkrCMiL4VZ1SvUtH9a5MsZupBk=
|
||||
github.com/status-im/go-multiaddr-ethv4 v1.2.0/go.mod h1:2VQ3C+9zEurcceasz12gPAtmEzCeyLUGPeKLSXYQKHo=
|
||||
github.com/status-im/go-waku v0.0.0-20211005021624-79e21dbb993c h1:NBAVMzp24RPq6cMkAgZfObU26naB6t5kXqCNe1wump4=
|
||||
github.com/status-im/go-waku v0.0.0-20211005021624-79e21dbb993c/go.mod h1:5ZtKtIFFYFWX44SHrRX1Kn2uMYfy/qOmkU7Hd3T4qVU=
|
||||
github.com/status-im/go-waku v0.0.0-20211006192541-761ae88bbdea h1:nzHN/fknFXRCzgjHTYqWPJZ5BDKjKRCmY4wtE7fuaMI=
|
||||
github.com/status-im/go-waku v0.0.0-20211006192541-761ae88bbdea/go.mod h1:SsSaPcFPFHeld/8SpJiOHM5PGXqR+V2fzsrOLD5Zkl8=
|
||||
github.com/status-im/go-waku-rendezvous v0.0.0-20211005020656-b53661c58574 h1:v2gpjWKyI+vZZugjjhPDqIhg6uNrGLusHh3ilvbv8/Y=
|
||||
github.com/status-im/go-waku-rendezvous v0.0.0-20211005020656-b53661c58574/go.mod h1:Fa1uJjMz9MpfZc2tC5xdN9q90xg1VphSnevxWiBbFO0=
|
||||
github.com/status-im/go-wakurelay-pubsub v0.4.3-0.20210729162817-adc68830282a h1:eCna/q/PuZVqtmOMBqytw9yzZwMNKpao4au0OJDvesI=
|
||||
github.com/status-im/go-wakurelay-pubsub v0.4.3-0.20210729162817-adc68830282a/go.mod h1:LSCVYR7mnBBsxVJghrGpQ3yJAAATEe6XeQQqGCZhwrE=
|
||||
github.com/status-im/go-wakurelay-pubsub v0.4.3-0.20211006151908-91851dda6375 h1:4lYWWH3SPnVrCmDmqUpQlXy65uFUZZCjH6gS9899iJc=
|
||||
github.com/status-im/go-wakurelay-pubsub v0.4.3-0.20211006151908-91851dda6375/go.mod h1:LSCVYR7mnBBsxVJghrGpQ3yJAAATEe6XeQQqGCZhwrE=
|
||||
github.com/status-im/keycard-go v0.0.0-20190316090335-8537d3370df4/go.mod h1:RZLeN1LMWmRsyYjvAu+I6Dm9QmlDaIIt+Y+4Kd7Tp+Q=
|
||||
github.com/status-im/keycard-go v0.0.0-20200402102358-957c09536969 h1:Oo2KZNP70KE0+IUJSidPj/BFS/RXNHmKIJOdckzml2E=
|
||||
github.com/status-im/keycard-go v0.0.0-20200402102358-957c09536969/go.mod h1:RZLeN1LMWmRsyYjvAu+I6Dm9QmlDaIIt+Y+4Kd7Tp+Q=
|
||||
|
|
|
|||
129
vendor/github.com/status-im/go-waku/waku/v2/node/connectedness.go
generated
vendored
Normal file
129
vendor/github.com/status-im/go-waku/waku/v2/node/connectedness.go
generated
vendored
Normal file
|
|
@ -0,0 +1,129 @@
|
|||
package node
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
|
||||
"github.com/libp2p/go-libp2p-core/host"
|
||||
"github.com/libp2p/go-libp2p-core/network"
|
||||
"github.com/libp2p/go-libp2p-core/peer"
|
||||
ma "github.com/multiformats/go-multiaddr"
|
||||
"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/store"
|
||||
wakurelay "github.com/status-im/go-wakurelay-pubsub"
|
||||
)
|
||||
|
||||
// A map of peer IDs to supported protocols
|
||||
type PeerStats map[peer.ID][]string
|
||||
|
||||
type ConnStatus struct {
|
||||
IsOnline bool
|
||||
HasHistory bool
|
||||
Peers PeerStats
|
||||
}
|
||||
|
||||
type ConnectionNotifier struct {
|
||||
h host.Host
|
||||
DisconnectChan chan peer.ID
|
||||
quit chan struct{}
|
||||
}
|
||||
|
||||
func NewConnectionNotifier(h host.Host) ConnectionNotifier {
|
||||
return ConnectionNotifier{
|
||||
h: h,
|
||||
DisconnectChan: make(chan peer.ID, 100),
|
||||
quit: make(chan struct{}),
|
||||
}
|
||||
}
|
||||
|
||||
func (c ConnectionNotifier) Listen(n network.Network, m ma.Multiaddr) {
|
||||
// called when network starts listening on an addr
|
||||
}
|
||||
|
||||
func (c ConnectionNotifier) ListenClose(n network.Network, m ma.Multiaddr) {
|
||||
// called when network stops listening on an addr
|
||||
}
|
||||
|
||||
func (c ConnectionNotifier) Connected(n network.Network, cc network.Conn) {
|
||||
// called when a connection opened
|
||||
log.Info(fmt.Sprintf("Peer %s connected", cc.RemotePeer()))
|
||||
}
|
||||
|
||||
func (c ConnectionNotifier) Disconnected(n network.Network, cc network.Conn) {
|
||||
// called when a connection closed
|
||||
log.Info(fmt.Sprintf("Peer %s disconnected", cc.RemotePeer()))
|
||||
c.DisconnectChan <- cc.RemotePeer()
|
||||
}
|
||||
|
||||
func (c ConnectionNotifier) OpenedStream(n network.Network, s network.Stream) {
|
||||
// called when a stream opened
|
||||
}
|
||||
|
||||
func (c ConnectionNotifier) ClosedStream(n network.Network, s network.Stream) {
|
||||
// called when a stream closed
|
||||
}
|
||||
|
||||
func (c ConnectionNotifier) Close() {
|
||||
close(c.quit)
|
||||
}
|
||||
|
||||
func (w *WakuNode) sendConnStatus() {
|
||||
isOnline, hasHistory := w.Status()
|
||||
if w.connStatusChan != nil {
|
||||
connStatus := ConnStatus{IsOnline: isOnline, HasHistory: hasHistory, Peers: w.Peers()}
|
||||
w.connStatusChan <- connStatus
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
func (w *WakuNode) connectednessListener() {
|
||||
for {
|
||||
select {
|
||||
case <-w.quit:
|
||||
return
|
||||
case <-w.protocolEventSub.Out():
|
||||
case <-w.identificationEventSub.Out():
|
||||
case <-w.connectionNotif.DisconnectChan:
|
||||
}
|
||||
w.sendConnStatus()
|
||||
}
|
||||
}
|
||||
|
||||
func (w *WakuNode) Status() (isOnline bool, hasHistory bool) {
|
||||
hasRelay := false
|
||||
hasLightPush := false
|
||||
hasStore := false
|
||||
hasFilter := false
|
||||
|
||||
for _, peer := range w.host.Network().Peers() {
|
||||
protocols, err := w.host.Peerstore().GetProtocols(peer)
|
||||
if err != nil {
|
||||
log.Warn(fmt.Errorf("could not read peer %s protocols", peer))
|
||||
}
|
||||
|
||||
for _, protocol := range protocols {
|
||||
if !hasRelay && protocol == string(wakurelay.WakuRelayID_v200) {
|
||||
hasRelay = true
|
||||
}
|
||||
if !hasLightPush && protocol == string(lightpush.LightPushID_v20beta1) {
|
||||
hasLightPush = true
|
||||
}
|
||||
if !hasStore && protocol == string(store.StoreID_v20beta3) {
|
||||
hasStore = true
|
||||
}
|
||||
if !hasFilter && protocol == string(filter.FilterID_v20beta1) {
|
||||
hasFilter = true
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if hasStore {
|
||||
hasHistory = true
|
||||
}
|
||||
|
||||
if hasRelay || hasLightPush && (hasStore || hasFilter) {
|
||||
isOnline = true
|
||||
}
|
||||
|
||||
return
|
||||
}
|
||||
|
|
@ -4,20 +4,15 @@ import (
|
|||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
|
||||
//"log/syslog"
|
||||
//"strconv"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
proto "github.com/golang/protobuf/proto"
|
||||
logging "github.com/ipfs/go-log"
|
||||
"github.com/libp2p/go-libp2p"
|
||||
"github.com/libp2p/go-libp2p/p2p/protocol/identify"
|
||||
|
||||
"github.com/libp2p/go-libp2p-core/event"
|
||||
"github.com/libp2p/go-libp2p-core/host"
|
||||
"github.com/libp2p/go-libp2p-core/network"
|
||||
"github.com/libp2p/go-libp2p-core/peer"
|
||||
"github.com/libp2p/go-libp2p-core/peerstore"
|
||||
p2pproto "github.com/libp2p/go-libp2p-core/protocol"
|
||||
|
|
@ -39,31 +34,17 @@ import (
|
|||
|
||||
var log = logging.Logger("wakunode")
|
||||
|
||||
//var logwriter, _ = syslog.New(syslog.LOG_ERR|syslog.LOG_LOCAL0, "WAKU")
|
||||
|
||||
type Message []byte
|
||||
|
||||
// A map of peer IDs to supported protocols
|
||||
type PeerStats map[peer.ID][]string
|
||||
|
||||
type ConnStatus struct {
|
||||
IsOnline bool
|
||||
HasHistory bool
|
||||
Peers PeerStats
|
||||
}
|
||||
|
||||
type WakuNode struct {
|
||||
host host.Host
|
||||
idService *identify.IDService
|
||||
opts *WakuNodeParameters
|
||||
|
||||
relay *relay.WakuRelay
|
||||
filter *filter.WakuFilter
|
||||
lightPush *lightpush.WakuLightPush
|
||||
host host.Host
|
||||
opts *WakuNodeParameters
|
||||
|
||||
relay *relay.WakuRelay
|
||||
filter *filter.WakuFilter
|
||||
lightPush *lightpush.WakuLightPush
|
||||
rendezvous *rendezvous.RendezvousService
|
||||
|
||||
ping *ping.PingService
|
||||
ping *ping.PingService
|
||||
|
||||
subscriptions map[relay.Topic][]*Subscription
|
||||
subscriptionsMutex sync.Mutex
|
||||
|
|
@ -72,7 +53,7 @@ type WakuNode struct {
|
|||
|
||||
filters filter.Filters
|
||||
|
||||
connectednessEventSub event.Subscription
|
||||
connectionNotif ConnectionNotifier
|
||||
protocolEventSub event.Subscription
|
||||
identificationEventSub event.Subscription
|
||||
|
||||
|
|
@ -80,120 +61,9 @@ type WakuNode struct {
|
|||
cancel context.CancelFunc
|
||||
quit chan struct{}
|
||||
|
||||
// Map of peers and their supported protocols
|
||||
peers PeerStats
|
||||
peersMutex sync.Mutex
|
||||
|
||||
// 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
|
||||
pingEventsChan chan interface{}
|
||||
}
|
||||
|
||||
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()
|
||||
|
||||
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)
|
||||
w.peers[ev.Peer] = peerProtocols
|
||||
} else {
|
||||
log.Debug("### Peer already exists")
|
||||
}
|
||||
} else if ev.Connectedness == network.NotConnected {
|
||||
log.Debug("Peer down: ", ev.Peer)
|
||||
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)
|
||||
}
|
||||
|
||||
}
|
||||
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()
|
||||
|
||||
_, 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)
|
||||
w.peers[ev.Peer] = peerProtocols
|
||||
}
|
||||
|
||||
}
|
||||
func (w *WakuNode) handlePeerIdentificationCompleted(ev event.EvtPeerIdentificationCompleted) {
|
||||
log.Debug("### EvtPeerIdentificationCompleted ", w.Host().ID(), " to ", ev.Peer)
|
||||
|
||||
w.peersMutex.Lock()
|
||||
defer w.peersMutex.Unlock()
|
||||
|
||||
peerProtocols, _ := w.host.Peerstore().GetProtocols(ev.Peer)
|
||||
log.Debug("identified protocols found for peer: ", ev.Peer, ", protocols: ", peerProtocols)
|
||||
w.peers[ev.Peer] = peerProtocols
|
||||
}
|
||||
func (w *WakuNode) processHostEvent(e interface{}) {
|
||||
if e == nil {
|
||||
log.Debug("processHostEvent nil event")
|
||||
return
|
||||
}
|
||||
isOnline := w.IsOnline()
|
||||
hasHistory := w.HasHistory()
|
||||
switch e := e.(type) {
|
||||
case event.EvtPeerConnectednessChanged:
|
||||
w.handleConnectednessChanged(e)
|
||||
case event.EvtPeerProtocolsUpdated:
|
||||
w.handleProtocolsUpdated(e)
|
||||
case event.EvtPeerIdentificationCompleted:
|
||||
w.handlePeerIdentificationCompleted(e)
|
||||
}
|
||||
|
||||
log.Debug("###processHostEvent before isOnline()")
|
||||
newIsOnline := w.IsOnline()
|
||||
log.Debug("###processHostEvent before hasHistory()")
|
||||
newHasHistory := w.HasHistory()
|
||||
log.Debug("###ConnStatus isOnline: ", isOnline, "/", newIsOnline, " hasHistory: ",
|
||||
hasHistory, "/", newHasHistory)
|
||||
if w.connStatusChan != nil {
|
||||
connStatus := ConnStatus{IsOnline: newIsOnline, HasHistory: newHasHistory, Peers: w.Peers()}
|
||||
log.Debug("New ConnStatus: ", connStatus)
|
||||
w.connStatusChan <- connStatus
|
||||
}
|
||||
|
||||
}
|
||||
func (w *WakuNode) connectednessListener() {
|
||||
for {
|
||||
var e interface{}
|
||||
log.Debug("connectednessListener before select")
|
||||
select {
|
||||
case e = <-w.connectednessEventSub.Out():
|
||||
log.Debug("connectednessListener connectednessEvent")
|
||||
case e = <-w.protocolEventSub.Out():
|
||||
log.Debug("connectednessListener protocolEvent")
|
||||
case e = <-w.identificationEventSub.Out():
|
||||
log.Debug("connectednessListener identificationEvent")
|
||||
case e = <-w.pingEventsChan:
|
||||
log.Debug("connectednessListener pingEvent")
|
||||
}
|
||||
log.Debug("connectednessListener after select")
|
||||
|
||||
w.processHostEvent(e)
|
||||
log.Debug("connectednessListener after processHostEvent")
|
||||
}
|
||||
}
|
||||
|
||||
func New(ctx context.Context, opts ...WakuNodeOption) (*WakuNode, error) {
|
||||
|
|
@ -228,28 +98,26 @@ func New(ctx context.Context, opts ...WakuNodeOption) (*WakuNode, error) {
|
|||
w := new(WakuNode)
|
||||
w.bcaster = NewBroadcaster(1024)
|
||||
w.host = host
|
||||
w.idService = identify.NewIDService(host)
|
||||
w.cancel = cancel
|
||||
w.ctx = ctx
|
||||
w.subscriptions = make(map[relay.Topic][]*Subscription)
|
||||
w.opts = params
|
||||
w.quit = make(chan struct{})
|
||||
w.peers = make(PeerStats)
|
||||
|
||||
// Subscribe to Connectedness events
|
||||
connectednessEventSub, _ := host.EventBus().Subscribe(new(event.EvtPeerConnectednessChanged))
|
||||
w.connectednessEventSub = connectednessEventSub
|
||||
if w.protocolEventSub, err = host.EventBus().Subscribe(new(event.EvtPeerProtocolsUpdated)); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
protocolEventSub, _ := host.EventBus().Subscribe(new(event.EvtPeerProtocolsUpdated))
|
||||
w.protocolEventSub = protocolEventSub
|
||||
|
||||
identificationEventSub, _ := host.EventBus().Subscribe(new(event.EvtPeerIdentificationCompleted))
|
||||
w.identificationEventSub = identificationEventSub
|
||||
if w.identificationEventSub, err = host.EventBus().Subscribe(new(event.EvtPeerIdentificationCompleted)); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
if params.connStatusChan != nil {
|
||||
w.connStatusChan = params.connStatusChan
|
||||
}
|
||||
w.pingEventsChan = make(chan interface{})
|
||||
|
||||
w.connectionNotif = NewConnectionNotifier(host)
|
||||
w.host.Network().Notify(w.connectionNotif)
|
||||
go w.connectednessListener()
|
||||
|
||||
if w.opts.keepAliveInterval > time.Duration(0) {
|
||||
|
|
@ -307,7 +175,7 @@ func (w *WakuNode) Stop() {
|
|||
|
||||
close(w.quit)
|
||||
|
||||
defer w.connectednessEventSub.Close()
|
||||
defer w.connectionNotif.Close()
|
||||
defer w.protocolEventSub.Close()
|
||||
defer w.identificationEventSub.Close()
|
||||
|
||||
|
|
@ -334,51 +202,6 @@ func (w *WakuNode) ID() string {
|
|||
return w.host.ID().Pretty()
|
||||
}
|
||||
|
||||
func (w *WakuNode) IsOnline() bool {
|
||||
w.peersMutex.Lock()
|
||||
defer w.peersMutex.Unlock()
|
||||
|
||||
hasRelay := false
|
||||
hasLightPush := false
|
||||
hasStore := false
|
||||
hasFilter := false
|
||||
for _, v := range w.peers {
|
||||
for _, protocol := range v {
|
||||
if !hasRelay && protocol == string(wakurelay.WakuRelayID_v200) {
|
||||
hasRelay = true
|
||||
}
|
||||
if !hasLightPush && protocol == string(lightpush.LightPushID_v20beta1) {
|
||||
hasLightPush = true
|
||||
}
|
||||
if !hasStore && protocol == string(store.StoreID_v20beta3) {
|
||||
hasStore = true
|
||||
}
|
||||
if !hasFilter && protocol == string(filter.FilterID_v20beta1) {
|
||||
hasFilter = true
|
||||
}
|
||||
if hasRelay || hasLightPush && (hasStore || hasFilter) {
|
||||
return true
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return false
|
||||
}
|
||||
|
||||
func (w *WakuNode) HasHistory() bool {
|
||||
w.peersMutex.Lock()
|
||||
defer w.peersMutex.Unlock()
|
||||
|
||||
for _, v := range w.peers {
|
||||
for _, protocol := range v {
|
||||
if protocol == string(store.StoreID_v20beta3) {
|
||||
return true
|
||||
}
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
func (w *WakuNode) ListenAddresses() []ma.Multiaddr {
|
||||
hostInfo, _ := ma.NewMultiaddr(fmt.Sprintf("/p2p/%s", w.host.ID().Pretty()))
|
||||
var result []ma.Multiaddr
|
||||
|
|
@ -418,12 +241,10 @@ func (w *WakuNode) mountFilter() error {
|
|||
w.filters.Notify(message, requestId) // Trigger filter handlers on a light node
|
||||
}
|
||||
}
|
||||
peerChan := make(chan *event.EvtPeerConnectednessChanged)
|
||||
w.filter = filter.NewWakuFilter(w.ctx, w.host, filterHandler, peerChan)
|
||||
w.peerListeners = append(w.peerListeners, peerChan)
|
||||
|
||||
w.filter = filter.NewWakuFilter(w.ctx, w.host, filterHandler)
|
||||
|
||||
return nil
|
||||
|
||||
}
|
||||
|
||||
func (w *WakuNode) mountLightPush() {
|
||||
|
|
@ -442,9 +263,7 @@ func (w *WakuNode) mountRendezvous() error {
|
|||
}
|
||||
|
||||
func (w *WakuNode) startStore() {
|
||||
peerChan := make(chan *event.EvtPeerConnectednessChanged)
|
||||
w.opts.store.Start(w.ctx, w.host, peerChan)
|
||||
w.peerListeners = append(w.peerListeners, peerChan)
|
||||
w.opts.store.Start(w.ctx, w.host)
|
||||
|
||||
if w.opts.shouldResume {
|
||||
if _, err := w.opts.store.Resume(string(relay.GetTopic(nil)), nil); err != nil {
|
||||
|
|
@ -740,12 +559,6 @@ func (w *WakuNode) connect(ctx context.Context, info peer.AddrInfo) error {
|
|||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
w.processHostEvent(event.EvtPeerConnectednessChanged{
|
||||
Peer: info.ID,
|
||||
Connectedness: network.Connected,
|
||||
})
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
|
|
@ -774,120 +587,37 @@ func (w *WakuNode) ClosePeerById(id peer.ID) error {
|
|||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
w.processHostEvent(event.EvtPeerConnectednessChanged{
|
||||
Peer: id,
|
||||
Connectedness: network.NotConnected,
|
||||
})
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (w *WakuNode) PeerCount() int {
|
||||
w.peersMutex.Lock()
|
||||
defer w.peersMutex.Unlock()
|
||||
return len(w.peers)
|
||||
return len(w.host.Network().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
|
||||
for _, peerID := range w.host.Network().Peers() {
|
||||
protocols, err := w.host.Peerstore().GetProtocols(peerID)
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
p[peerID] = protocols
|
||||
}
|
||||
|
||||
return p
|
||||
}
|
||||
|
||||
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)
|
||||
go func() {
|
||||
// This map contains peers that we're
|
||||
// waiting for the ping response from
|
||||
peerMap := make(map[peer.ID]<-chan ping.Result)
|
||||
var mu sync.Mutex
|
||||
for {
|
||||
select {
|
||||
case <-ticker.C:
|
||||
for _, p := range w.host.Peerstore().Peers() {
|
||||
if p == w.host.ID() {
|
||||
log.Info("###PING skip ", p)
|
||||
continue
|
||||
}
|
||||
mu.Lock()
|
||||
_, ok := peerMap[p]
|
||||
mu.Unlock()
|
||||
|
||||
var s = p.Pretty()
|
||||
s = s[len(s)-4:]
|
||||
if !ok {
|
||||
log.Info("###PING ", s)
|
||||
result := w.ping.Ping(w.ctx, p)
|
||||
mu.Lock()
|
||||
peerMap[p] = result
|
||||
mu.Unlock()
|
||||
|
||||
go func(peerID peer.ID) {
|
||||
peerFound := false
|
||||
w.peersMutex.Lock()
|
||||
for p := range w.peers {
|
||||
if p == peerID {
|
||||
peerFound = true
|
||||
break
|
||||
}
|
||||
}
|
||||
defer w.peersMutex.Unlock()
|
||||
log.Debug("###PING before fetching result")
|
||||
|
||||
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,
|
||||
Connectedness: network.Connected,
|
||||
}
|
||||
peerConns := w.host.Network().ConnsToPeer(peerID)
|
||||
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)
|
||||
}
|
||||
} else if peerFound && isError {
|
||||
w.pingEventsChan <- event.EvtPeerConnectednessChanged{
|
||||
Peer: peerID,
|
||||
Connectedness: network.NotConnected,
|
||||
}
|
||||
}
|
||||
|
||||
mu.Lock()
|
||||
delete(peerMap, peerID)
|
||||
mu.Unlock()
|
||||
}(p)
|
||||
} else {
|
||||
log.Info("###PING " + s + " already pinged")
|
||||
// logwriter.Write([]byte("###PING " + s + " already pinged"))
|
||||
}
|
||||
for _, peer := range w.host.Network().Peers() {
|
||||
log.Debug("Pinging", peer)
|
||||
w.ping.Ping(w.ctx, peer)
|
||||
}
|
||||
case <-w.quit:
|
||||
ticker.Stop()
|
||||
|
|
|
|||
|
|
@ -6,7 +6,6 @@ import (
|
|||
"fmt"
|
||||
|
||||
logging "github.com/ipfs/go-log"
|
||||
"github.com/libp2p/go-libp2p-core/event"
|
||||
"github.com/libp2p/go-libp2p-core/host"
|
||||
"github.com/libp2p/go-libp2p-core/network"
|
||||
"github.com/libp2p/go-libp2p-core/peer"
|
||||
|
|
@ -34,7 +33,7 @@ type (
|
|||
Filters map[string]Filter
|
||||
|
||||
Subscriber struct {
|
||||
peer string
|
||||
peer peer.ID
|
||||
requestId string
|
||||
filter pb.FilterRequest // @TODO MAKE THIS A SEQUENCE AGAIN?
|
||||
}
|
||||
|
|
@ -47,7 +46,6 @@ type (
|
|||
subscribers []Subscriber
|
||||
pushHandler MessagePushHandler
|
||||
MsgC chan *protocol.Envelope
|
||||
peerChan chan *event.EvtPeerConnectednessChanged
|
||||
}
|
||||
)
|
||||
|
||||
|
|
@ -107,16 +105,16 @@ func (wf *WakuFilter) onRequest(s network.Stream) {
|
|||
// We're on a full node.
|
||||
// This is a filter request coming from a light node.
|
||||
if filterRPCRequest.Request.Subscribe {
|
||||
subscriber := Subscriber{peer: string(s.Conn().RemotePeer()), requestId: filterRPCRequest.RequestId, filter: *filterRPCRequest.Request}
|
||||
subscriber := Subscriber{peer: s.Conn().RemotePeer(), requestId: filterRPCRequest.RequestId, filter: *filterRPCRequest.Request}
|
||||
wf.subscribers = append(wf.subscribers, subscriber)
|
||||
log.Info("filter full node, add a filter subscriber: ", subscriber.peer)
|
||||
|
||||
stats.Record(wf.ctx, metrics.FilterSubscriptions.M(int64(len(wf.subscribers))))
|
||||
} else {
|
||||
peerId := string(s.Conn().RemotePeer())
|
||||
log.Info("filter full node, remove a filter subscriber: ", peerId)
|
||||
peerId := s.Conn().RemotePeer()
|
||||
log.Info("filter full node, remove a filter subscriber: ", peerId.Pretty())
|
||||
contentFilters := filterRPCRequest.Request.ContentFilters
|
||||
var peerIdsToRemove []string
|
||||
var peerIdsToRemove []peer.ID
|
||||
for _, subscriber := range wf.subscribers {
|
||||
if subscriber.peer != peerId {
|
||||
continue
|
||||
|
|
@ -158,26 +156,7 @@ func (wf *WakuFilter) onRequest(s network.Stream) {
|
|||
}
|
||||
}
|
||||
|
||||
func (wf *WakuFilter) peerListener() {
|
||||
for e := range wf.peerChan {
|
||||
if e.Connectedness == network.NotConnected {
|
||||
log.Info("filter Notification received ", e.Peer)
|
||||
i := 0
|
||||
// Delete subscribers matching deleted peer
|
||||
for _, s := range wf.subscribers {
|
||||
if s.peer != string(e.Peer) {
|
||||
wf.subscribers[i] = s
|
||||
i++
|
||||
}
|
||||
}
|
||||
|
||||
log.Info("filter, deleted subscribers: ", len(wf.subscribers)-i)
|
||||
wf.subscribers = wf.subscribers[:i]
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func NewWakuFilter(ctx context.Context, host host.Host, handler MessagePushHandler, peerChan chan *event.EvtPeerConnectednessChanged) *WakuFilter {
|
||||
func NewWakuFilter(ctx context.Context, host host.Host, handler MessagePushHandler) *WakuFilter {
|
||||
ctx, err := tag.New(ctx, tag.Insert(metrics.KeyType, "filter"))
|
||||
if err != nil {
|
||||
log.Error(err)
|
||||
|
|
@ -188,11 +167,9 @@ func NewWakuFilter(ctx context.Context, host host.Host, handler MessagePushHandl
|
|||
wf.MsgC = make(chan *protocol.Envelope)
|
||||
wf.h = host
|
||||
wf.pushHandler = handler
|
||||
wf.peerChan = peerChan
|
||||
|
||||
wf.h.SetStreamHandlerMatch(FilterID_v20beta1, protocol.PrefixTextMatch(string(FilterID_v20beta1)), wf.onRequest)
|
||||
go wf.FilterListener()
|
||||
go wf.peerListener()
|
||||
|
||||
return wf
|
||||
}
|
||||
|
|
@ -221,7 +198,7 @@ func (wf *WakuFilter) FilterListener() {
|
|||
log.Info("pushing a message to light node: ", pushRPC)
|
||||
|
||||
conn, err := wf.h.NewStream(wf.ctx, peer.ID(subscriber.peer), FilterID_v20beta1)
|
||||
|
||||
// TODO: keep track of errors to automatically unsubscribe a peer?
|
||||
if err != nil {
|
||||
// @TODO more sophisticated error handling here
|
||||
log.Error("failed to open peer stream")
|
||||
|
|
|
|||
|
|
@ -13,7 +13,6 @@ import (
|
|||
"time"
|
||||
|
||||
logging "github.com/ipfs/go-log"
|
||||
"github.com/libp2p/go-libp2p-core/event"
|
||||
"github.com/libp2p/go-libp2p-core/host"
|
||||
"github.com/libp2p/go-libp2p-core/network"
|
||||
"github.com/libp2p/go-libp2p-core/peer"
|
||||
|
|
@ -211,8 +210,6 @@ type WakuStore struct {
|
|||
storeMsgs bool
|
||||
msgProvider MessageProvider
|
||||
h host.Host
|
||||
|
||||
peerChan chan *event.EvtPeerConnectednessChanged
|
||||
}
|
||||
|
||||
func NewWakuStore(shouldStoreMessages bool, p MessageProvider) *WakuStore {
|
||||
|
|
@ -229,18 +226,9 @@ func (store *WakuStore) SetMsgProvider(p MessageProvider) {
|
|||
store.msgProvider = p
|
||||
}
|
||||
|
||||
func (store *WakuStore) peerListener() {
|
||||
for e := range store.peerChan {
|
||||
if e.Connectedness == network.NotConnected {
|
||||
log.Info("Notification received ", e.Peer)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (store *WakuStore) Start(ctx context.Context, h host.Host, peerChan chan *event.EvtPeerConnectednessChanged) {
|
||||
func (store *WakuStore) Start(ctx context.Context, h host.Host) {
|
||||
store.h = h
|
||||
store.ctx = ctx
|
||||
store.peerChan = peerChan
|
||||
|
||||
if !store.storeMsgs {
|
||||
log.Info("Store protocol started (messages aren't stored)")
|
||||
|
|
@ -279,8 +267,6 @@ func (store *WakuStore) Start(ctx context.Context, h host.Host, peerChan chan *e
|
|||
}
|
||||
}
|
||||
|
||||
go store.peerListener()
|
||||
|
||||
log.Info("Store protocol started")
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -14,7 +14,6 @@ import (
|
|||
|
||||
"github.com/libp2p/go-libp2p-core/crypto"
|
||||
"github.com/libp2p/go-libp2p-core/discovery"
|
||||
"github.com/libp2p/go-libp2p-core/event"
|
||||
"github.com/libp2p/go-libp2p-core/host"
|
||||
"github.com/libp2p/go-libp2p-core/network"
|
||||
"github.com/libp2p/go-libp2p-core/peer"
|
||||
|
|
@ -507,8 +506,6 @@ func WithMaxMessageSize(maxMessageSize int) Option {
|
|||
|
||||
// processLoop handles all inputs arriving on the channels
|
||||
func (p *PubSub) processLoop(ctx context.Context) {
|
||||
em, _ := p.host.EventBus().Emitter(new(event.EvtPeerConnectednessChanged))
|
||||
|
||||
defer func() {
|
||||
// Clean up go routines.
|
||||
for _, ch := range p.peers {
|
||||
|
|
@ -516,7 +513,6 @@ func (p *PubSub) processLoop(ctx context.Context) {
|
|||
}
|
||||
p.peers = nil
|
||||
p.topics = nil
|
||||
em.Close() // MUST call this after being done with the emitter
|
||||
}()
|
||||
|
||||
for {
|
||||
|
|
@ -537,8 +533,6 @@ func (p *PubSub) processLoop(ctx context.Context) {
|
|||
go p.handleNewPeer(ctx, pid, messages)
|
||||
p.peers[pid] = messages
|
||||
|
||||
em.Emit(event.EvtPeerConnectednessChanged{Peer: pid, Connectedness: network.Connected})
|
||||
|
||||
case s := <-p.newPeerStream:
|
||||
pid := s.Conn().RemotePeer()
|
||||
|
||||
|
|
@ -560,7 +554,6 @@ func (p *PubSub) processLoop(ctx context.Context) {
|
|||
|
||||
case pid := <-p.newPeerError:
|
||||
delete(p.peers, pid)
|
||||
em.Emit(event.EvtPeerConnectednessChanged{Peer: pid, Connectedness: network.NotConnected})
|
||||
|
||||
case pid := <-p.peerDead:
|
||||
ch, ok := p.peers[pid]
|
||||
|
|
@ -590,7 +583,6 @@ func (p *PubSub) processLoop(ctx context.Context) {
|
|||
}
|
||||
|
||||
p.rt.RemovePeer(pid)
|
||||
em.Emit(event.EvtPeerConnectednessChanged{Peer: pid, Connectedness: network.NotConnected})
|
||||
|
||||
case treq := <-p.getTopics:
|
||||
var out []string
|
||||
|
|
|
|||
|
|
@ -427,7 +427,7 @@ github.com/spacemonkeygo/spacelog
|
|||
github.com/status-im/doubleratchet
|
||||
# github.com/status-im/go-multiaddr-ethv4 v1.2.0
|
||||
github.com/status-im/go-multiaddr-ethv4
|
||||
# github.com/status-im/go-waku v0.0.0-20211005021624-79e21dbb993c
|
||||
# github.com/status-im/go-waku v0.0.0-20211006192541-761ae88bbdea
|
||||
github.com/status-im/go-waku/waku/v2/metrics
|
||||
github.com/status-im/go-waku/waku/v2/node
|
||||
github.com/status-im/go-waku/waku/v2/protocol
|
||||
|
|
@ -440,7 +440,7 @@ github.com/status-im/go-waku/waku/v2/utils
|
|||
# github.com/status-im/go-waku-rendezvous v0.0.0-20211005020656-b53661c58574
|
||||
github.com/status-im/go-waku-rendezvous
|
||||
github.com/status-im/go-waku-rendezvous/pb
|
||||
# github.com/status-im/go-wakurelay-pubsub v0.4.3-0.20210729162817-adc68830282a
|
||||
# github.com/status-im/go-wakurelay-pubsub v0.4.3-0.20211006151908-91851dda6375
|
||||
github.com/status-im/go-wakurelay-pubsub
|
||||
github.com/status-im/go-wakurelay-pubsub/pb
|
||||
# github.com/status-im/keycard-go v0.0.0-20200402102358-957c09536969
|
||||
|
|
|
|||
Loading…
Reference in New Issue