Skip to content

Commit a35650e

Browse files
pfeairhellerbrainlag
pfeairheller
authored and
brainlag
committed
Allowing NSQD and NSQLOOKUPD hostnames to be discovered from the environment during test execution.
(cherry picked from commit 91387c4)
1 parent 855a6e6 commit a35650e

File tree

3 files changed

+49
-24
lines changed

3 files changed

+49
-24
lines changed

‎src/test/java/com/github/brainlag/nsq/NSQConsumerTest.java

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -22,7 +22,7 @@ public class NSQConsumerTest {
2222
public void testLongRunningConsumer() throws NSQException, TimeoutException, InterruptedException {
2323
AtomicInteger counter = new AtomicInteger(0);
2424
NSQLookup lookup = new DefaultNSQLookup();
25-
lookup.addLookupAddress("localhost", 4161);
25+
lookup.addLookupAddress(Nsq.getNsqLookupdHost(), 4161);
2626

2727
NSQConsumer consumer = new NSQConsumer(lookup, "test1", "testconsumer", (message) -> {
2828
LogManager.getLogger(this).info("Processing message: " + new String(message.getMessage()));
@@ -43,7 +43,7 @@ public void testLongRunningConsumer() throws NSQException, TimeoutException, Int
4343
consumer.start();
4444

4545
NSQProducer producer = new NSQProducer();
46-
producer.addAddress("localhost", 4150);
46+
producer.addAddress(Nsq.getNsqdHost(), 4150);
4747
producer.start();
4848
String msg = "test-one-message";
4949
producer.produce("test1", msg.getBytes());

‎src/test/java/com/github/brainlag/nsq/NSQProducerTest.java

Lines changed: 26 additions & 22 deletions
Original file line numberDiff line numberDiff line change
@@ -16,7 +16,11 @@
1616
import java.util.Date;
1717
import java.util.List;
1818
import java.util.Set;
19-
import java.util.concurrent.*;
19+
import java.util.concurrent.ExecutorService;
20+
import java.util.concurrent.LinkedBlockingQueue;
21+
import java.util.concurrent.ThreadPoolExecutor;
22+
import java.util.concurrent.TimeUnit;
23+
import java.util.concurrent.TimeoutException;
2024
import java.util.concurrent.atomic.AtomicInteger;
2125

2226
import static org.junit.Assert.assertEquals;
@@ -65,11 +69,11 @@ private NSQConfig getSslAndDeflateConfig() throws SSLException {
6569
public void testProduceOneMsgSnappy() throws NSQException, TimeoutException, InterruptedException {
6670
AtomicInteger counter = new AtomicInteger(0);
6771
NSQLookup lookup = new DefaultNSQLookup();
68-
lookup.addLookupAddress("localhost", 4161);
72+
lookup.addLookupAddress(Nsq.getNsqLookupdHost(), 4161);
6973

7074
NSQProducer producer = new NSQProducer();
7175
producer.setConfig(getSnappyConfig());
72-
producer.addAddress("localhost", 4150);
76+
producer.addAddress(Nsq.getNsqdHost(), 4150);
7377
producer.start();
7478
String msg = randomString();
7579
producer.produce("test3", msg.getBytes());
@@ -93,11 +97,11 @@ public void testProduceOneMsgDeflate() throws NSQException, TimeoutException, In
9397
System.setProperty("io.netty.noJdkZlibDecoder", "false");
9498
AtomicInteger counter = new AtomicInteger(0);
9599
NSQLookup lookup = new DefaultNSQLookup();
96-
lookup.addLookupAddress("localhost", 4161);
100+
lookup.addLookupAddress(Nsq.getNsqLookupdHost(), 4161);
97101

98102
NSQProducer producer = new NSQProducer();
99103
producer.setConfig(getDeflateConfig());
100-
producer.addAddress("localhost", 4150);
104+
producer.addAddress(Nsq.getNsqdHost(), 4150);
101105
producer.start();
102106
String msg = randomString();
103107
producer.produce("test3", msg.getBytes());
@@ -120,11 +124,11 @@ public void testProduceOneMsgDeflate() throws NSQException, TimeoutException, In
120124
public void testProduceOneMsgSsl() throws InterruptedException, NSQException, TimeoutException, SSLException {
121125
AtomicInteger counter = new AtomicInteger(0);
122126
NSQLookup lookup = new DefaultNSQLookup();
123-
lookup.addLookupAddress("localhost", 4161);
127+
lookup.addLookupAddress(Nsq.getNsqLookupdHost(), 4161);
124128

125129
NSQProducer producer = new NSQProducer();
126130
producer.setConfig(getSslConfig());
127-
producer.addAddress("localhost", 4150);
131+
producer.addAddress(Nsq.getNsqdHost(), 4150);
128132
producer.start();
129133
String msg = randomString();
130134
producer.produce("test3", msg.getBytes());
@@ -147,11 +151,11 @@ public void testProduceOneMsgSsl() throws InterruptedException, NSQException, Ti
147151
public void testProduceOneMsgSslAndSnappy() throws InterruptedException, NSQException, TimeoutException, SSLException {
148152
AtomicInteger counter = new AtomicInteger(0);
149153
NSQLookup lookup = new DefaultNSQLookup();
150-
lookup.addLookupAddress("localhost", 4161);
154+
lookup.addLookupAddress(Nsq.getNsqLookupdHost(), 4161);
151155

152156
NSQProducer producer = new NSQProducer();
153157
producer.setConfig(getSslAndSnappyConfig());
154-
producer.addAddress("localhost", 4150);
158+
producer.addAddress(Nsq.getNsqdHost(), 4150);
155159
producer.start();
156160
String msg = randomString();
157161
producer.produce("test3", msg.getBytes());
@@ -175,11 +179,11 @@ public void testProduceOneMsgSslAndDeflat() throws InterruptedException, NSQExce
175179
System.setProperty("io.netty.noJdkZlibDecoder", "false");
176180
AtomicInteger counter = new AtomicInteger(0);
177181
NSQLookup lookup = new DefaultNSQLookup();
178-
lookup.addLookupAddress("localhost", 4161);
182+
lookup.addLookupAddress(Nsq.getNsqLookupdHost(), 4161);
179183

180184
NSQProducer producer = new NSQProducer();
181185
producer.setConfig(getSslAndDeflateConfig());
182-
producer.addAddress("localhost", 4150);
186+
producer.addAddress(Nsq.getNsqdHost(), 4150);
183187
producer.start();
184188
String msg = randomString();
185189
producer.produce("test3", msg.getBytes());
@@ -203,7 +207,7 @@ public void testProduceOneMsgSslAndDeflat() throws InterruptedException, NSQExce
203207
public void testProduceMoreMsg() throws NSQException, TimeoutException, InterruptedException {
204208
AtomicInteger counter = new AtomicInteger(0);
205209
NSQLookup lookup = new DefaultNSQLookup();
206-
lookup.addLookupAddress("localhost", 4161);
210+
lookup.addLookupAddress(Nsq.getNsqLookupdHost(), 4161);
207211

208212
NSQConsumer consumer = new NSQConsumer(lookup, "test3", "testconsumer", (message) -> {
209213
LogManager.getLogger(this).info("Processing message: " + new String(message.getMessage()));
@@ -213,7 +217,7 @@ public void testProduceMoreMsg() throws NSQException, TimeoutException, Interrup
213217
consumer.start();
214218

215219
NSQProducer producer = new NSQProducer();
216-
producer.addAddress("localhost", 4150);
220+
producer.addAddress(Nsq.getNsqdHost(), 4150);
217221
producer.start();
218222
for (int i = 0; i < 1000; i++) {
219223
String msg = randomString();
@@ -232,7 +236,7 @@ public void testProduceMoreMsg() throws NSQException, TimeoutException, Interrup
232236
public void testParallelProducer() throws NSQException, TimeoutException, InterruptedException {
233237
AtomicInteger counter = new AtomicInteger(0);
234238
NSQLookup lookup = new DefaultNSQLookup();
235-
lookup.addLookupAddress("localhost", 4161);
239+
lookup.addLookupAddress(Nsq.getNsqLookupdHost(), 4161);
236240

237241
NSQConsumer consumer = new NSQConsumer(lookup, "test3", "testconsumer", (message) -> {
238242
LogManager.getLogger(this).info("Processing message: " + new String(message.getMessage()));
@@ -242,7 +246,7 @@ public void testParallelProducer() throws NSQException, TimeoutException, Interr
242246
consumer.start();
243247

244248
NSQProducer producer = new NSQProducer();
245-
producer.addAddress("localhost", 4150);
249+
producer.addAddress(Nsq.getNsqdHost(), 4150);
246250
producer.start();
247251
for (int n = 0; n < 5; n++) {
248252
new Thread(() -> {
@@ -268,7 +272,7 @@ public void testParallelProducer() throws NSQException, TimeoutException, Interr
268272
public void testMultiMessage() throws NSQException, TimeoutException, InterruptedException {
269273
AtomicInteger counter = new AtomicInteger(0);
270274
NSQLookup lookup = new DefaultNSQLookup();
271-
lookup.addLookupAddress("localhost", 4161);
275+
lookup.addLookupAddress(Nsq.getNsqLookupdHost(), 4161);
272276

273277
NSQConsumer consumer = new NSQConsumer(lookup, "test3", "testconsumer", (message) -> {
274278
LogManager.getLogger(this).info("Processing message: " + new String(message.getMessage()));
@@ -278,7 +282,7 @@ public void testMultiMessage() throws NSQException, TimeoutException, Interrupte
278282
consumer.start();
279283

280284
NSQProducer producer = new NSQProducer();
281-
producer.addAddress("localhost", 4150);
285+
producer.addAddress(Nsq.getNsqdHost(), 4150);
282286
producer.start();
283287
List<byte[]> messages = Lists.newArrayList();
284288
for (int i = 0; i < 50; i++) {
@@ -298,7 +302,7 @@ public void testMultiMessage() throws NSQException, TimeoutException, Interrupte
298302
public void testBackoff() throws InterruptedException, NSQException, TimeoutException {
299303
AtomicInteger counter = new AtomicInteger(0);
300304
NSQLookup lookup = new DefaultNSQLookup();
301-
lookup.addLookupAddress("localhost", 4161);
305+
lookup.addLookupAddress(Nsq.getNsqLookupdHost(), 4161);
302306

303307
NSQConsumer consumer = new NSQConsumer(lookup, "test3", "testconsumer", (message) -> {
304308
LogManager.getLogger(this).info("Processing message: " + new String(message.getMessage()));
@@ -313,7 +317,7 @@ public void testBackoff() throws InterruptedException, NSQException, TimeoutExce
313317
consumer.start();
314318

315319
NSQProducer producer = new NSQProducer();
316-
producer.addAddress("localhost", 4150);
320+
producer.addAddress(Nsq.getNsqdHost(), 4150);
317321
producer.start();
318322
for (int i = 0; i < 20; i++) {
319323
String msg = randomString();
@@ -332,7 +336,7 @@ public void testBackoff() throws InterruptedException, NSQException, TimeoutExce
332336
public void testScheduledCallback() throws NSQException, TimeoutException, InterruptedException {
333337
AtomicInteger counter = new AtomicInteger(0);
334338
NSQLookup lookup = new DefaultNSQLookup();
335-
lookup.addLookupAddress("localhost", 4161);
339+
lookup.addLookupAddress(Nsq.getNsqLookupdHost(), 4161);
336340

337341
NSQConsumer consumer = new NSQConsumer(lookup, "test3", "testconsumer", (message) -> {});
338342
consumer.scheduleRun(() -> counter.incrementAndGet(), 1000, 1000, TimeUnit.MILLISECONDS);
@@ -346,11 +350,11 @@ public void testScheduledCallback() throws NSQException, TimeoutException, Inter
346350
@Test
347351
public void testEphemeralTopic() throws InterruptedException, NSQException, TimeoutException {
348352
NSQLookup lookup = new DefaultNSQLookup();
349-
lookup.addLookupAddress("localhost", 4161);
353+
lookup.addLookupAddress(Nsq.getNsqLookupdHost(), 4161);
350354

351355
NSQProducer producer = new NSQProducer();
352356
producer.setConfig(getDeflateConfig());
353-
producer.addAddress("localhost", 4150);
357+
producer.addAddress(Nsq.getNsqdHost(), 4150);
354358
producer.start();
355359
String msg = randomString();
356360
producer.produce("testephem#ephemeral", msg.getBytes());

‎src/test/java/com/github/brainlag/nsq/Nsq.java

Lines changed: 21 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,21 @@
1+
package com.github.brainlag.nsq;
2+
3+
public class Nsq {
4+
5+
public static String getNsqdHost() {
6+
String hostName = System.getenv("NSQD_HOST");
7+
if (hostName == null) {
8+
hostName = "localhost";
9+
}
10+
return hostName;
11+
}
12+
13+
public static String getNsqLookupdHost() {
14+
String hostName = System.getenv("NSQLOOKUPD_HOST");
15+
if (hostName == null) {
16+
hostName = "localhost";
17+
}
18+
return hostName;
19+
}
20+
21+
}

0 commit comments

Comments
 (0)