/*
 * 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.artemis.tests.integration.routing;

import javax.jms.ConnectionFactory;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;

import org.apache.activemq.artemis.api.core.TransportConfiguration;
import org.apache.activemq.artemis.core.config.Configuration;
import org.apache.activemq.artemis.core.config.routing.ConnectionRouterConfiguration;
import org.apache.activemq.artemis.core.config.routing.CacheConfiguration;
import org.apache.activemq.artemis.core.config.routing.NamedPropertyConfiguration;
import org.apache.activemq.artemis.core.config.routing.PoolConfiguration;
import org.apache.activemq.artemis.core.remoting.impl.netty.TransportConstants;
import org.apache.activemq.artemis.core.server.routing.KeyType;
import org.apache.activemq.artemis.jms.client.ActiveMQConnectionFactory;
import org.apache.activemq.artemis.tests.integration.cluster.distribution.ClusterTestBase;
import org.apache.activemq.artemis.tests.util.ActiveMQTestBase;
import org.apache.qpid.jms.JmsConnectionFactory;

public class RoutingTestBase extends ClusterTestBase {
   protected static final String AMQP_PROTOCOL = "AMQP";
   protected static final String CORE_PROTOCOL = "CORE";
   protected static final String OPENWIRE_PROTOCOL = "OPENWIRE";

   protected static final String CLUSTER_POOL = "CLUSTER";
   protected static final String DISCOVERY_POOL = "DISCOVERY";
   protected static final String STATIC_POOL = "STATIC";

   protected static final String CONNECTION_ROUTER_NAME = "bb1";

   protected static final String DEFAULT_CONNECTOR_NAME = "DEFAULT";

   protected static final String GROUP_ADDRESS = ActiveMQTestBase.getUDPDiscoveryAddress();

   protected static final int GROUP_PORT = ActiveMQTestBase.getUDPDiscoveryPort();

   protected static final int MULTIPLE_TARGETS = 3;


   protected TransportConfiguration getDefaultServerAcceptor(final int node) {
      return getServer(node).getConfiguration().getAcceptorConfigurations().stream().findFirst().get();
   }

   protected TransportConfiguration getDefaultServerConnector(final int node) {
      Map<String, TransportConfiguration> connectorConfigurations = getServer(node).getConfiguration().getConnectorConfigurations();
      TransportConfiguration connector = connectorConfigurations.get(DEFAULT_CONNECTOR_NAME);
      return connector != null ? connector : connectorConfigurations.values().stream().findFirst().get();
   }

   protected TransportConfiguration setupDefaultServerConnector(final int node) {
      TransportConfiguration defaultServerConnector = getDefaultServerConnector(node);

      if (!defaultServerConnector.getName().equals(DEFAULT_CONNECTOR_NAME)) {
         defaultServerConnector = new TransportConfiguration(defaultServerConnector.getFactoryClassName(),
            defaultServerConnector.getParams(), DEFAULT_CONNECTOR_NAME, defaultServerConnector.getExtraParams());

         getServer(node).getConfiguration().getConnectorConfigurations().put(DEFAULT_CONNECTOR_NAME, defaultServerConnector);
      }

      return defaultServerConnector;
   }

   protected void setupRouterServerWithCluster(final int node, final KeyType keyType, final String policyName, final Map<String, String> properties, final boolean localTargetEnabled, final String localTargetFilter, final int quorumSize, String clusterConnection) {
      Configuration configuration = getServer(node).getConfiguration();
      ConnectionRouterConfiguration connectionRouterConfiguration = new ConnectionRouterConfiguration().setName(CONNECTION_ROUTER_NAME);

      setupDefaultServerConnector(node);

      connectionRouterConfiguration.setKeyType(keyType).setLocalTargetFilter(localTargetFilter)
         .setPoolConfiguration(new PoolConfiguration().setCheckPeriod(1000).setQuorumSize(quorumSize)
            .setLocalTargetEnabled(localTargetEnabled).setClusterConnection(clusterConnection))
         .setPolicyConfiguration(new NamedPropertyConfiguration().setName(policyName).setProperties(properties));

      configuration.setConnectionRouters(Collections.singletonList(connectionRouterConfiguration));

      TransportConfiguration acceptor = getDefaultServerAcceptor(node);
      acceptor.getParams().put("router", CONNECTION_ROUTER_NAME);
   }

   protected void setupRouterServerWithDiscovery(final int node, final KeyType keyType, final String policyName, final Map<String, String> properties, final boolean localTargetEnabled, final String localTargetFilter, final int quorumSize) {
      Configuration configuration = getServer(node).getConfiguration();
      ConnectionRouterConfiguration connectionRouterConfiguration = new ConnectionRouterConfiguration().setName(CONNECTION_ROUTER_NAME);

      setupDefaultServerConnector(node);

      connectionRouterConfiguration.setKeyType(keyType).setLocalTargetFilter(localTargetFilter)
         .setPoolConfiguration(new PoolConfiguration().setCheckPeriod(1000).setQuorumSize(quorumSize)
            .setLocalTargetEnabled(localTargetEnabled).setDiscoveryGroupName("dg1"))
         .setPolicyConfiguration(new NamedPropertyConfiguration().setName(policyName).setProperties(properties));

      configuration.setConnectionRouters(Collections.singletonList(connectionRouterConfiguration));

      TransportConfiguration acceptor = getDefaultServerAcceptor(node);
      acceptor.getParams().put("router", CONNECTION_ROUTER_NAME);
   }

   protected void setupRouterServerWithStaticConnectors(final int node, final KeyType keyType, final String policyName, final Map<String, String> properties, final boolean localTargetEnabled, final String localTargetFilter, final int quorumSize, final int... targetNodes) {
      Configuration configuration = getServer(node).getConfiguration();
      ConnectionRouterConfiguration connectionRouterConfiguration = new ConnectionRouterConfiguration().setName(CONNECTION_ROUTER_NAME);

      setupDefaultServerConnector(node);

      List<String> staticConnectors = new ArrayList<>();
      for (int targetNode : targetNodes) {
         TransportConfiguration connector = getDefaultServerConnector(targetNode);
         configuration.getConnectorConfigurations().put(connector.getName(), connector);
         staticConnectors.add(connector.getName());
      }

      connectionRouterConfiguration.setKeyType(keyType).setLocalTargetFilter(localTargetFilter)
         .setPoolConfiguration(new PoolConfiguration().setCheckPeriod(1000).setQuorumSize(quorumSize)
            .setLocalTargetEnabled(localTargetEnabled).setStaticConnectors(staticConnectors))
         .setPolicyConfiguration(new NamedPropertyConfiguration().setName(policyName).setProperties(properties));

      configuration.setConnectionRouters(Collections.singletonList(connectionRouterConfiguration));

      TransportConfiguration acceptor = getDefaultServerAcceptor(node);
      acceptor.getParams().put("router", CONNECTION_ROUTER_NAME);
   }


   protected void setupRouterServerWithLocalTarget(final int node, final KeyType keyType, final String targetKeyFilter, final String localTargetFilter) {

      Configuration configuration = getServer(node).getConfiguration();
      ConnectionRouterConfiguration connectionRouterConfiguration = new ConnectionRouterConfiguration().setName(CONNECTION_ROUTER_NAME);
      connectionRouterConfiguration.setKeyType(keyType).setLocalTargetFilter(localTargetFilter).setKeyFilter(targetKeyFilter);

      configuration.setConnectionRouters(Collections.singletonList(connectionRouterConfiguration));

      TransportConfiguration acceptor = getDefaultServerAcceptor(node);
      acceptor.getParams().put("router", CONNECTION_ROUTER_NAME);

   }

   protected void setupRouterLocalCache(final int node, boolean persisted, int timeout) {

      Configuration configuration = getServer(node).getConfiguration();
      ConnectionRouterConfiguration connectionRouterConfiguration = configuration.getConnectionRouters().stream()
         .filter(config -> config.getName().equals(CONNECTION_ROUTER_NAME)).findFirst().get();

      connectionRouterConfiguration.setCacheConfiguration(
         new CacheConfiguration().setPersisted(persisted).setTimeout(timeout));
   }

   protected ConnectionFactory createFactory(String protocol, boolean sslEnabled, String host, int port, String clientID, String user, String password) throws Exception {
      return createFactory(protocol, sslEnabled,  host, port, clientID, user, password, -1);
   }

   protected ConnectionFactory createFactory(String protocol, boolean sslEnabled, String host, int port, String clientID, String user, String password, int retries) throws Exception {
      switch (protocol) {
         case CORE_PROTOCOL: {
            StringBuilder urlBuilder = new StringBuilder();

            urlBuilder.append("tcp://");
            urlBuilder.append(host);
            urlBuilder.append(":");
            urlBuilder.append(port);
            urlBuilder.append("?ha=true&reconnectAttempts=10&initialConnectAttempts=" + retries);

            urlBuilder.append("&sniHost=");
            urlBuilder.append(host);

            if (clientID != null) {
               urlBuilder.append("&clientID=");
               urlBuilder.append(clientID);
            }

            if (sslEnabled) {
               urlBuilder.append("&");
               urlBuilder.append(TransportConstants.SSL_ENABLED_PROP_NAME);
               urlBuilder.append("=");
               urlBuilder.append(true);

               urlBuilder.append("&");
               urlBuilder.append(TransportConstants.TRUSTSTORE_PATH_PROP_NAME);
               urlBuilder.append("=");
               urlBuilder.append("server-ca-truststore.jks");

               urlBuilder.append("&");
               urlBuilder.append(TransportConstants.TRUSTSTORE_PASSWORD_PROP_NAME);
               urlBuilder.append("=");
               urlBuilder.append("securepass");
            }

            return new ActiveMQConnectionFactory(urlBuilder.toString(), user, password);
         }
         case AMQP_PROTOCOL: {
            StringBuilder urlBuilder = new StringBuilder();

            urlBuilder.append("failover:(");

            if (sslEnabled) {
               urlBuilder.append("amqps://");
               urlBuilder.append(host);
               urlBuilder.append(":");
               urlBuilder.append(port);

               urlBuilder.append("?transport.trustStoreLocation=");
               urlBuilder.append(getClass().getClassLoader().getResource("server-ca-truststore.jks").getFile());
               urlBuilder.append("&transport.trustStorePassword=securepass)");
            } else {
               urlBuilder.append("amqp://");
               urlBuilder.append(host);
               urlBuilder.append(":");
               urlBuilder.append(port);
               urlBuilder.append(")");
            }

            urlBuilder.append("?failover.startupMaxReconnectAttempts=" + retries);

            if (clientID != null) {
               urlBuilder.append("&jms.clientID=");
               urlBuilder.append(clientID);
            }

            return new JmsConnectionFactory(user, password, urlBuilder.toString());
         }
         case OPENWIRE_PROTOCOL: {
            StringBuilder urlBuilder = new StringBuilder();

            urlBuilder.append("failover:(");

            if (sslEnabled) {
               urlBuilder.append("ssl://");
               urlBuilder.append(host);
               urlBuilder.append(":");
               urlBuilder.append(port);
               urlBuilder.append(")");
            } else {
               urlBuilder.append("tcp://");
               urlBuilder.append(host);
               urlBuilder.append(":");
               urlBuilder.append(port);
               urlBuilder.append(")");
            }

            urlBuilder.append("?randomize=false&startupMaxReconnectAttempts=" + retries + "&maxReconnectAttempts=" + retries);

            if (clientID != null) {
               urlBuilder.append("&jms.clientID=");
               urlBuilder.append(clientID);
            }

            if (sslEnabled) {
               org.apache.activemq.ActiveMQSslConnectionFactory sslConnectionFactory = new org.apache.activemq.ActiveMQSslConnectionFactory(urlBuilder.toString());
               sslConnectionFactory.setUserName(user);
               sslConnectionFactory.setPassword(password);
               sslConnectionFactory.setTrustStore("server-ca-truststore.jks");
               sslConnectionFactory.setTrustStorePassword("securepass");
               return sslConnectionFactory;
            } else {
               return new org.apache.activemq.ActiveMQConnectionFactory(user, password, urlBuilder.toString());
            }
         }
         default:
            throw new IllegalStateException("Unexpected value: " + protocol);
      }
   }
}
