/*
 * 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.cluster.distribution;

import java.io.EOFException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;

import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
import org.apache.activemq.artemis.api.core.ActiveMQBuffers;
import org.apache.activemq.artemis.api.core.BroadcastEndpoint;
import org.apache.activemq.artemis.api.core.BroadcastEndpointFactory;
import org.apache.activemq.artemis.api.core.SimpleString;
import org.apache.activemq.artemis.api.core.TransportConfiguration;
import org.apache.activemq.artemis.core.server.cluster.impl.MessageLoadBalancingType;
import org.apache.activemq.artemis.logs.AssertionLoggerHandler;
import org.apache.activemq.artemis.tests.util.ActiveMQTestBase;
import org.apache.activemq.artemis.tests.util.Wait;
import org.junit.jupiter.api.Test;

public class SymmetricClusterWithDiscoveryTest extends SymmetricClusterTest {

   protected final String groupAddress = ActiveMQTestBase.getUDPDiscoveryAddress();

   protected final int groupPort = ActiveMQTestBase.getUDPDiscoveryPort();

   @Override
   protected boolean isNetty() {
      return false;
   }

   @Override
   @Test
   public void testStartStopServers() throws Exception {
      // When using discovery starting and stopping it too fast could have a race condition with UDP
      doTestStartStopServers(false);
   }

   @Override
   @Test
   public void testStartStopServersWithPartition() throws Exception {
      // When using discovery starting and stopping it too fast could have a race condition with UDP
      doTestStartStopServers(true);
   }

   @Override
   protected void setupCluster() throws Exception {
      setupCluster(MessageLoadBalancingType.ON_DEMAND);
   }

   @Override
   protected void setupCluster(final MessageLoadBalancingType messageLoadBalancingType) throws Exception {
      setupDiscoveryClusterConnection("cluster0", 0, "dg1", "queues", messageLoadBalancingType, 1, isNetty());

      setupDiscoveryClusterConnection("cluster1", 1, "dg1", "queues", messageLoadBalancingType, 1, isNetty());

      setupDiscoveryClusterConnection("cluster2", 2, "dg1", "queues", messageLoadBalancingType, 1, isNetty());

      setupDiscoveryClusterConnection("cluster3", 3, "dg1", "queues", messageLoadBalancingType, 1, isNetty());

      setupDiscoveryClusterConnection("cluster4", 4, "dg1", "queues", messageLoadBalancingType, 1, isNetty());
   }

   @Override
   protected void setupServers() throws Exception {
      setupPrimaryServerWithDiscovery(0, groupAddress, groupPort, isFileStorage(), isNetty(), false);
      setupPrimaryServerWithDiscovery(1, groupAddress, groupPort, isFileStorage(), isNetty(), false);
      setupPrimaryServerWithDiscovery(2, groupAddress, groupPort, isFileStorage(), isNetty(), false);
      setupPrimaryServerWithDiscovery(3, groupAddress, groupPort, isFileStorage(), isNetty(), false);
      setupPrimaryServerWithDiscovery(4, groupAddress, groupPort, isFileStorage(), isNetty(), false);
   }

   /*
    * This is like testStopStartServers but we make sure we pause longer than discovery group timeout
    * before restarting (5 seconds)
    */
   @Test
   public void testStartStopServersWithPauseBeforeRestarting() throws Exception {
      doTestStartStopServers(false);
   }

   @Test
   public void testStartStopServersWithWrongConnectorConfigurations() throws Exception {
      try (AssertionLoggerHandler loggerHandler = new AssertionLoggerHandler()) {
         setupCluster();

         for (int node = 0; node < 5; node++) {
            final int serverNode = node;

            // Set wrong connector configurations
            Map<String, TransportConfiguration> wrongConnectorConfigurations = new HashMap<>();
            getServer(node).getConfiguration().getConnectorConfigurations().forEach((key, transportConfiguration) -> {
               TransportConfiguration wrongtransportConfiguration = new TransportConfiguration(
                       transportConfiguration.getFactoryClassName(),
                       new HashMap<>(transportConfiguration.getParams()),
                       transportConfiguration.getName(),
                       new HashMap<>(transportConfiguration.getExtraParams()));
               if (isNetty()) {
                  wrongtransportConfiguration.getParams().put("port", org.apache.activemq.artemis.core.remoting.impl.netty.TransportConstants.DEFAULT_PORT + serverNode + 10);
               } else {
                  wrongtransportConfiguration.getParams().put("serverId", String.valueOf(serverNode + 10));
               }
               wrongConnectorConfigurations.put(key, wrongtransportConfiguration);
            });
            getServer(node).getConfiguration().setConnectorConfigurations(wrongConnectorConfigurations);

            // Reduce the discovery stopping timeout to speed up the test
            getServer(node).getConfiguration().getDiscoveryGroupConfigurations().forEach((s, discoveryGroupConfiguration) -> discoveryGroupConfiguration.setStoppingTimeout(1));

            // Reduce the topology scanner attempts to speed up the test
            getServer(node).getConfiguration().getClusterConfigurations().forEach(
                    clusterConnectionConfiguration -> clusterConnectionConfiguration.setTopologyScannerAttempts(1));
         }

         startServers();

         validateTopologySize(1, 0, 1, 2, 3, 4);

         Wait.assertTrue(() -> loggerHandler.findText("AMQ224144"));

         stopServers();
      }
   }

   @Override
   protected void setupProxy(int nodeId) {
      getServer(nodeId).getConfiguration().getDiscoveryGroupConfigurations().get("dg1").
          setBroadcastEndpointFactory(new ProxyBroadcastEndpointFactory(nodeId, getServer(nodeId).
          getConfiguration().getDiscoveryGroupConfigurations().get("dg1").getBroadcastEndpointFactory()));
   }

   @Override
   protected void enablePartition() {
      super.enablePartition();

      ProxyBroadcastEndpointFactory.interceptor = (nodeId, data) -> {
         ActiveMQBuffer buffer = ActiveMQBuffers.wrappedBuffer(data);
         SimpleString originatingNodeID = SimpleString.of(buffer.readString());

         for (int i = 0; i < servers.length; i++) {
            if (servers[i] != null && originatingNodeID.equals(servers[i].getNodeID())) {
               int partitionId = (i % 5) / 3;
               int targetPartitionId = (nodeId % 5) / 3;

               return partitionId == targetPartitionId;
            }
         }

         return false;
      };
   }

   @Override
   protected void disablePartition() {
      super.disablePartition();

      ProxyBroadcastEndpointFactory.interceptor = null;
   }

   public interface ProxyBroadcastEndpointInterceptor {
      boolean allowBroadcast(int nodeId, byte[] data);
   }

   public class ProxyBroadcastEndpointFactory implements BroadcastEndpointFactory {

      public static volatile ProxyBroadcastEndpointInterceptor interceptor;

      private int nodeId;
      private final BroadcastEndpointFactory rawBroadcastEndpointFactory;

      public ProxyBroadcastEndpointFactory(int nodeId, BroadcastEndpointFactory rawBroadcastEndpointFactory) {
         this.nodeId = nodeId;
         this.rawBroadcastEndpointFactory = rawBroadcastEndpointFactory;
      }

      @Override
      public BroadcastEndpoint createBroadcastEndpoint() throws Exception {
         return new ProxyBroadcastEndpoint(nodeId, rawBroadcastEndpointFactory.createBroadcastEndpoint());
      }

      private class ProxyBroadcastEndpoint implements BroadcastEndpoint {

         private int nodeId;
         private volatile boolean open;
         private final BroadcastEndpoint rawBroadcastEndpoint;

         ProxyBroadcastEndpoint(int nodeId, BroadcastEndpoint rawBroadcastEndpoint) {
            this.nodeId = nodeId;
            this.rawBroadcastEndpoint = rawBroadcastEndpoint;
         }

         @Override
         public void openClient() throws Exception {
            open = true;
            rawBroadcastEndpoint.openClient();
         }

         @Override
         public void openBroadcaster() throws Exception {
            open = true;
            rawBroadcastEndpoint.openBroadcaster();
         }

         @Override
         public void close(boolean isBroadcast) throws Exception {
            open = false;
            rawBroadcastEndpoint.close(isBroadcast);
         }

         @Override
         public void broadcast(byte[] data) throws Exception {
            rawBroadcastEndpoint.broadcast(data);
         }

         @Override
         public byte[] receiveBroadcast() throws Exception {
            return receiveBroadcast(Long.MAX_VALUE, TimeUnit.DAYS);
         }

         @Override
         public byte[] receiveBroadcast(long time, TimeUnit unit) throws Exception {
            while (open) {
               byte[] data = rawBroadcastEndpoint.receiveBroadcast(time, unit);

               ProxyBroadcastEndpointInterceptor interceptor = ProxyBroadcastEndpointFactory.interceptor;
               if (interceptor == null || interceptor.allowBroadcast(nodeId, data)) {
                  return data;
               }
            }
            throw new EOFException();
         }
      }
   }
}
