status-im
/
status-go
mirror of https://github.com/status-im/status-go.git
2
0
Fork 0
Code Issues Projects Releases Wiki Activity

refactor_: decouple go-waku api from go-waku wakunode

This commit is contained in:
Richard Ramos 2024-10-15 19:18:27 -04:00
parent 61f3d00029
commit 38043d3825
No known key found for this signature in database
GPG Key ID: 1CE87DB518195760
11 changed files with 213 additions and 58 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
go.mod
View File

@ -95,7 +95,7 @@ require (
github.com/schollz/peerdiscovery v1.7.0
github.com/siphiuel/lc-proxy-wrapper v0.0.0-20230516150924-246507cee8c7
github.com/urfave/cli/v2 v2.27.2
github.com/waku-org/go-waku v0.8.1-0.20241014185851-76275f6fb835
github.com/waku-org/go-waku v0.8.1-0.20241015194815-37f936d74705
github.com/wk8/go-ordered-map/v2 v2.1.7
github.com/yeqown/go-qrcode/v2 v2.2.1
github.com/yeqown/go-qrcode/writer/standard v1.2.1

4
go.sum
View File

@ -2140,8 +2140,8 @@ github.com/waku-org/go-libp2p-pubsub v0.12.0-gowaku.0.20240823143342-b0f2429ca27
github.com/waku-org/go-libp2p-pubsub v0.12.0-gowaku.0.20240823143342-b0f2429ca27f/go.mod h1:Oi0zw9aw8/Y5GC99zt+Ef2gYAl+0nZlwdJonDyOz/sE=
github.com/waku-org/go-libp2p-rendezvous v0.0.0-20240110193335-a67d1cc760a0 h1:R4YYx2QamhBRl/moIxkDCNW+OP7AHbyWLBygDc/xIMo=
github.com/waku-org/go-libp2p-rendezvous v0.0.0-20240110193335-a67d1cc760a0/go.mod h1:EhZP9fee0DYjKH/IOQvoNSy1tSHp2iZadsHGphcAJgY=
github.com/waku-org/go-waku v0.8.1-0.20241014185851-76275f6fb835 h1:Vp6BhXiDEilmchHy8OLMZVhugudsnvveNkAKD5nhAGk=
github.com/waku-org/go-waku v0.8.1-0.20241014185851-76275f6fb835/go.mod h1:1BRnyg2mQ2aBNLTBaPq6vEvobzywGykPOhGQFbHGf74=
github.com/waku-org/go-waku v0.8.1-0.20241015194815-37f936d74705 h1:i1vIOgWIQn0jing5jxqO9rG676jPoShiTLknE/pRaWc=
github.com/waku-org/go-waku v0.8.1-0.20241015194815-37f936d74705/go.mod h1:1BRnyg2mQ2aBNLTBaPq6vEvobzywGykPOhGQFbHGf74=
github.com/waku-org/go-zerokit-rln v0.1.14-0.20240102145250-fa738c0bdf59 h1:jisj+OCI6QydLtFq3Pyhu49wl9ytPN7oAHjMfepHDrA=
github.com/waku-org/go-zerokit-rln v0.1.14-0.20240102145250-fa738c0bdf59/go.mod h1:1PdBdPzyTaKt3VnpAHk3zj+r9dXPFOr3IHZP9nFle6E=
github.com/waku-org/go-zerokit-rln-apple v0.0.0-20230916172309-ee0ee61dde2b h1:KgZVhsLkxsj5gb/FfndSCQu6VYwALrCOgYI3poR95yE=

17
vendor/github.com/waku-org/go-waku/waku/v2/api/common/result.go generated vendored Normal file
View File

@ -0,0 +1,17 @@
package common
import (
"context"
"github.com/libp2p/go-libp2p/core/peer"
"github.com/waku-org/go-waku/waku/v2/protocol/store"
"github.com/waku-org/go-waku/waku/v2/protocol/store/pb"
)
type StoreRequestResult interface {
Cursor() []byte
IsComplete() bool
PeerID() peer.ID
Next(ctx context.Context, opts ...store.RequestOption) error // TODO: see how to decouple store.RequestOption
Messages() []*pb.WakuMessageKeyValue
}

33
vendor/github.com/waku-org/go-waku/waku/v2/api/missing/default_requestor.go generated vendored Normal file
View File

@ -0,0 +1,33 @@
package missing
import (
"context"
"github.com/libp2p/go-libp2p/core/peer"
"github.com/waku-org/go-waku/waku/v2/api/common"
"github.com/waku-org/go-waku/waku/v2/protocol"
"github.com/waku-org/go-waku/waku/v2/protocol/pb"
"github.com/waku-org/go-waku/waku/v2/protocol/store"
)
func NewDefaultStorenodeRequestor(store *store.WakuStore) StorenodeRequestor {
return &defaultStorenodeRequestor{
store: store,
}
}
type defaultStorenodeRequestor struct {
store *store.WakuStore
}
func (d *defaultStorenodeRequestor) GetMessagesByHash(ctx context.Context, peerID peer.ID, pageSize uint64, messageHashes []pb.MessageHash) (common.StoreRequestResult, error) {
return d.store.QueryByHash(ctx, messageHashes, store.WithPeer(peerID), store.WithPaging(false, pageSize))
}
func (d *defaultStorenodeRequestor) QueryWithCriteria(ctx context.Context, peerID peer.ID, pageSize uint64, pubsubTopic string, contentTopics []string, from *int64, to *int64) (common.StoreRequestResult, error) {
return d.store.Query(ctx, store.FilterCriteria{
ContentFilter: protocol.NewContentFilter(pubsubTopic, contentTopics...),
TimeStart: from,
TimeEnd: to,
}, store.WithPeer(peerID), store.WithPaging(false, pageSize), store.IncludeData(false))
}

50
vendor/github.com/waku-org/go-waku/waku/v2/api/missing/missing_messages.go generated vendored
View File

@ -11,9 +11,9 @@ import (
"github.com/libp2p/go-libp2p/core/peer"
"github.com/waku-org/go-waku/logging"
"github.com/waku-org/go-waku/waku/v2/api/common"
"github.com/waku-org/go-waku/waku/v2/protocol"
"github.com/waku-org/go-waku/waku/v2/protocol/pb"
"github.com/waku-org/go-waku/waku/v2/protocol/store"
"github.com/waku-org/go-waku/waku/v2/timesource"
"github.com/waku-org/go-waku/waku/v2/utils"
"go.uber.org/zap"
@ -22,6 +22,7 @@ import (
const maxContentTopicsPerRequest = 10
const maxMsgHashesPerRequest = 50
const messageFetchPageSize = 100
// MessageTracker should keep track of messages it has seen before and
// provide a way to determine whether a message exists or not. This
@ -30,25 +31,30 @@ type MessageTracker interface {
MessageExists(pb.MessageHash) (bool, error)
}
type StorenodeRequestor interface {
GetMessagesByHash(ctx context.Context, peerID peer.ID, pageSize uint64, messageHashes []pb.MessageHash) (common.StoreRequestResult, error)
QueryWithCriteria(ctx context.Context, peerID peer.ID, pageSize uint64, pubsubTopic string, contentTopics []string, from *int64, to *int64) (common.StoreRequestResult, error)
}
// MissingMessageVerifier is used to periodically retrieve missing messages from store nodes that have some specific criteria
type MissingMessageVerifier struct {
ctx context.Context
params missingMessageVerifierParams
messageTracker MessageTracker
storenodeRequestor StorenodeRequestor
messageTracker MessageTracker
criteriaInterest map[string]criteriaInterest // Track message verification requests and when was the last time a pubsub topic was verified for missing messages
criteriaInterestMu sync.RWMutex
C <-chan *protocol.Envelope
store *store.WakuStore
timesource timesource.Timesource
logger *zap.Logger
}
// NewMissingMessageVerifier creates an instance of a MissingMessageVerifier
func NewMissingMessageVerifier(store *store.WakuStore, messageTracker MessageTracker, timesource timesource.Timesource, logger *zap.Logger, options ...MissingMessageVerifierOption) *MissingMessageVerifier {
func NewMissingMessageVerifier(storenodeRequester StorenodeRequestor, messageTracker MessageTracker, timesource timesource.Timesource, logger *zap.Logger, options ...MissingMessageVerifierOption) *MissingMessageVerifier {
options = append(defaultMissingMessagesVerifierOptions, options...)
params := missingMessageVerifierParams{}
for _, opt := range options {
@ -56,11 +62,11 @@ func NewMissingMessageVerifier(store *store.WakuStore, messageTracker MessageTra
}
return &MissingMessageVerifier{
store: store,
timesource: timesource,
messageTracker: messageTracker,
logger: logger.Named("missing-msg-verifier"),
params: params,
storenodeRequestor: storenodeRequester,
timesource: timesource,
messageTracker: messageTracker,
logger: logger.Named("missing-msg-verifier"),
params: params,
}
}
@ -178,7 +184,7 @@ func (m *MissingMessageVerifier) fetchHistory(c chan<- *protocol.Envelope, inter
}
}
func (m *MissingMessageVerifier) storeQueryWithRetry(ctx context.Context, queryFunc func(ctx context.Context) (store.Result, error), logger *zap.Logger, logMsg string) (store.Result, error) {
func (m *MissingMessageVerifier) storeQueryWithRetry(ctx context.Context, queryFunc func(ctx context.Context) (common.StoreRequestResult, error), logger *zap.Logger, logMsg string) (common.StoreRequestResult, error) {
retry := true
count := 1
for retry && count <= m.params.maxAttemptsToRetrieveHistory {
@ -212,12 +218,16 @@ func (m *MissingMessageVerifier) fetchMessagesBatch(c chan<- *protocol.Envelope,
logging.Epoch("to", now),
)
result, err := m.storeQueryWithRetry(interest.ctx, func(ctx context.Context) (store.Result, error) {
return m.store.Query(ctx, store.FilterCriteria{
ContentFilter: protocol.NewContentFilter(interest.contentFilter.PubsubTopic, contentTopics[batchFrom:batchTo]...),
TimeStart: proto.Int64(interest.lastChecked.Add(-m.params.delay).UnixNano()),
TimeEnd: proto.Int64(now.Add(-m.params.delay).UnixNano()),
}, store.WithPeer(interest.peerID), store.WithPaging(false, 100), store.IncludeData(false))
result, err := m.storeQueryWithRetry(interest.ctx, func(ctx context.Context) (common.StoreRequestResult, error) {
return m.storenodeRequestor.QueryWithCriteria(
ctx,
interest.peerID,
messageFetchPageSize,
interest.contentFilter.PubsubTopic,
contentTopics[batchFrom:batchTo],
proto.Int64(interest.lastChecked.Add(-m.params.delay).UnixNano()),
proto.Int64(now.Add(-m.params.delay).UnixNano()),
)
}, logger, "retrieving history to check for missing messages")
if err != nil {
if !errors.Is(err, context.Canceled) {
@ -243,7 +253,7 @@ func (m *MissingMessageVerifier) fetchMessagesBatch(c chan<- *protocol.Envelope,
missingHashes = append(missingHashes, hash)
}
result, err = m.storeQueryWithRetry(interest.ctx, func(ctx context.Context) (store.Result, error) {
result, err = m.storeQueryWithRetry(interest.ctx, func(ctx context.Context) (common.StoreRequestResult, error) {
if err = result.Next(ctx); err != nil {
return nil, err
}
@ -282,10 +292,10 @@ func (m *MissingMessageVerifier) fetchMessagesBatch(c chan<- *protocol.Envelope,
defer utils.LogOnPanic()
defer wg.Wait()
result, err := m.storeQueryWithRetry(interest.ctx, func(ctx context.Context) (store.Result, error) {
result, err := m.storeQueryWithRetry(interest.ctx, func(ctx context.Context) (common.StoreRequestResult, error) {
queryCtx, cancel := context.WithTimeout(ctx, m.params.storeQueryTimeout)
defer cancel()
return m.store.QueryByHash(queryCtx, messageHashes, store.WithPeer(interest.peerID), store.WithPaging(false, maxMsgHashesPerRequest))
return m.storenodeRequestor.GetMessagesByHash(queryCtx, interest.peerID, maxMsgHashesPerRequest, messageHashes)
}, logger, "retrieving missing messages")
if err != nil {
if !errors.Is(err, context.Canceled) {
@ -303,7 +313,7 @@ func (m *MissingMessageVerifier) fetchMessagesBatch(c chan<- *protocol.Envelope,
}
}
result, err = m.storeQueryWithRetry(interest.ctx, func(ctx context.Context) (store.Result, error) {
result, err = m.storeQueryWithRetry(interest.ctx, func(ctx context.Context) (common.StoreRequestResult, error) {
if err = result.Next(ctx); err != nil {
return nil, err
}

50
vendor/github.com/waku-org/go-waku/waku/v2/api/publish/default_publisher.go generated vendored Normal file
View File

@ -0,0 +1,50 @@
package publish
import (
"context"
"errors"
"github.com/libp2p/go-libp2p/core/peer"
"github.com/waku-org/go-waku/waku/v2/protocol/lightpush"
"github.com/waku-org/go-waku/waku/v2/protocol/pb"
"github.com/waku-org/go-waku/waku/v2/protocol/relay"
)
var ErrRelayNotAvailable = errors.New("relay is not available")
var ErrLightpushNotAvailable = errors.New("lightpush is not available")
func NewDefaultPublisher(lightpush *lightpush.WakuLightPush, relay *relay.WakuRelay) Publisher {
return &defaultPublisher{
lightpush: lightpush,
relay: relay,
}
}
type defaultPublisher struct {
lightpush *lightpush.WakuLightPush
relay *relay.WakuRelay
}
func (d *defaultPublisher) RelayListPeers(pubsubTopic string) ([]peer.ID, error) {
if d.relay == nil {
return nil, ErrRelayNotAvailable
}
return d.relay.PubSub().ListPeers(pubsubTopic), nil
}
func (d *defaultPublisher) RelayPublish(ctx context.Context, message *pb.WakuMessage, pubsubTopic string) (pb.MessageHash, error) {
if d.relay == nil {
return pb.MessageHash{}, ErrRelayNotAvailable
}
return d.relay.Publish(ctx, message, relay.WithPubSubTopic(pubsubTopic))
}
func (d *defaultPublisher) LightpushPublish(ctx context.Context, message *pb.WakuMessage, pubsubTopic string, maxPeers int) (pb.MessageHash, error) {
if d.lightpush == nil {
return pb.MessageHash{}, ErrLightpushNotAvailable
}
return d.lightpush.Publish(ctx, message, lightpush.WithPubSubTopic(pubsubTopic), lightpush.WithMaxPeers(maxPeers))
}

39
vendor/github.com/waku-org/go-waku/waku/v2/api/publish/default_verifier.go generated vendored Normal file
View File

@ -0,0 +1,39 @@
package publish
import (
"context"
"github.com/libp2p/go-libp2p/core/peer"
"github.com/waku-org/go-waku/waku/v2/protocol/pb"
"github.com/waku-org/go-waku/waku/v2/protocol/store"
)
func NewDefaultStorenodeMessageVerifier(store *store.WakuStore) StorenodeMessageVerifier {
return &defaultStorenodeMessageVerifier{
store: store,
}
}
type defaultStorenodeMessageVerifier struct {
store *store.WakuStore
}
func (d *defaultStorenodeMessageVerifier) MessageHashesExist(ctx context.Context, requestID []byte, peerID peer.ID, pageSize uint64, messageHashes []pb.MessageHash) ([]pb.MessageHash, error) {
var opts []store.RequestOption
opts = append(opts, store.WithRequestID(requestID))
opts = append(opts, store.WithPeer(peerID))
opts = append(opts, store.WithPaging(false, pageSize))
opts = append(opts, store.IncludeData(false))
response, err := d.store.QueryByHash(ctx, messageHashes, opts...)
if err != nil {
return nil, err
}
result := make([]pb.MessageHash, len(response.Messages()))
for i, msg := range response.Messages() {
result[i] = msg.WakuMessageHash()
}
return result, nil
}

26
vendor/github.com/waku-org/go-waku/waku/v2/api/publish/message_check.go generated vendored
View File

@ -8,11 +8,11 @@ import (
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/common/hexutil"
"github.com/libp2p/go-libp2p/core/peer"
apicommon "github.com/waku-org/go-waku/waku/v2/api/common"
"github.com/waku-org/go-waku/waku/v2/api/history"
"github.com/waku-org/go-waku/waku/v2/protocol"
"github.com/waku-org/go-waku/waku/v2/protocol/pb"
"github.com/waku-org/go-waku/waku/v2/protocol/store"
"github.com/waku-org/go-waku/waku/v2/timesource"
"github.com/waku-org/go-waku/waku/v2/utils"
"go.uber.org/zap"
@ -31,6 +31,11 @@ type ISentCheck interface {
DeleteByMessageIDs(messageIDs []common.Hash)
}
type StorenodeMessageVerifier interface {
// MessagesExist returns a list of the messages it found from a list of message hashes
MessageHashesExist(ctx context.Context, requestID []byte, peerID peer.ID, pageSize uint64, messageHashes []pb.MessageHash) ([]pb.MessageHash, error)
}
// MessageSentCheck tracks the outgoing messages and check against store node
// if the message sent time has passed the `messageSentPeriod`, the message id will be includes for the next query
// if the message keeps missing after `messageExpiredPerid`, the message id will be expired
@ -40,7 +45,7 @@ type MessageSentCheck struct {
messageStoredChan chan common.Hash
messageExpiredChan chan common.Hash
ctx context.Context
store *store.WakuStore
messageVerifier StorenodeMessageVerifier
storenodeCycle *history.StorenodeCycle
timesource timesource.Timesource
logger *zap.Logger
@ -52,14 +57,14 @@ type MessageSentCheck struct {
}
// NewMessageSentCheck creates a new instance of MessageSentCheck with default parameters
func NewMessageSentCheck(ctx context.Context, store *store.WakuStore, cycle *history.StorenodeCycle, timesource timesource.Timesource, msgStoredChan chan common.Hash, msgExpiredChan chan common.Hash, logger *zap.Logger) *MessageSentCheck {
func NewMessageSentCheck(ctx context.Context, messageVerifier StorenodeMessageVerifier, cycle *history.StorenodeCycle, timesource timesource.Timesource, msgStoredChan chan common.Hash, msgExpiredChan chan common.Hash, logger *zap.Logger) *MessageSentCheck {
return &MessageSentCheck{
messageIDs: make(map[string]map[common.Hash]uint32),
messageIDsMu: sync.RWMutex{},
messageStoredChan: msgStoredChan,
messageExpiredChan: msgExpiredChan,
ctx: ctx,
store: store,
messageVerifier: messageVerifier,
storenodeCycle: cycle,
timesource: timesource,
logger: logger,
@ -212,12 +217,7 @@ func (m *MessageSentCheck) messageHashBasedQuery(ctx context.Context, hashes []c
return []common.Hash{}
}
var opts []store.RequestOption
requestID := protocol.GenerateRequestID()
opts = append(opts, store.WithRequestID(requestID))
opts = append(opts, store.WithPeer(selectedPeer))
opts = append(opts, store.WithPaging(false, m.maxHashQueryLength))
opts = append(opts, store.IncludeData(false))
messageHashes := make([]pb.MessageHash, len(hashes))
for i, hash := range hashes {
@ -228,20 +228,20 @@ func (m *MessageSentCheck) messageHashBasedQuery(ctx context.Context, hashes []c
queryCtx, cancel := context.WithTimeout(ctx, m.storeQueryTimeout)
defer cancel()
result, err := m.store.QueryByHash(queryCtx, messageHashes, opts...)
result, err := m.messageVerifier.MessageHashesExist(queryCtx, requestID, selectedPeer, m.maxHashQueryLength, messageHashes)
if err != nil {
m.logger.Error("store.queryByHash failed", zap.String("requestID", hexutil.Encode(requestID)), zap.Stringer("peerID", selectedPeer), zap.Error(err))
return []common.Hash{}
}
m.logger.Debug("store.queryByHash result", zap.String("requestID", hexutil.Encode(requestID)), zap.Int("messages", len(result.Messages())))
m.logger.Debug("store.queryByHash result", zap.String("requestID", hexutil.Encode(requestID)), zap.Int("messages", len(result)))
var ackHashes []common.Hash
var missedHashes []common.Hash
for i, hash := range hashes {
found := false
for _, msg := range result.Messages() {
if bytes.Equal(msg.GetMessageHash(), hash.Bytes()) {
for _, msgHash := range result {
if bytes.Equal(msgHash.Bytes(), hash.Bytes()) {
found = true
break
}

42
vendor/github.com/waku-org/go-waku/waku/v2/api/publish/message_sender.go generated vendored
View File

@ -6,9 +6,9 @@ import (
"time"
"github.com/ethereum/go-ethereum/common"
"github.com/libp2p/go-libp2p/core/peer"
"github.com/waku-org/go-waku/waku/v2/protocol"
"github.com/waku-org/go-waku/waku/v2/protocol/lightpush"
"github.com/waku-org/go-waku/waku/v2/protocol/relay"
"github.com/waku-org/go-waku/waku/v2/protocol/pb"
"go.uber.org/zap"
"golang.org/x/time/rate"
)
@ -36,10 +36,20 @@ func (pm PublishMethod) String() string {
}
}
type Publisher interface {
// RelayListPeers returns the list of peers for a pubsub topic
RelayListPeers(pubsubTopic string) ([]peer.ID, error)
// RelayPublish publishes a message via WakuRelay
RelayPublish(ctx context.Context, message *pb.WakuMessage, pubsubTopic string) (pb.MessageHash, error)
// LightpushPublish publishes a message via WakuLightPush
LightpushPublish(ctx context.Context, message *pb.WakuMessage, pubsubTopic string, maxPeers int) (pb.MessageHash, error)
}
type MessageSender struct {
publishMethod PublishMethod
lightPush *lightpush.WakuLightPush
relay *relay.WakuRelay
publisher Publisher
messageSentCheck ISentCheck
rateLimiter *PublishRateLimiter
logger *zap.Logger
@ -64,14 +74,13 @@ func (r *Request) WithPublishMethod(publishMethod PublishMethod) *Request {
return r
}
func NewMessageSender(publishMethod PublishMethod, lightPush *lightpush.WakuLightPush, relay *relay.WakuRelay, logger *zap.Logger) (*MessageSender, error) {
func NewMessageSender(publishMethod PublishMethod, publisher Publisher, logger *zap.Logger) (*MessageSender, error) {
if publishMethod == UnknownMethod {
return nil, errors.New("publish method is required")
}
return &MessageSender{
publishMethod: publishMethod,
lightPush: lightPush,
relay: relay,
publisher: publisher,
rateLimiter: NewPublishRateLimiter(DefaultPublishingLimiterRate, DefaultPublishingLimitBurst),
logger: logger,
}, nil
@ -108,26 +117,23 @@ func (ms *MessageSender) Send(req *Request) error {
switch publishMethod {
case LightPush:
if ms.lightPush == nil {
return errors.New("lightpush is not available")
}
logger.Info("publishing message via lightpush")
_, err := ms.lightPush.Publish(
_, err := ms.publisher.LightpushPublish(
req.ctx,
req.envelope.Message(),
lightpush.WithPubSubTopic(req.envelope.PubsubTopic()),
lightpush.WithMaxPeers(DefaultPeersToPublishForLightpush),
req.envelope.PubsubTopic(),
DefaultPeersToPublishForLightpush,
)
if err != nil {
return err
}
case Relay:
if ms.relay == nil {
return errors.New("relay is not available")
peers, err := ms.publisher.RelayListPeers(req.envelope.PubsubTopic())
if err != nil {
return err
}
peerCnt := len(ms.relay.PubSub().ListPeers(req.envelope.PubsubTopic()))
logger.Info("publishing message via relay", zap.Int("peerCnt", peerCnt))
_, err := ms.relay.Publish(req.ctx, req.envelope.Message(), relay.WithPubSubTopic(req.envelope.PubsubTopic()))
logger.Info("publishing message via relay", zap.Int("peerCnt", len(peers)))
_, err = ms.publisher.RelayPublish(req.ctx, req.envelope.Message(), req.envelope.PubsubTopic())
if err != nil {
return err
}

2
vendor/modules.txt vendored
View File

@ -1031,7 +1031,7 @@ github.com/waku-org/go-discover/discover/v5wire
github.com/waku-org/go-libp2p-rendezvous
github.com/waku-org/go-libp2p-rendezvous/db
github.com/waku-org/go-libp2p-rendezvous/pb
# github.com/waku-org/go-waku v0.8.1-0.20241014185851-76275f6fb835
# github.com/waku-org/go-waku v0.8.1-0.20241015194815-37f936d74705
## explicit; go 1.21
github.com/waku-org/go-waku/logging
github.com/waku-org/go-waku/tests

6
wakuv2/waku.go
View File

@ -1145,7 +1145,7 @@ func (w *Waku) Start() error {
if w.cfg.EnableMissingMessageVerification {
w.missingMsgVerifier = missing.NewMissingMessageVerifier(
w.node.Store(),
missing.NewDefaultStorenodeRequestor(w.node.Store()),
w,
w.node.Timesource(),
w.logger)
@ -1293,7 +1293,7 @@ func (w *Waku) startMessageSender() error {
publishMethod = publish.LightPush
}
sender, err := publish.NewMessageSender(publishMethod, w.node.Lightpush(), w.node.Relay(), w.logger)
sender, err := publish.NewMessageSender(publishMethod, publish.NewDefaultPublisher(w.node.Lightpush(), w.node.Relay()), w.logger)
if err != nil {
w.logger.Error("failed to create message sender", zap.Error(err))
return err
@ -1302,7 +1302,7 @@ func (w *Waku) startMessageSender() error {
if w.cfg.EnableStoreConfirmationForMessagesSent {
msgStoredChan := make(chan gethcommon.Hash, 1000)
msgExpiredChan := make(chan gethcommon.Hash, 1000)
messageSentCheck := publish.NewMessageSentCheck(w.ctx, w.node.Store(), w.StorenodeCycle, w.node.Timesource(), msgStoredChan, msgExpiredChan, w.logger)
messageSentCheck := publish.NewMessageSentCheck(w.ctx, publish.NewDefaultStorenodeMessageVerifier(w.node.Store()), w.StorenodeCycle, w.node.Timesource(), msgStoredChan, msgExpiredChan, w.logger)
sender.WithMessageSentCheck(messageSentCheck)
w.wg.Add(1)