-
Notifications
You must be signed in to change notification settings - Fork 248
/
Copy pathmessage_publishing.go
118 lines (98 loc) · 3.23 KB
/
message_publishing.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
package wakuv2
import (
"errors"
"go.uber.org/zap"
"github.com/waku-org/go-waku/waku/v2/api/publish"
"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/relay"
gethcommon "github.com/ethereum/go-ethereum/common"
gocommon "github.com/status-im/status-go/common"
"github.com/status-im/status-go/wakuv2/common"
)
// Send injects a message into the waku send queue, to be distributed in the
// network in the coming cycles.
func (w *Waku) Send(pubsubTopic string, msg *pb.WakuMessage, priority *int) ([]byte, error) {
pubsubTopic = w.GetPubsubTopic(pubsubTopic)
if w.protectedTopicStore != nil {
privKey, err := w.protectedTopicStore.FetchPrivateKey(pubsubTopic)
if err != nil {
return nil, err
}
if privKey != nil {
err = relay.SignMessage(privKey, msg, pubsubTopic)
if err != nil {
return nil, err
}
}
}
envelope := protocol.NewEnvelope(msg, msg.GetTimestamp(), pubsubTopic)
if priority != nil {
err := w.sendQueue.Push(w.ctx, envelope, *priority)
if err != nil {
return nil, err
}
} else {
err := w.sendQueue.Push(w.ctx, envelope)
if err != nil {
return nil, err
}
}
w.poolMu.Lock()
alreadyCached := w.envelopeCache.Has(gethcommon.BytesToHash(envelope.Hash().Bytes()))
w.poolMu.Unlock()
if !alreadyCached {
recvMessage := common.NewReceivedMessage(envelope, common.SendMessageType)
w.postEvent(recvMessage) // notify the local node about the new message
w.addEnvelope(recvMessage)
}
return envelope.Hash().Bytes(), nil
}
func (w *Waku) broadcast() {
defer gocommon.LogOnPanic()
defer w.wg.Done()
for {
var envelope *protocol.Envelope
select {
case envelope = <-w.sendQueue.Pop(w.ctx):
case <-w.ctx.Done():
return
}
w.wg.Add(1)
go w.publishEnvelope(envelope)
}
}
func (w *Waku) publishEnvelope(envelope *protocol.Envelope) {
defer gocommon.LogOnPanic()
defer w.wg.Done()
logger := w.logger.With(zap.Stringer("envelopeHash", envelope.Hash()), zap.String("pubsubTopic", envelope.PubsubTopic()), zap.String("contentTopic", envelope.Message().ContentTopic), zap.Int64("timestamp", envelope.Message().GetTimestamp()))
var err error
// only used in testing to simulate going offline
if w.cfg.SkipPublishToTopic {
logger.Info("skipping publish to topic")
err = errors.New("test send failure")
} else {
err = w.messageSender.Send(publish.NewRequest(w.ctx, envelope))
}
if w.statusTelemetryClient != nil {
if err == nil {
w.statusTelemetryClient.PushSentEnvelope(w.ctx, SentEnvelope{Envelope: envelope, PublishMethod: w.messageSender.PublishMethod()})
} else {
w.statusTelemetryClient.PushErrorSendingEnvelope(w.ctx, ErrorSendingEnvelope{Error: err, SentEnvelope: SentEnvelope{Envelope: envelope, PublishMethod: w.messageSender.PublishMethod()}})
}
}
if err != nil {
logger.Error("could not send message", zap.Error(err))
w.SendEnvelopeEvent(common.EnvelopeEvent{
Hash: gethcommon.BytesToHash(envelope.Hash().Bytes()),
Event: common.EventEnvelopeExpired,
})
return
}
if !w.cfg.EnableStoreConfirmationForMessagesSent {
w.SendEnvelopeEvent(common.EnvelopeEvent{
Hash: gethcommon.BytesToHash(envelope.Hash().Bytes()),
Event: common.EventEnvelopeSent,
})
}
}