/*
 * 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.isolated.client;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.Session;
import java.lang.invoke.MethodHandles;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.activemq.artemis.api.core.ActiveMQException;
import org.apache.activemq.artemis.api.core.QueueConfiguration;
import org.apache.activemq.artemis.api.core.RoutingType;
import org.apache.activemq.artemis.api.core.SimpleString;
import org.apache.activemq.artemis.core.persistence.OperationContext;
import org.apache.activemq.artemis.core.server.ActiveMQServer;
import org.apache.activemq.artemis.core.server.Queue;
import org.apache.activemq.artemis.core.server.ServerConsumer;
import org.apache.activemq.artemis.core.server.plugin.ActiveMQServerConsumerPlugin;
import org.apache.activemq.artemis.core.server.plugin.ActiveMQServerSessionPlugin;
import org.apache.activemq.artemis.logs.AssertionLoggerHandler;
import org.apache.activemq.artemis.spi.core.protocol.RemotingConnection;
import org.apache.activemq.artemis.spi.core.protocol.SessionCallback;
import org.apache.activemq.artemis.tests.util.ActiveMQTestBase;
import org.apache.activemq.artemis.tests.util.CFUtil;
import org.apache.activemq.artemis.utils.RandomUtil;
import org.apache.activemq.artemis.utils.ReusableLatch;
import org.apache.activemq.artemis.utils.ThreadDumpUtil;
import org.apache.activemq.artemis.utils.Wait;
import org.apache.qpid.protonj2.test.driver.ProtonTestClient;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.Timeout;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.junit.jupiter.api.Assertions.fail;

public class ConnectionDroppedTest extends ActiveMQTestBase {

   private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());

   public ConnectionDroppedTest() {
      // this is the reason why I'm putting this test on the "isolated" package.
      disableCheckThread();
   }

   @Test
   @Timeout(20)
   public void testConsumerDroppedWithProtonTestClient() throws Exception {
      int NUMBER_OF_CONNECTIONS = 100;
      ActiveMQServer server = createServer(true, createDefaultConfig(true));
      server.start();
      Queue serverQueue = server.createQueue(QueueConfiguration.of("test-queue").setRoutingType(RoutingType.ANYCAST).setAddress("test-queue").setAutoCreated(false));

      ExecutorService executorService = Executors.newFixedThreadPool(NUMBER_OF_CONNECTIONS);
      runAfter(executorService::shutdownNow);

      CountDownLatch done = new CountDownLatch(NUMBER_OF_CONNECTIONS);
      AtomicInteger errors = new AtomicInteger(0);

      for (int i = 0; i < NUMBER_OF_CONNECTIONS; i++) {
         executorService.execute(() -> {
            try (ProtonTestClient peer = new ProtonTestClient()) {
               peer.queueClientSaslAnonymousConnect();
               peer.remoteOpen().queue();
               peer.expectOpen();
               peer.remoteBegin().queue();
               peer.expectBegin();
               peer.remoteAttach().ofReceiver().withName(RandomUtil.randomString()).withSenderSettleModeUnsettled().withReceivervSettlesFirst().withTarget().also().withSource().withAddress("test-queue").withExpiryPolicyOnLinkDetach().withDurabilityOfNone().withCapabilities("queue").withOutcomes("amqp:accepted:list", "amqp:rejected:list").also().queue();
               peer.dropAfterLastHandler(1000); // This closes the netty connection after the attach is written
               peer.connect("localhost", 61616);

               // Waits for all the commands to fire and the drop action to be run.
               peer.waitForScriptToComplete();
            } catch (Throwable e) {
               errors.incrementAndGet();
               logger.warn(e.getMessage(), e);
            } finally {
               done.countDown();
            }
         });
      }

      assertTrue(done.await(10, TimeUnit.SECONDS));

      assertEquals(0, errors.get());

      Wait.assertEquals(0, () -> serverQueue.getConsumers().size(), 5000, 100);
   }

   @Test
   @Timeout(20)
   public void testRegularClose() throws Exception {
      int NUMBER_OF_CONNECTIONS = 100;
      int REPEATS = 10;
      ActiveMQServer server = createServer(true, createDefaultConfig(true));
      server.start();
      Queue serverQueue = server.createQueue(QueueConfiguration.of("test-queue").setRoutingType(RoutingType.ANYCAST).setAddress("test-queue").setAutoCreated(false));

      ExecutorService executorService = Executors.newFixedThreadPool(NUMBER_OF_CONNECTIONS);
      runAfter(executorService::shutdownNow);

      CountDownLatch done = new CountDownLatch(NUMBER_OF_CONNECTIONS);
      AtomicInteger errors = new AtomicInteger(0);
      AssertionLoggerHandler loggerHandler = new AssertionLoggerHandler();
      runAfter(loggerHandler::stop);

      for (int i = 0; i < NUMBER_OF_CONNECTIONS; i++) {
         executorService.execute(() -> {
            for (int r = 0; r < REPEATS; r++) {
               try (ProtonTestClient peer = new ProtonTestClient()) {
                  peer.queueClientSaslAnonymousConnect();
                  peer.remoteOpen().queue();
                  peer.expectOpen();
                  peer.remoteBegin().queue();
                  peer.expectBegin();
                  peer.remoteAttach().ofReceiver().withName(RandomUtil.randomString()).withSenderSettleModeUnsettled().withReceivervSettlesFirst().withTarget().also().withSource().withAddress("test-queue").withExpiryPolicyOnLinkDetach().withDurabilityOfNone().withCapabilities("queue").withOutcomes("amqp:accepted:list", "amqp:rejected:list").also().queue();
                  peer.expectAttach();
                  peer.remoteClose().queue();
                  peer.expectClose();

                  peer.connect("localhost", 61616);

                  peer.waitForScriptToComplete();
               } catch (Throwable e) {
                  errors.incrementAndGet();
                  logger.warn(e.getMessage(), e);
                  break;
               }
            }
            done.countDown();
         });
      }

      assertTrue(done.await(10, TimeUnit.SECONDS));

      assertEquals(0, errors.get());

      assertFalse(loggerHandler.findText("AMQ212037"));

      assertFalse(loggerHandler.findText("Connection failure"));
      assertFalse(loggerHandler.findText("REMOTE_DISCONNECT"));
      assertFalse(loggerHandler.findText("AMQ222061"));
      assertFalse(loggerHandler.findText("AMQ222107"));

      Wait.assertEquals(0, () -> serverQueue.getConsumers().size(), 5000, 100);
      Wait.assertEquals(0, server::getConnectionCount, 5000);

   }

   @Test
   public void testConsumerDroppedAMQP() throws Throwable {
      testConsumerDroppedWithRegularClient("AMQP");

   }

   @Test
   public void testConsumerDroppedCORE() throws Throwable {
      testConsumerDroppedWithRegularClient("CORE");
   }

   @Test
   public void testConsumerDroppedOpenWire() throws Throwable {
      testConsumerDroppedWithRegularClient("OPENWIRE");
   }

   public void testConsumerDroppedWithRegularClient(final String protocol) throws Throwable {
      int NUMBER_OF_CONNECTIONS = 25;
      int REPEATS = 10;
      ActiveMQServer server = createServer(true, createDefaultConfig(true));
      server.start();
      Queue serverQueue = server.createQueue(QueueConfiguration.of("test-queue").setRoutingType(RoutingType.ANYCAST).setAddress("test-queue").setAutoCreated(false));

      ExecutorService executorService = Executors.newFixedThreadPool(NUMBER_OF_CONNECTIONS);
      runAfter(executorService::shutdownNow);

      CountDownLatch done = new CountDownLatch(NUMBER_OF_CONNECTIONS);

      ConnectionFactory factory = CFUtil.createConnectionFactory(protocol, "tcp://localhost:61616");

      final AtomicBoolean running = new AtomicBoolean(true);

      runAfter(() -> running.set(false));

      CyclicBarrier flagStart = new CyclicBarrier(NUMBER_OF_CONNECTIONS + 1);
      flagStart.reset();

      for (int i = 0; i < NUMBER_OF_CONNECTIONS; i++) {
         final int t = i;
         executorService.execute(() -> {
            try {
               boolean alreadyStarted = false;
               AtomicBoolean ex = new AtomicBoolean(true);
               while (running.get()) {
                  try {
                     // do not be tempted to use try (connection = factory.createConnection())
                     // this is because we don't need to close the connection after a network failure on this test.
                     Connection connection = factory.createConnection();

                     synchronized (ConnectionDroppedTest.this) {
                        runAfter(connection::close);
                     }
                     connection.setExceptionListener(exception -> ex.set(true));
                     flagStart.await(60, TimeUnit.SECONDS);

                     connection.start();

                     Session session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
                     javax.jms.Queue jmsQueue = session.createQueue("test-queue");

                     while (running.get() && !ex.get()) {
                        if (!alreadyStarted) {
                           alreadyStarted = true;
                        }
                        System.out.println("Consumer");
                        MessageConsumer consumer = session.createConsumer(jmsQueue);
                        Thread.sleep(500);
                     }

                     if (!protocol.equals("CORE")) {
                        connection.close();
                     }
                  } catch (Exception e) {
                     logger.debug(e.getMessage(), e);
                  }
               }
            } finally {
               done.countDown();
            }
         });
      }

      for (int i = 0; i < REPEATS; i++) {
         try {
            flagStart.await(60, TimeUnit.SECONDS); // align all the clients at the same spot
         } catch (Throwable throwable) {
            logger.info(ThreadDumpUtil.threadDump("timed out flagstart"));
            throw throwable;
         }

         logger.info("*******************************************************************************************************************************\nloop kill {}" + "\n*******************************************************************************************************************************", i);
         server.getRemotingService().getConnections().forEach(r -> {
            r.fail(new ActiveMQException("it's a simulation"));
         });

      }

      running.set(false);
      try {
         flagStart.await(1, TimeUnit.SECONDS);
      } catch (Exception ignored) {
      }
      if (!done.await(10, TimeUnit.SECONDS)) {
         logger.warn(ThreadDumpUtil.threadDump("Threads are still running"));
         fail("Threads are still running");
      }

      Wait.assertEquals(0, () -> serverQueue.getConsumers().size(), 5000, 100);

   }

   @Test
   public void testDropConsumerProtonJ2TestClient() throws Throwable {
      ReusableLatch latchCreating = new ReusableLatch(1);
      ReusableLatch blockCreate = new ReusableLatch(1);
      ReusableLatch done = new ReusableLatch(1);
      ActiveMQServer server = createServer(true, createDefaultConfig(true));
      server.start();

      int TEST_REPEATS = 4;

      server.registerBrokerPlugin(new ActiveMQServerSessionPlugin() {
         @Override
         public void beforeCreateSession(String name,
                                         String username,
                                         int minLargeMessageSize,
                                         RemotingConnection connection,
                                         boolean autoCommitSends,
                                         boolean autoCommitAcks,
                                         boolean preAcknowledge,
                                         boolean xa,
                                         String defaultAddress,
                                         SessionCallback callback,
                                         boolean autoCreateQueues,
                                         OperationContext context,
                                         Map<SimpleString, RoutingType> prefixes) throws ActiveMQException {
            latchCreating.countDown();
            try {
               blockCreate.await(10, TimeUnit.HOURS);
            } catch (Exception e) {
               logger.warn(e.getMessage(), e);
            } finally {
               done.countDown();
            }
         }
      });

      AtomicBoolean running = new AtomicBoolean(true);
      ExecutorService executorService = Executors.newFixedThreadPool(1);
      runAfter(executorService::shutdownNow);
      runAfter(() -> running.set(false));
      Queue serverQueue = server.createQueue(QueueConfiguration.of(getName()).setRoutingType(RoutingType.ANYCAST).setAddress(getName()).setAutoCreated(false));

      for (int i = 0; i < TEST_REPEATS; i++) {
         assertEquals(0, serverQueue.getConsumerCount());
         latchCreating.setCount(1);
         blockCreate.setCount(1);
         done.setCount(1);

         ProtonTestClient peer = new ProtonTestClient();

         executorService.execute(() -> {

            try {
               peer.queueClientSaslAnonymousConnect();
               peer.remoteOpen().queue();
               peer.remoteBegin().queue();
               peer.remoteAttach().ofReceiver().withName(RandomUtil.randomString()).withSenderSettleModeUnsettled().withReceivervSettlesFirst().withTarget().also().withSource().withAddress(getName()).withExpiryPolicyOnLinkDetach().withDurabilityOfNone().withCapabilities("queue").withOutcomes("amqp:accepted:list", "amqp:rejected:list").also().queue();

               peer.connect("localhost", 61616);

               peer.waitForScriptToCompleteIgnoreErrors();
            } catch (Throwable e) {
               logger.warn(e.getMessage(), e);
            }
         });

         assertTrue(latchCreating.await(10, TimeUnit.SECONDS));
         server.getRemotingService().getConnections().forEach(r -> {
            r.fail(new ActiveMQException("it's a simulation"));
         });
         blockCreate.countDown();
         assertTrue(done.await(10, TimeUnit.SECONDS));

         { // double checking the executor is done
            CountDownLatch check = new CountDownLatch(1);
            executorService.execute(check::countDown);
            assertTrue(check.await(1, TimeUnit.SECONDS));
         }

         Thread.sleep(100); // I need some time for the error condition to kick in

         Wait.assertEquals(0, server::getConnectionCount, 5000);

         Wait.assertEquals(0, serverQueue::getConsumerCount, 5000);
      }

   }

   @Test
   public void testDropCreateConsumerAMQP() throws Throwable {
      testDropCreateConsumer("AMQP");
   }

   @Test
   public void testDropCreateConsumerOPENWIRE() throws Throwable {
      testDropCreateConsumer("OPENWIRE");
   }

   @Test
   public void testDropCreateConsumerCORE() throws Throwable {
      testDropCreateConsumer("CORE");
   }

   private void testDropCreateConsumer(String protocol) throws Throwable {
      CountDownLatch latchCreating = new CountDownLatch(1);
      CountDownLatch blockCreate = new CountDownLatch(1);
      CountDownLatch done = new CountDownLatch(1);
      CountDownLatch doneConsumer = new CountDownLatch(1);
      ActiveMQServer server = createServer(true, createDefaultConfig(true));
      server.start();

      Queue serverQueue = server.createQueue(QueueConfiguration.of(getName()).setRoutingType(RoutingType.ANYCAST).setAddress(getName()).setAutoCreated(false));

      AtomicBoolean running = new AtomicBoolean(true);
      ExecutorService executorService = Executors.newFixedThreadPool(1);
      runAfter(executorService::shutdownNow);
      runAfter(() -> running.set(false));

      CountDownLatch received = new CountDownLatch(1);

      executorService.execute(() -> {
         ConnectionFactory connectionFactory;

         switch (protocol) {
            case "AMQP":
               connectionFactory = CFUtil.createConnectionFactory(protocol, "failover:(amqp://localhost:61616)?failover.maxReconnectAttempts=20");
               break;
            case "OPENWIRE":
               connectionFactory = new org.apache.activemq.ActiveMQConnectionFactory( "failover:(tcp://localhost:61616)?maxReconnectAttempts=20");
               break;
            case "CORE":
               connectionFactory = new org.apache.activemq.artemis.jms.client.ActiveMQConnectionFactory("tcp://localhost:61616?ha=true;reconnectAttempts=20;callTimeout=1000");
               break;
            default:
               logger.warn("Invalid protocol {}", protocol);
               connectionFactory = null;
         }

         try (Connection connection = connectionFactory.createConnection()) {
            logger.info("Connected on thread..");
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            MessageConsumer consumer = session.createConsumer(session.createQueue(getName()));
            connection.start();
            while (running.get()) {
               try {
                  logger.info("Receiving");
                  received.countDown();
                  Message message = consumer.receive(100);
               } catch (Exception e) {
                  logger.warn(e.getMessage(), e);
               }
            }
         } catch (Exception e) {
            logger.warn(e.getMessage(), e);
         } finally {
            doneConsumer.countDown();
         }
      });

      assertTrue(received.await(10, TimeUnit.SECONDS));

      server.registerBrokerPlugin(new ActiveMQServerSessionPlugin() {
         @Override
         public void beforeCreateSession(String name,
                                         String username,
                                         int minLargeMessageSize,
                                         RemotingConnection connection,
                                         boolean autoCommitSends,
                                         boolean autoCommitAcks,
                                         boolean preAcknowledge,
                                         boolean xa,
                                         String defaultAddress,
                                         SessionCallback callback,
                                         boolean autoCreateQueues,
                                         OperationContext context,
                                         Map<SimpleString, RoutingType> prefixes) throws ActiveMQException {
            latchCreating.countDown();
            try {
               blockCreate.await(10, TimeUnit.HOURS);
            } catch (Exception e) {
               logger.warn(e.getMessage(), e);
            } finally {
               done.countDown();
            }
         }
      });


      server.getRemotingService().getConnections().forEach(r -> {
         r.fail(new ActiveMQException("it's a simulation"));
      });
      assertTrue(latchCreating.await(10, TimeUnit.SECONDS));
      server.getRemotingService().getConnections().forEach(r -> {
         r.fail(new ActiveMQException("it's a simulation 2nd time"));
      });
      blockCreate.countDown();
      assertTrue(done.await(10, TimeUnit.SECONDS));

      running.set(false);
      assertTrue(doneConsumer.await(40, TimeUnit.SECONDS));

      Thread.sleep(100);

      { // double checking the executor is done
         CountDownLatch check = new CountDownLatch(1);
         executorService.execute(check::countDown);
         assertTrue(check.await(1, TimeUnit.SECONDS));
      }

      Wait.assertEquals(0, server::getConnectionCount, 5000);
      Wait.assertEquals(0, serverQueue::getConsumerCount, 5000);

   }


   @Test
   @Timeout(10)
   public void testForceDropOpenWire() throws Throwable {
      ActiveMQServer server = createServer(true, createDefaultConfig(true));
      server.start();

      Queue serverQueue = server.createQueue(QueueConfiguration.of("test-queue").setRoutingType(RoutingType.ANYCAST).setAddress("test-queue").setAutoCreated(false));

      CountDownLatch beforeCreateCalled = new CountDownLatch(1);
      CountDownLatch goCreateConsumer = new CountDownLatch(1);
      server.registerBrokerPlugin(new ActiveMQServerConsumerPlugin() {
         @Override
         public void afterCreateConsumer(ServerConsumer consumer) throws ActiveMQException {
            if (consumer.getQueue() == serverQueue) {
               logger.info("Creating a consumer at {}", consumer.getQueue());
               beforeCreateCalled.countDown();
               try {
                  goCreateConsumer.await(5, TimeUnit.MINUTES);
               } catch (Exception e) {
                  logger.warn(e.getMessage(), e);
               }
            }
         }
      });

      ExecutorService executorService = Executors.newFixedThreadPool(1);
      runAfter(executorService::shutdownNow);

      ConnectionFactory factory = CFUtil.createConnectionFactory("OPENWIRE", "tcp://localhost:61616");

      CountDownLatch done = new CountDownLatch(1);

      executorService.execute(() -> {
         try (Connection connection = factory.createConnection();
              Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
              MessageConsumer consumer = session.createConsumer(session.createQueue("test-queue"))) {
         } catch (Exception e) {
            logger.warn(e.getMessage(), e);
         } finally {
            done.countDown();
         }
      });

      assertTrue(beforeCreateCalled.await(5, TimeUnit.MINUTES));

      server.getRemotingService().getConnections().forEach(r -> {
         r.fail(new ActiveMQException("this is a simulation"));
      });

      goCreateConsumer.countDown();

      Wait.assertEquals(0, serverQueue::getConsumerCount);
   }



}
