/**
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You 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 org.apache.activemq.bugs;

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.broker.*;
import org.apache.activemq.command.*;
import org.apache.activemq.filter.DestinationMapEntry;
import org.apache.activemq.security.*;
import org.apache.activemq.transport.InactivityIOException;
import org.apache.activemq.transport.Transport;
import org.apache.activemq.transport.TransportFactory;
import org.apache.activemq.transport.TransportListener;
import org.apache.activemq.transport.failover.FailoverTransport;
import org.apache.activemq.util.Wait;
import org.junit.After;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Test;

import javax.jms.Connection;
import javax.jms.*;
import java.io.IOException;
import java.net.Socket;
import java.net.URI;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;

public class TlsSessionResumeFailoverReconnectTest {

    private BrokerService brokerService;
    private String sslUrl, nioSslUrl;

    public static final String KEYSTORE_TYPE = "jks";
    public static final String PASSWORD = "password";
    public static final String SERVER_KEYSTORE = "src/test/resources/org/apache/activemq/security/broker1.ks";
    public static final String TRUST_KEYSTORE = "src/test/resources/org/apache/activemq/security/broker1.ks";


    @BeforeClass
    public static void before() throws Exception {
        System.setProperty("javax.net.ssl.trustStore", TRUST_KEYSTORE);
        System.setProperty("javax.net.ssl.trustStorePassword", PASSWORD);
        System.setProperty("javax.net.ssl.trustStoreType", KEYSTORE_TYPE);
        System.setProperty("javax.net.ssl.keyStore", SERVER_KEYSTORE);
        System.setProperty("javax.net.ssl.keyStoreType", KEYSTORE_TYPE);
        System.setProperty("javax.net.ssl.keyStorePassword", PASSWORD);
        // Choose a value that's informative: ssl,handshake,data,trustmanager or all
        //System.setProperty("javax.net.debug", "handshake");
    }


    public BrokerPlugin configureAuthentication() throws Exception {
        List<AuthenticationUser> users = new ArrayList<>();
        users.add(new AuthenticationUser("publisher", "123", "publisher"));
        users.add(new AuthenticationUser("subscriber", "123", "subscriber"));
        users.add(new AuthenticationUser("admin", "123", "publisher,subscriber"));
        users.add(new AuthenticationUser("noConnAdvisoryPerm", "123", "noConnAdvisoryPermGroup"));


        SimpleAuthenticationPlugin authenticationPlugin = new SimpleAuthenticationPlugin(users);

        return authenticationPlugin;
    }

    public BrokerPlugin configureAuthorization() throws Exception {

        @SuppressWarnings("rawtypes")
        List<DestinationMapEntry> authorizationEntries = new ArrayList<>();

        AuthorizationEntry entry = new AuthorizationEntry();
        entry.setTopic("dcu.>");
        entry.setRead("subscriber");
        entry.setWrite("publisher");
        entry.setAdmin("publisher,subscriber");
        authorizationEntries.add(entry);

        entry = new AuthorizationEntry();
        entry.setTopic("ActiveMQ.Advisory.>");
        entry.setRead("publisher,subscriber");
        entry.setWrite("publisher,subscriber,noConnAdvisoryPermGroup");
        entry.setAdmin("publisher,subscriber,noConnAdvisoryPermGroup");
        authorizationEntries.add(entry);

        entry = new AuthorizationEntry();
        entry.setTopic("ActiveMQ.Advisory.Connection.>");
        entry.setRead("noConnAdvisoryPermGroup,publisher,subscriber");
        entry.setWrite("noConnAdvisoryPermGroup,publisher,subscriber");
        entry.setAdmin("noConnAdvisoryPermGroup,publisher,subscriber");
        authorizationEntries.add(entry);

        entry = new AuthorizationEntry();
        entry.setQueue("Q.*");
        entry.setRead("noConnAdvisoryPermGroup");
        entry.setWrite("noConnAdvisoryPermGroup");
        entry.setAdmin("noConnAdvisoryPermGroup");
        authorizationEntries.add(entry);

        DefaultAuthorizationMap authorizationMap = new DefaultAuthorizationMap(authorizationEntries);
        AuthorizationPlugin authorizationPlugin = new AuthorizationPlugin(authorizationMap);

        return authorizationPlugin;
    }

    @Before
    public void setup() throws Exception {

        brokerService = new BrokerService();
        brokerService.setPersistent(false);

        TransportConnector transportConnectorSsl = new TransportConnector();
        transportConnectorSsl.setUri(new URI("ssl://localhost:0?transport.useInactivityMonitor=false"));
        transportConnectorSsl.setName("ssl");

        TransportConnector transportConnectorNioSsl = new TransportConnector();
        transportConnectorNioSsl.setUri(new URI("nio+ssl://localhost:0?transport.useInactivityMonitor=false"));
        transportConnectorNioSsl.setName("nio+ssl");

        brokerService.addConnector(transportConnectorSsl);
        brokerService.addConnector(transportConnectorNioSsl);

        ArrayList<BrokerPlugin> plugins = new ArrayList<>();
        plugins.add(configureAuthentication());
        plugins.add(configureAuthorization());
        BrokerPlugin[] array = new BrokerPlugin[plugins.size()];
        brokerService.setPlugins(plugins.toArray(array));

        brokerService.start();
        brokerService.waitUntilStarted();

        sslUrl = transportConnectorSsl.getPublishableConnectString();
        nioSslUrl = transportConnectorNioSsl.getPublishableConnectString();

    }

    @After
    public void shutdown() throws Exception {
        if (brokerService != null) {
            brokerService.stop();
            brokerService.waitUntilStopped();
            brokerService = null;
        }
    }

    @Test(timeout=3000000)
    public void testFailoverReconnectAuthTLSResumeNio() throws Exception {
        doTestReconnect(true);
    }

    @Test(timeout=3000000)
    public void testFailoverReconnectAuthTLSResume() throws Exception {
        doTestReconnect(false);
    }

    public void doTestReconnect(final boolean useNio) throws Exception {

        URI uri = new URI(useNio ? nioSslUrl : sslUrl);
        ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("failover:(ssl://localhost:" + uri.getPort() + "?wireFormat.maxInactivityDuration=1)");
        factory.setWatchTopicAdvisories(false);

        final int numConnections = 60;

        for (int repeats = 0; repeats < 1; repeats++) {

            ArrayList<ActiveMQConnection> connections = new ArrayList<>(numConnections);
            for (int i = 0; i < numConnections; i++) {
                ActiveMQConnection connection1 = (ActiveMQConnection) factory.createConnection("noConnAdvisoryPerm", "123");
                connection1.setClientID("CID" + i);
                connection1.start();


                for (int j = 0; j < 1; j++) {
                    Session session1 = connection1.createSession(false, Session.AUTO_ACKNOWLEDGE);
                    Queue queue = session1.createQueue("Q." + j);
                    for (int k = 0; k < 1; k++) {
                        MessageConsumer consumer = session1.createConsumer(queue);
                    }
                }
                connections.add(connection1);
            }

            final CountDownLatch caught = new CountDownLatch(numConnections);
            final CountDownLatch resumed = new CountDownLatch(numConnections);

            for (ActiveMQConnection c : connections) {
                final FailoverTransport failoverTransport = c.getTransport().narrow(FailoverTransport.class);
                final TransportListener delegate = failoverTransport.getTransportListener();
                failoverTransport.setTransportListener(new TransportListener() {
                    @Override
                    public void onCommand(Object command) {
                        delegate.onCommand(command);
                    }

                    @Override
                    public void onException(IOException error) {
                        delegate.onException(error);
                    }

                    @Override
                    public void transportInterupted() {
                        caught.countDown();
                        delegate.transportInterupted();
                    }

                    @Override
                    public void transportResumed() {
                        resumed.countDown();
                        delegate.transportResumed();
                    }
                });

                assertTrue(Wait.waitFor(new Wait.Condition() {
                    @Override
                    public boolean isSatisified() throws Exception {
                        return failoverTransport.isConnected();
                    }
                }));

            }


            ExecutorService executorService = Executors.newSingleThreadExecutor();
            executorService.submit(new Runnable() {
                @Override
                public void run() {
                    for (TransportConnection transportConnection : brokerService.getTransportConnectorByName(useNio ? "nio+ssl" : "ssl").getConnections()) {
                        transportConnection.serviceException(new InactivityIOException("BOOM"));
                    }
                }
            });
            assertTrue(caught.await(10, TimeUnit.SECONDS));
            assertTrue(resumed.await(20, TimeUnit.SECONDS));

            for (ActiveMQConnection c : connections) {
                FailoverTransport failoverTransport = c.getTransport().narrow(FailoverTransport.class);
                assertTrue("Connected ok: " + c, failoverTransport.isConnected());
            }

            for (ActiveMQConnection c : connections) {
                c.close();
            }
        }
    }
}
