/*
 * Copyright (c) 2015, Christoph Engelbert (aka noctarius) and
 * contributors. All rights reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.noctarius.hazelcast.kubernetes;

import com.hazelcast.logging.ILogger;
import com.hazelcast.logging.Logger;
import com.hazelcast.nio.Address;
import com.hazelcast.spi.discovery.DiscoveryNode;
import com.hazelcast.spi.discovery.SimpleDiscoveryNode;
import org.xbill.DNS.Lookup;
import org.xbill.DNS.Record;
import org.xbill.DNS.SRVRecord;
import org.xbill.DNS.TextParseException;
import org.xbill.DNS.Type;

import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;

final class DnsEndpointResolver extends HazelcastKubernetesDiscoveryStrategy.EndpointResolver {

    private static final ILogger LOGGER = Logger.getLogger(DnsEndpointResolver.class);

    private final String serviceDns;

    public DnsEndpointResolver(ILogger logger, String serviceDns) {
        super(logger);
        this.serviceDns = serviceDns;
    }

    List<DiscoveryNode> resolve() {
        try {
            Lookup lookup = new Lookup(serviceDns, Type.SRV);
            Record[] records = lookup.run();

            if (lookup.getResult() != Lookup.SUCCESSFUL) {
                LOGGER.warning("DNS lookup for serviceDns '" + serviceDns + "' failed");
                return Collections.emptyList();
            }

            List<DiscoveryNode> discoveredNodes = new ArrayList<DiscoveryNode>();

            if (records.length > 0) {
                // Get only the first record, because all of them have the same name
                // Example:
                // nslookup u219692-hazelcast.u219692-hazelcast.svc.cluster.local 172.30.0.1
                //      Server:         172.30.0.1
                //      Address:        172.30.0.1#53
                //
                //      Name:   u219692-hazelcast.u219692-hazelcast.svc.cluster.local
                //      Address: 10.1.2.8
                //      Name:   u219692-hazelcast.u219692-hazelcast.svc.cluster.local
                //      Address: 10.1.5.28
                //      Name:   u219692-hazelcast.u219692-hazelcast.svc.cluster.local
                //      Address: 10.1.9.33
                SRVRecord srv = (SRVRecord) records[0];
                InetAddress[] inetAddress = getAllAdresses(srv);
                int port = srv.getPort();

                for (InetAddress i : inetAddress) {
                    Address address = new Address(i, port);

                    if (LOGGER.isFinestEnabled()) {
                        LOGGER.finest("Found node ip-address is: " + address);
                    }
                    discoveredNodes.add(new SimpleDiscoveryNode(address));
                }
            } else {
                LOGGER.warning("Could not find any service for serviceDns '" + serviceDns + "' failed");
                return Collections.emptyList();
            }

            return discoveredNodes;

        } catch (TextParseException e) {
            throw new RuntimeException("Could not resolve services via DNS", e);
        } catch (UnknownHostException e) {
            throw new RuntimeException("Could not resolve services via DNS", e);
        }
    }

    private InetAddress[] getAllAdresses(SRVRecord srv) throws UnknownHostException {
        try {
            return org.xbill.DNS.Address.getAllByName(srv.getTarget().canonicalize().toString(true));
        } catch (UnknownHostException e) {
            LOGGER.severe("Parsing DNS records failed", e);
            throw e;
        }
    }
}

@@ -700,16 +700,59 @@
 netes;%0A%0A
+import com.hazelcast.config.NetworkConfig;%0A
 import c
@@ -983,143 +983,9 @@
 DNS.
-Lookup;%0Aimport org.xbill.DNS.Record;%0Aimport org.xbill.DNS.SRVRecord;%0Aimport org.xbill.DNS.TextParseException;%0Aimport org.xbill.DNS.Type
+*
 ;%0A%0Ai
@@ -2845,16 +2845,33 @@
  port = 
+getHazelcastPort(
 srv.getP
@@ -2875,16 +2875,17 @@
 etPort()
+)
 ;%0A%0A     
@@ -3723,24 +3723,150 @@
    %7D%0A    %7D%0A%0A
+    private int getHazelcastPort(int port)%7B%0A        if(port%3E0) return port;%0A        return NetworkConfig.DEFAULT_PORT;%0A    %7D%0A%0A
     private 

check for portnumber > 0 or return Hazelcast default port