/*
 * Copyright 2016 Red Hat Inc.
 *
 * 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 enmasse.queue.scheduler;

import io.vertx.core.AbstractVerticle;
import io.vertx.core.json.JsonObject;
import io.vertx.proton.ProtonClient;
import io.vertx.proton.ProtonConnection;
import io.vertx.proton.ProtonReceiver;
import io.vertx.proton.ProtonServer;
import org.apache.qpid.proton.amqp.messaging.AmqpValue;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ExecutorService;

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

    private final SchedulerState schedulerState = new SchedulerState();
    private final BrokerFactory brokerFactory;
    private final ExecutorService executorService;
    private volatile ProtonServer server;

    private final int port;

    public QueueScheduler(ExecutorService executorService, BrokerFactory brokerFactory, int listenPort) {
        this.executorService = executorService;
        this.brokerFactory = brokerFactory;
        this.port = listenPort;
    }

    @Override
    public void start() {
        server = ProtonServer.create(vertx);
        server.connectHandler(connection -> {
            log.info("Got connection open!");
            connection.setContainer("queue-scheduler");
            connection.openHandler(conn -> {
                log.info("Connection opened from " + conn.result().getRemoteContainer());
                executorService.execute(() -> {
                    try {
                        schedulerState.brokerAdded(conn.result().getRemoteContainer(), brokerFactory.createBroker(connection).get());
                    } catch (Exception e) {
                        log.error("Error adding broker", e);
                    }
                });
            }).closeHandler(conn -> {
                executorService.execute(() -> schedulerState.brokerRemoved(conn.result().getRemoteContainer()));
                connection.close();
                connection.disconnect();
                log.info("Broker connection closed");
            }).disconnectHandler(protonConnection -> {
                connection.disconnect();
                log.info("Broker connection disconnected");
            }).open();
        });
        server.listen(port, event -> {
            if (event.succeeded()) {
                log.info("QueueScheduler is up and running");
            } else {
                log.error("Error starting queue scheduler", event.cause());
            }
        });
    }

    @Override
    public void stop() {
        log.info("Stopping server!");
        if (server != null) {
            server.close();
        }
    }

    @Override
    public void addressesChanged(Map<String, Set<String>> addressMap) {
        executorService.execute(() -> {
            try {
                schedulerState.addressesChanged(addressMap);
            } catch (Exception e) {
                log.error("Error handling address change: ", e);
            }
        });
    }

    public int getPort() {
        if (server == null) {
            return 0;
        } else {
            return server.actualPort();
        }
    }
}

@@ -688,23 +688,15 @@
 ore.
-json.JsonObject
+Handler
 ;%0Aim
@@ -713,26 +713,25 @@
 rtx.
-proton.ProtonClien
+core.net.NetSocke
 t;%0Ai
@@ -803,13 +803,11 @@
 oton
-Recei
+Ser
 ver;
@@ -826,35 +826,51 @@
 ertx.proton.
+sasl.
 ProtonS
-erve
+aslAuthenticato
 r;%0Aimport or
@@ -894,32 +894,24 @@
 ton.
-amqp.messaging.AmqpValue
+engine.Transport
 ;%0Aim
@@ -970,66 +970,8 @@
 y;%0A%0A
-import java.util.HashSet;%0Aimport java.util.LinkedHashMap;%0A
 impo
@@ -1891,349 +1891,367 @@
 ver.
-connectHandler(connection -%3E %7B%0A            log.info(%22Got connection open!%22);%0A            connection.setContainer(%22queue-scheduler%22);%0A            connection.openHandler(conn -%3E %7B%0A                log.info(%22Connection opened from %22 + conn.result().getRemoteContainer());%0A                executorService.execute(() -%3E %7B%0A                    try %7B
+saslAuthenticatorFactory(() -%3E new ProtonSaslAuthenticator() %7B%0A            @Override%0A            public void init(NetSocket socket, ProtonConnection protonConnection, Transport transport) %7B%0A            %7D%0A%0A            @Override%0A            public void process(Handler%3CBoolean%3E completionHandler) %7B%0A                completionHandler.handle(true);%0A            %7D%0A
 %0A   
@@ -2251,32 +2251,42 @@
  %7D%0A%0A            
+@Override%0A
             sche
@@ -2285,161 +2285,65 @@
     
-schedulerState.brokerAdded(conn.result().getRemoteContainer(), brokerFactory.createBroker(connection).get());%0A                    %7D catch (Exception e) %7B
+public boolean succeeded() %7B%0A                return true;
 %0A   
@@ -2347,32 +2347,34 @@
             
+%7D%0A
         
     log.erro
@@ -2365,88 +2365,111 @@
     
-    log.error(%22Error adding broker%22, e);%0A                    %7D%0A                %7D
+%7D);%0A        server.connectHandler(connection -%3E %7B%0A            connection.setContainer(%22queue-scheduler%22
 );%0A 
@@ -2479,18 +2479,26 @@
         
-%7D)
+connection
 .closeHa
@@ -2935,16 +2935,452 @@
 .open();
+%0A%0A            log.info(%22Connection opened from %22 + connection.getRemoteContainer());%0A                executorService.execute(() -%3E %7B%0A                    try %7B%0A                        schedulerState.brokerAdded(connection.getRemoteContainer(), brokerFactory.createBroker(connection).get());%0A                    %7D catch (Exception e) %7B%0A                        log.error(%22Error adding broker%22, e);%0A                    %7D%0A                %7D);
 %0A       

Disable sasl