package com.github.brainlag.nsq.lookup;

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();

    @Override
    public void addLookupAddress(String addr, int port) {
        if (!addr.startsWith("http")) {
            addr = "http://" + addr;
        }
        addr = addr + ":" + port;
        this.addresses.add(addr);
    }

    @Override
    public Set<ServerAddress> lookup(String topic) {
        Set<ServerAddress> addresses = Sets.newHashSet();

        for (String addr : getLookupAddresses()) {
            try {
                ObjectMapper mapper = new ObjectMapper();
                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) {
                    String host = node.get("broadcast_address").asText();
                    ServerAddress address = new ServerAddress(host, node.get("tcp_port").asInt());
                    addresses.add(address);
                }
            } catch (IOException e) {
                LogManager.getLogger(this).warn("Unable to connect to address {}", addr);
                LogManager.getLogger(this).debug(e.getMessage());
            }
        }
        if (addresses.isEmpty()) {
            LogManager.getLogger(this).warn("Unable to connect to any NSQ Lookup servers, servers tried: " + this.addresses.toString());
        }
        return addresses;
    }

    public Set<String> getLookupAddresses() {
        return addresses;
    }
}

@@ -403,16 +403,17 @@
 il.Set;%0A
+%0A
 public c
@@ -1771,15 +1771,35 @@
 s %7B%7D
-%22, addr
+ for topic %7B%7D%22, addr, topic
 );%0A 
@@ -2027,19 +2027,33 @@
  tried: 
-%22 +
+%7B%7D on topic: %7B%7D%22,
  this.ad
@@ -2059,27 +2059,23 @@
 ddresses
-.toString()
+, topic
 );%0A     

add topic in debug logs when lookup fails