@@ -29,7 +29,11 @@ import (
29
29
30
30
const DefaultPointerPrefixLength = 14
31
31
32
- var log = logging .MustGetLogger ("retriever" )
32
+ var (
33
+ // Initialize a clear pointerList for the DHT on start
34
+ pointerList = []string {}
35
+ log = logging .MustGetLogger ("retriever" )
36
+ )
33
37
34
38
type MRConfig struct {
35
39
Db repo.Datastore
@@ -66,6 +70,20 @@ type offlineMessage struct {
66
70
env pb.Envelope
67
71
}
68
72
73
+ func stringInSlice (str string , list []string ) bool {
74
+ for _ , v := range list {
75
+ if v == str {
76
+ return true
77
+ }
78
+ }
79
+ return false
80
+ }
81
+
82
+ // Reset on startup
83
+ func (m * MessageRetriever ) ResetPointerList () {
84
+ pointerList = []string {}
85
+ }
86
+
69
87
func NewMessageRetriever (cfg MRConfig ) * MessageRetriever {
70
88
var client * http.Client
71
89
if cfg .Dialer != nil {
@@ -100,8 +118,8 @@ func (m *MessageRetriever) Run() {
100
118
peers := time .NewTicker (time .Minute )
101
119
defer dht .Stop ()
102
120
defer peers .Stop ()
103
- go m .fetchPointersFromDHT ()
104
121
go m .fetchPointersFromPushNodes ()
122
+ go m .fetchPointersFromDHT ()
105
123
for {
106
124
select {
107
125
case <- dht .C :
@@ -159,7 +177,8 @@ func (m *MessageRetriever) downloadMessages(peerOut chan ps.PeerInfo) {
159
177
inFlight := make (map [string ]bool )
160
178
// Iterate over the pointers, adding 1 to the waitgroup for each pointer found
161
179
for p := range peerOut {
162
- if len (p .Addrs ) > 0 && ! m .db .OfflineMessages ().Has (p .Addrs [0 ].String ()) && ! inFlight [p .Addrs [0 ].String ()] {
180
+ if len (p .Addrs ) > 0 && ! m .db .OfflineMessages ().Has (p .Addrs [0 ].String ()) && ! stringInSlice (p .Addrs [0 ].String (), pointerList ) && ! inFlight [p .Addrs [0 ].String ()] {
181
+ pointerList = append (pointerList , p .Addrs [0 ].String ())
163
182
inFlight [p .Addrs [0 ].String ()] = true
164
183
log .Debugf ("Found pointer with location %s" , p .Addrs [0 ].String ())
165
184
// IPFS
0 commit comments