Skip to content

Commit aab65e9

Browse files
mreifersonmreiferson
committed
nsqd: cleanup gossipDelegate
1 parent ff45df6 commit aab65e9

File tree

3 files changed

+10
-45
lines changed

3 files changed

+10
-45
lines changed

‎nsqd/gossip.go‎

Lines changed: 0 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -261,9 +261,6 @@ func (n *NSQD) serfEventLoop() {
261261
default:
262262
n.logf("WARNING: un-handled Serf event: %#v", ev)
263263
}
264-
if n.getOpts().gossipDelegate != nil {
265-
n.getOpts().gossipDelegate.notify()
266-
}
267264
case <-n.exitChan:
268265
goto exit
269266
}

‎nsqd/gossip_test.go‎

Lines changed: 10 additions & 31 deletions
Original file line numberDiff line numberDiff line change
@@ -7,25 +7,11 @@ import (
77
"time"
88
)
99

10-
type gossipTester struct {
11-
c chan struct{}
12-
}
13-
14-
func (g gossipTester) notify() {
15-
g.c <- struct{}{}
16-
select {
17-
case g.c <- struct{}{}:
18-
default:
19-
}
20-
}
21-
2210
func TestGossip(t *testing.T) {
2311
var nsqds []*NSQD
2412
var seedNode *NSQD
2513
var tcpPorts []int
2614

27-
convergenceTester := gossipTester{make(chan struct{}, 20)}
28-
2915
num := 3
3016
for i := 0; i < num; i++ {
3117
// find an open port
@@ -39,7 +25,6 @@ func TestGossip(t *testing.T) {
3925
opts.Logger = newTestLogger(t)
4026
opts.GossipAddress = addr.String()
4127
opts.BroadcastAddress = "127.0.0.1"
42-
opts.gossipDelegate = convergenceTester
4328
if seedNode != nil {
4429
opts.GossipSeedAddresses = []string{seedNode.getOpts().GossipAddress}
4530
}
@@ -57,7 +42,7 @@ func TestGossip(t *testing.T) {
5742
sort.Ints(tcpPorts)
5843

5944
// wait for convergence
60-
converged := converge(5*time.Second, nsqds, convergenceTester.c, func() bool {
45+
converged := converge(5*time.Second, nsqds, func() bool {
6146
for _, nsqd := range nsqds {
6247
if len(nsqd.rdb.FindProducers("client", "", "")) != num {
6348
return false
@@ -86,7 +71,7 @@ func TestGossip(t *testing.T) {
8671
topic.GetChannel("ch")
8772
firstPort := nsqds[0].tcpListener.Addr().(*net.TCPAddr).Port
8873

89-
converged = converge(10*time.Second, nsqds, convergenceTester.c, func() bool {
74+
converged = converge(10*time.Second, nsqds, func() bool {
9075
for _, nsqd := range nsqds {
9176
if len(nsqd.rdb.FindProducers("topic", topicName, "")) != 1 ||
9277
len(nsqd.rdb.FindProducers("channel", topicName, "ch")) != 1 {
@@ -114,8 +99,6 @@ func TestGossipResync(t *testing.T) {
11499
var seedNode *NSQD
115100
var tcpPorts []int
116101

117-
convergenceTester := gossipTester{make(chan struct{}, 20)}
118-
119102
num := 3
120103
for i := 0; i < num; i++ {
121104
// find an open port
@@ -133,7 +116,6 @@ func TestGossipResync(t *testing.T) {
133116
opts.GossipReconnectTimeout = 100 * time.Millisecond
134117
opts.GossipSuspicionMult = 1
135118
opts.GossipProbeInterval = 100 * time.Millisecond
136-
opts.gossipDelegate = convergenceTester
137119
if seedNode != nil {
138120
opts.GossipSeedAddresses = []string{seedNode.getOpts().GossipAddress}
139121
}
@@ -156,7 +138,7 @@ func TestGossipResync(t *testing.T) {
156138
topic.GetChannel("ch")
157139
firstPort := nsqds[0].tcpListener.Addr().(*net.TCPAddr).Port
158140

159-
converged := converge(10*time.Second, nsqds, convergenceTester.c, func() bool {
141+
converged := converge(10*time.Second, nsqds, func() bool {
160142
for _, nsqd := range nsqds {
161143
if len(nsqd.rdb.FindProducers("topic", topicName, "")) != 1 ||
162144
len(nsqd.rdb.FindProducers("channel", topicName, "ch")) != 1 {
@@ -182,7 +164,7 @@ func TestGossipResync(t *testing.T) {
182164
stillAlive := nsqds[:num-1]
183165

184166
// check that other nodes see it as closed
185-
converged = converge(10*time.Second, stillAlive, convergenceTester.c, func() bool {
167+
converged = converge(10*time.Second, stillAlive, func() bool {
186168
for _, nsqd := range stillAlive {
187169
if len(nsqd.serf.Members()) != len(stillAlive) {
188170
return false
@@ -198,7 +180,7 @@ func TestGossipResync(t *testing.T) {
198180
nsqds[num-1] = nsqd
199181

200182
// check that other nodes see it as back open
201-
converged = converge(10*time.Second, nsqds, convergenceTester.c, func() bool {
183+
converged = converge(10*time.Second, nsqds, func() bool {
202184
for _, nsqd := range nsqds {
203185
if len(nsqd.serf.Members()) != len(nsqds) {
204186
return false
@@ -209,7 +191,7 @@ func TestGossipResync(t *testing.T) {
209191
equal(t, converged, true)
210192

211193
// check that all nodes see the restarted first node
212-
converged = converge(10*time.Second, nsqds, convergenceTester.c, func() bool {
194+
converged = converge(10*time.Second, nsqds, func() bool {
213195
for _, nsqd := range nsqds {
214196
if len(nsqd.rdb.FindProducers("topic", topicName, "")) != 1 ||
215197
len(nsqd.rdb.FindProducers("channel", topicName, "ch")) != 1 {
@@ -237,8 +219,6 @@ func TestRegossip(t *testing.T) {
237219
var seedNode *NSQD
238220
var tcpPorts []int
239221

240-
convergenceTester := gossipTester{make(chan struct{}, 20)}
241-
242222
num := 3
243223
for i := 0; i < num; i++ {
244224
// find an open port
@@ -253,7 +233,6 @@ func TestRegossip(t *testing.T) {
253233
opts.GossipAddress = addr.String()
254234
opts.BroadcastAddress = "127.0.0.1"
255235
opts.GossipRegossipInterval = 1 * time.Second
256-
opts.gossipDelegate = convergenceTester
257236
if seedNode != nil {
258237
opts.GossipSeedAddresses = []string{seedNode.getOpts().GossipAddress}
259238
}
@@ -276,7 +255,7 @@ func TestRegossip(t *testing.T) {
276255
topic.GetChannel("ch")
277256
firstPort := nsqds[0].tcpListener.Addr().(*net.TCPAddr).Port
278257

279-
converged := converge(10*time.Second, nsqds, convergenceTester.c, func() bool {
258+
converged := converge(10*time.Second, nsqds, func() bool {
280259
for _, nsqd := range nsqds {
281260
if len(nsqd.rdb.FindProducers("topic", topicName, "")) != 1 ||
282261
len(nsqd.rdb.FindProducers("channel", topicName, "ch")) != 1 {
@@ -305,7 +284,7 @@ func TestRegossip(t *testing.T) {
305284
}
306285

307286
// wait for regossip
308-
converged = converge(10*time.Second, nsqds, convergenceTester.c, func() bool {
287+
converged = converge(10*time.Second, nsqds, func() bool {
309288
for _, nsqd := range nsqds {
310289
if len(nsqd.rdb.FindProducers("topic", topicName, "")) != 1 ||
311290
len(nsqd.rdb.FindProducers("channel", topicName, "ch")) != 1 {
@@ -328,12 +307,12 @@ func TestRegossip(t *testing.T) {
328307
}
329308
}
330309

331-
func converge(timeout time.Duration, nsqds []*NSQD, notifyChan chan struct{}, isConverged func() bool) bool {
310+
func converge(timeout time.Duration, nsqds []*NSQD, isConverged func() bool) bool {
332311
for {
333312
select {
334313
case <-time.After(timeout):
335314
return false
336-
case <-notifyChan:
315+
case <-time.After(10 * time.Millisecond):
337316
if isConverged() {
338317
goto exit
339318
}

‎nsqd/options.go‎

Lines changed: 0 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -80,18 +80,8 @@ type Options struct {
8080
SnappyEnabled bool `flag:"snappy"`
8181

8282
Logger logger
83-
84-
gossipDelegate gossipDelegate
85-
}
86-
87-
type gossipDelegate interface {
88-
notify()
8983
}
9084

91-
type nilGossipDelegate struct{}
92-
93-
func (_ nilGossipDelegate) notify() {}
94-
9585
func NewOptions() *Options {
9686
hostname, err := os.Hostname()
9787
if err != nil {
@@ -150,7 +140,6 @@ func NewOptions() *Options {
150140

151141
Logger: log.New(os.Stderr, "[nsqd] ", log.Ldate|log.Ltime|log.Lmicroseconds),
152142

153-
gossipDelegate: nilGossipDelegate{},
154143
GossipRegossipInterval: 60 * time.Second,
155144
GossipProbeInterval: 1 * time.Second,
156145
GossipSuspicionMult: 5,

0 commit comments

Comments
 (0)