besaid.zone
octet-stream.net
+15
-4
jetstream/jetstream.go
+15
-4
jetstream/jetstream.go
···
72
// existing instances of the closure when j.WantedDids is mutated
73
return func(ctx context.Context, evt *models.Event) error {
74
75
// empty filter => all dids allowed
76
-
if len(j.wantedDids) == 0 {
77
-
return processFunc(ctx, evt)
78
}
79
80
-
if _, ok := j.wantedDids[evt.Did]; ok {
81
return processFunc(ctx, evt)
82
} else {
83
return nil
···
122
123
go func() {
124
if j.waitForDid {
125
-
for len(j.wantedDids) == 0 {
126
time.Sleep(time.Second)
127
}
128
}
···
72
// existing instances of the closure when j.WantedDids is mutated
73
return func(ctx context.Context, evt *models.Event) error {
74
75
+
j.mu.RLock()
76
// empty filter => all dids allowed
77
+
matches := len(j.wantedDids) == 0
78
+
if !matches {
79
+
if _, ok := j.wantedDids[evt.Did]; ok {
80
+
matches = true
81
+
}
82
}
83
+
j.mu.RUnlock()
84
85
+
if matches {
86
return processFunc(ctx, evt)
87
} else {
88
return nil
···
127
128
go func() {
129
if j.waitForDid {
130
+
for {
131
+
j.mu.RLock()
132
+
hasDid := len(j.wantedDids) != 0
133
+
j.mu.RUnlock()
134
+
if hasDid {
135
+
break
136
+
}
137
time.Sleep(time.Second)
138
}
139
}