package enmasse.queue.scheduler;

import enmasse.address.controller.admin.OpenShiftHelper;
import enmasse.address.controller.model.LabelKeys;
import io.fabric8.kubernetes.client.KubernetesClientException;
import io.fabric8.kubernetes.client.Watcher;
import io.fabric8.openshift.api.model.DeploymentConfig;
import io.fabric8.openshift.api.model.DeploymentConfigList;
import io.fabric8.openshift.client.OpenShiftClient;
import io.vertx.core.AbstractVerticle;
import io.vertx.proton.ProtonServer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * Acts as an arbiter deciding in which broker a queue should run.
 */
public class QueueScheduler extends AbstractVerticle implements Watcher<DeploymentConfig> {
    private static final Logger log = LoggerFactory.getLogger(QueueScheduler.class.getName());

    private final OpenShiftClient openShiftClient;
    private final QueueState queueState = new QueueState();
    private final ExecutorService executorService = Executors.newSingleThreadScheduledExecutor();

    private final int port;

    public QueueScheduler(OpenShiftClient openShiftClient, int port) {
        this.openShiftClient = openShiftClient;
        this.port = port;
    }

    @Override
    public void start() {
        ProtonServer server = ProtonServer.create(vertx);
        server.connectHandler(connection -> {
            connection.setContainer("queue-scheduler");
            connection.openHandler(conn -> {
                log.info("Connection opened from " + conn.result().getRemoteContainer());
                executorService.execute(() -> queueState.brokerAdded(new Artemis(connection)));
            }).closeHandler(conn -> {
                executorService.execute(() -> queueState.brokerRemoved(conn.result().getRemoteContainer()));
                connection.close();
                connection.disconnect();
                log.info("Connection closed");
            }).disconnectHandler(protonConnection -> {
                connection.disconnect();
                log.info("Disconnected");
            }).open();
        });
        server.listen(port, event -> {
            if (event.succeeded()) {
                log.info("QueueScheduler is up and running on port " + port);
            } else {
                log.error("Error starting queue scheduler", event.cause());
            }
        });

        vertx.executeBlocking(promise -> {
            try {
                DeploymentConfigList configs = openShiftClient.deploymentConfigs().withLabel("type", "address-config").list();
                for (DeploymentConfig config : configs.getItems()) {
                    eventReceived(Action.ADDED, config);
                }
                openShiftClient.deploymentConfigs().withResourceVersion(configs.getMetadata().getResourceVersion()).watch(this);
                promise.complete(configs);
            } catch (Exception e) {
                promise.fail(e);
            }

        }, result -> {
            if (!result.succeeded()) {
                log.warn("Error getting deployment configs", result.cause());
            }
        });
    }


    @Override
    public void eventReceived(Action action, DeploymentConfig resource) {
        switch (action) {
            case ADDED:
                log.info("Deployment config was added");
                addressesChanged(resource);
                break;
            case DELETED:
                addressesDeleted(resource);
                break;
            case ERROR:
                log.error("Error with action", action);
                break;
            case MODIFIED:
                addressesChanged(resource);
                break;
        }
    }

    private void addressesChanged(DeploymentConfig deploymentConfig) {
        String deploymentId = deploymentConfig.getMetadata().getName();
        Set<String> addresses = OpenShiftHelper.parseAddressAnnotation(deploymentConfig.getMetadata().getAnnotations().get(LabelKeys.ADDRESS_LIST));

        executorService.execute(() -> queueState.deploymentUpdated(deploymentId, addresses));
    }

    private void addressesDeleted(DeploymentConfig deploymentConfig) {
        executorService.execute(() -> queueState.deploymentDeleted(deploymentConfig.getMetadata().getName()));
    }

    @Override
    public void onClose(KubernetesClientException cause) {
        log.warn("Watcher closed", cause);
    }
}

@@ -46,375 +46,290 @@
 sse.
-address.controller.admin.OpenShiftHelper;%0Aimport enmasse.address.controller.model.LabelKeys;%0Aimport io.fabric8.kubernetes.client.KubernetesClientException;%0Aimport io.fabric8.kubernetes.client.Watcher;%0Aimport io.fabric8.openshift.api.model.DeploymentConfig;%0Aimport io.fabric8.openshift.api.model.DeploymentConfigList;%0Aimport io.fabric8.openshift.client.OpenShiftClient
+config.LabelKeys;%0Aimport io.fabric8.kubernetes.api.model.ConfigMap;%0Aimport io.fabric8.kubernetes.api.model.ConfigMapList;%0Aimport io.fabric8.kubernetes.client.KubernetesClient;%0Aimport io.fabric8.kubernetes.client.KubernetesClientException;%0Aimport io.fabric8.kubernetes.client.Watcher
 ;%0Aim
@@ -464,30 +464,8 @@
 y;%0A%0A
-import java.util.Set;%0A
 impo
@@ -692,32 +692,25 @@
 Watcher%3C
-Deployment
 Config
+Map
 %3E %7B%0A    
@@ -823,33 +823,35 @@
 nal 
-OpenShiftClient openShift
+KubernetesClient kubernetes
 Clie
@@ -1072,33 +1072,35 @@
 ler(
-OpenShiftClient openShift
+KubernetesClient kubernetes
 Clie
@@ -1132,35 +1132,37 @@
 his.
-openShiftClient = openShift
+kubernetesClient = kubernetes
 Clie
@@ -2428,32 +2428,25 @@
         
-Deployment
 Config
+Map
 List con
@@ -2456,40 +2456,34 @@
 s = 
-openShiftClient.deploymentC
+kubernetesClient.c
 onfig
+Map
 s().
@@ -2547,32 +2547,25 @@
    for (
-Deployment
 Config
+Map
  config 
@@ -2683,40 +2683,34 @@
     
-openShiftClient.deploymentC
+kubernetesClient.c
 onfig
+Map
 s().
@@ -3146,32 +3146,25 @@
 action, 
-Deployment
 Config
+Map
  resourc
@@ -3687,41 +3687,27 @@
 ged(
-DeploymentConfig deploymentC
+ConfigMap c
 onfig
+Map
 ) %7B%0A
@@ -3725,152 +3725,27 @@
 ing 
-deploymentId = deploymentConfig.getMetadata().getName();%0A        Set%3CString%3E addresses = OpenShiftHelper.parseAddressAnnotation(deploymentConfig
+groupId = configMap
 .get
@@ -3762,18 +3762,13 @@
 .get
-Annotation
+Label
 s().
@@ -3785,21 +3785,16 @@
 eys.
-ADDRESS_LIST)
+GROUP_ID
 );%0A%0A
@@ -3864,31 +3864,45 @@
 ted(
-deploymentId, addresses
+groupId, configMap.getData().keySet()
 ));%0A
@@ -3946,44 +3946,116 @@
 ted(
-DeploymentConfig deploymentConfig) %7B
+ConfigMap configMap) %7B%0A        String groupId = configMap.getMetadata().getLabels().get(LabelKeys.GROUP_ID);
 %0A   
@@ -4122,48 +4122,15 @@
 ted(
-deploymentConfig.getMetadata().getName()
+groupId
 ));%0A

Use config maps for looking up address config