Merge pull request #22 from klucar/ephemeral_topic

Adds URL encoding to DefaultNSQLookup to handle ephemeral topics

Author: brainlag

src/main/java/com/github/brainlag/nsq/lookup/DefaultNSQLookup.java

@@ -3,11 +3,13 @@
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.github.brainlag.nsq.ServerAddress;
+import com.google.common.base.Charsets;
 import com.google.common.collect.Sets;
 import org.apache.logging.log4j.LogManager;
 
 import java.io.IOException;
 import java.net.URL;
+import java.net.URLEncoder;
 import java.util.Set;
 public class DefaultNSQLookup implements NSQLookup {
     Set<String> addresses = Sets.newHashSet();
@@ -28,7 +30,8 @@ public Set<ServerAddress> lookup(String topic) {
         for (String addr : getLookupAddresses()) {
             try {
                 ObjectMapper mapper = new ObjectMapper();
-                JsonNode jsonNode = mapper.readTree(new URL(addr + "/lookup?topic=" + topic));
+                String topicEncoded = URLEncoder.encode(topic, Charsets.UTF_8.name());
+                JsonNode jsonNode = mapper.readTree(new URL(addr + "/lookup?topic=" + topicEncoded));
                 LogManager.getLogger(this).debug("Server connection information: " + jsonNode.toString());
                 JsonNode producers = jsonNode.get("data").get("producers");
                 for (JsonNode node : producers) {

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

@@ -15,6 +15,7 @@
 import java.io.File;
 import java.util.Date;
 import java.util.List;
+import java.util.Set;
 import java.util.concurrent.*;
 import java.util.concurrent.atomic.AtomicInteger;
 
@@ -342,6 +343,23 @@ public void testScheduledCallback() throws NSQException, TimeoutException, Inter
         consumer.shutdown();
     }
 
+    @Test
+    public void testEphemeralTopic() throws InterruptedException, NSQException, TimeoutException {
+        NSQLookup lookup = new DefaultNSQLookup();
+        lookup.addLookupAddress("localhost", 4161);
+
+        NSQProducer producer = new NSQProducer();
+        producer.setConfig(getDeflateConfig());
+        producer.addAddress("localhost", 4150);
+        producer.start();
+        String msg = randomString();
+        producer.produce("testephem#ephemeral", msg.getBytes());
+        producer.shutdown();
+
+        Set<ServerAddress> servers = lookup.lookup("testephem#ephemeral");
+        assertEquals("Could not find servers for ephemeral topic", 1, servers.size());
+    }
+
     public static ExecutorService newBackoffThreadExecutor() {
         return new ThreadPoolExecutor(1, 1,
                 0L, TimeUnit.MILLISECONDS,