/*
 * 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.smoke.mmfactory;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.MessageProducer;
import javax.jms.Queue;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.jms.Topic;
import javax.management.MBeanServerConnection;
import javax.management.MBeanServerInvocationHandler;
import javax.management.ObjectName;
import javax.management.remote.JMXConnector;
import java.io.File;
import java.io.FileOutputStream;
import java.io.PrintStream;
import java.util.Arrays;
import java.util.Collection;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.activemq.artemis.api.config.ActiveMQDefaultConfiguration;
import org.apache.activemq.artemis.api.core.RoutingType;
import org.apache.activemq.artemis.api.core.SimpleString;
import org.apache.activemq.artemis.api.core.management.ObjectNameBuilder;
import org.apache.activemq.artemis.api.core.management.QueueControl;
import org.apache.activemq.artemis.tests.smoke.common.SmokeTestBase;
import org.apache.activemq.artemis.utils.RandomUtil;
import org.apache.activemq.artemis.utils.SpawnedVMSupport;
import org.apache.activemq.artemis.utils.Wait;
import org.apache.qpid.jms.JmsConnectionFactory;
import org.jboss.logging.Logger;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;

@RunWith(Parameterized.class)
public class MMSFactoryTest extends SmokeTestBase {

   private static final Logger logger = Logger.getLogger(MMSFactoryTest.class);

   public static final String SERVER_NAME_0 = "mmfactory";
   private static final String JMX_SERVER_HOSTNAME = "localhost";
   private static final int JMX_SERVER_PORT = 11099;

   final String theprotocol;
   final int BATCH_SIZE;
   // how many times the server will be restarted
   final int restarts;
   // how many times the clients will run per restart
   final int clientRuns;
   // As the produces sends messages, a client will be killed every X messages. This is it!
   final int killClientEveryX;


   @Parameterized.Parameters(name = "protocol={0}, batchSize={1}, restarts={2}, clientRuns={3}, killEveryX={4}")
   public static Collection<Object[]> parameters() {
      return Arrays.asList(new Object[][]{{"CORE", 2000, 2, 2, 500}, {"AMQP", 2000, 2, 2, 500}});
   }

   public MMSFactoryTest(String protocol, int batchSize, int restarts, int clientRuns, int killClientEveryX) {
      this.theprotocol = protocol;
      this.BATCH_SIZE = batchSize;
      this.restarts = restarts;
      this.clientRuns = clientRuns;
      this.killClientEveryX = killClientEveryX;
   }


   public static void main(String[] arg) {

      try {
         Consumer consumer = new Consumer(arg[0], Integer.parseInt(arg[1]), arg[2], Integer.parseInt(arg[3]), arg[4], Integer.parseInt(arg[5]));
         //consumer.run();
         consumer.runListener();

         while (true) {
            Thread.sleep(10_000);
         }
      } catch (Throwable e) {
         System.exit(1);
      }
   }

   public static String getConsumerLog(int id) {
      return getServerLocation(SERVER_NAME_0) + "/data/" + "Consumer" + id + ".log";
   }

   public static ConnectionFactory createConnectionFactory(String protocol, String uri) {
      if (protocol.toUpperCase().equals("OPENWIRE")) {
         return new org.apache.activemq.ActiveMQConnectionFactory(uri);
      } else if (protocol.toUpperCase().equals("AMQP")) {

         if (uri.startsWith("tcp://")) {
            // replacing tcp:// by amqp://
            uri = "amqp" + uri.substring(3);
         }
         return new JmsConnectionFactory(uri);
      } else if (protocol.toUpperCase().equals("CORE") || protocol.toUpperCase().equals("ARTEMIS")) {
         return new org.apache.activemq.artemis.jms.client.ActiveMQConnectionFactory(uri);
      } else {
         throw new IllegalStateException("Unkown:" + protocol);
      }
   }

   Process startConsumerProcess(String protocol,
                                int slowTime,
                                String queueName,
                                int credits,
                                int consumerID) throws Exception {

      return SpawnedVMSupport.spawnVM(MMSFactoryTest.class.getName(), protocol, "" + slowTime, queueName, "" + credits, getConsumerLog(consumerID), "" + consumerID);
   }

   Process serverProcess;

   @Before
   public void before() throws Exception {
      cleanupData(SERVER_NAME_0);
      disableCheckThread();
      serverProcess = startServer(SERVER_NAME_0, 0, 30000);
   }

   @Test
   public void testMMSorting() throws Exception {
      for (int i = 0; i < restarts; i++) {
         logger.debug("*******************************************************************************************************************************");
         logger.debug("Starting " + clientRuns);
         logger.debug("*******************************************************************************************************************************");
         testMMSorting(clientRuns * i, clientRuns * (i + 1));

         stopServerWithFile(getServerLocation(SERVER_NAME_0));
         Thread.sleep(1000);

         try {
            serverProcess.destroyForcibly();
         } catch (Throwable ignored) {
         }
         serverProcess = startServer(SERVER_NAME_0, 0, 30000);
      }
   }

   public void testMMSorting(int countStart, int countEnd) throws Exception {

      JMXConnector jmxConnector = getJmxConnector(JMX_SERVER_HOSTNAME, JMX_SERVER_PORT);

      MBeanServerConnection mBeanServerConnection = jmxConnector.getMBeanServerConnection();
      String brokerName = "0.0.0.0";  // configured e.g. in broker.xml <broker-name> element
      ObjectNameBuilder objectNameBuilder = ObjectNameBuilder.create(ActiveMQDefaultConfiguration.getDefaultJmxDomain(), brokerName, true);
      //ActiveMQServerControl activeMQServerControl = MBeanServerInvocationHandler.newProxyInstance(mBeanServerConnection, objectNameBuilder.getActiveMQServerObjectName(), ActiveMQServerControl.class, false);

      ObjectName queueObjectName = objectNameBuilder.getQueueObjectName(SimpleString.toSimpleString("MMFactory"), SimpleString.toSimpleString("MMConsumer"), RoutingType.MULTICAST);
      QueueControl queueControl = MBeanServerInvocationHandler.newProxyInstance(mBeanServerConnection, queueObjectName, QueueControl.class, false);

      final int NUMBER_OF_CONSUMERS = 6;
      final int NUMBER_OF_FASTCONSUMERS = 0; // not using at the moment

      Process[] consumers = new Process[NUMBER_OF_CONSUMERS + NUMBER_OF_FASTCONSUMERS];
      int[] timeForConsumers = new int[NUMBER_OF_CONSUMERS + NUMBER_OF_FASTCONSUMERS];

      for (int i = 0; i < NUMBER_OF_CONSUMERS; i++) {
         timeForConsumers[i] = (i % 2 == 0 ? 200 : 300);
      }

      timeForConsumers[1] = 100;
      timeForConsumers[5] = 500;

      for (int i = NUMBER_OF_CONSUMERS; i < NUMBER_OF_CONSUMERS + NUMBER_OF_FASTCONSUMERS; i++) {
         timeForConsumers[i] = 0;
      }

      for (int i = 0; i < consumers.length; i++) {
         consumers[i] = startConsumerProcess(theprotocol, timeForConsumers[i], "MMFactory::MMConsumer", 100, i);
      }

      Process deadConsumer = startConsumerProcess(theprotocol, 0, "MMFactory::MMDeadConsumer", 100, 0);

      Process dlqProcess = startConsumerProcess(theprotocol, 0, "DLQ", 100, 1000);

      AtomicInteger retryNumber = new AtomicInteger(0);
      int expectedTotalSize = 0;

      ConnectionFactory factory = createConnectionFactory(theprotocol, "tcp://localhost:61616");
      try (Connection connection = factory.createConnection()) {
         Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
         Topic queue = session.createTopic("MMFactory");
         MessageProducer mmsFactory = session.createProducer(queue);

         Topic controlTopic = session.createTopic("MMControl");


         String largeString;
         {
            StringBuffer largeStringBuffer = new StringBuffer();

            while (largeStringBuffer.length() < 10) {
               largeStringBuffer.append(RandomUtil.randomString());
            }
            largeString = largeStringBuffer.toString();
         }

         try {
            for (int run = countStart; run <= countEnd; run++) {
               AtomicInteger lastTime = new AtomicInteger((int)queueControl.getMessagesAcknowledged());
               for (int i = 0; i < BATCH_SIZE; i++) {
                  if (i > 0 && i % killClientEveryX == 0) {
                     System.out.println("REconnecting...");
                     logger.debug("Reconnecting...");
                     consumers[0].destroyForcibly();
                     consumers[0] = startConsumerProcess(theprotocol, timeForConsumers[0], "MMFactory::MMConsumer", 100, 0);
                     consumers[1].destroyForcibly();
                     consumers[1] = startConsumerProcess(theprotocol, timeForConsumers[1], "MMFactory::MMConsumer", 100, 1);
                     logger.debug("...Reconnected");
                     logger.debug("retry=" + retryNumber + ",sent=" + i + ", acked on this batch = " + (queueControl.getMessagesAcknowledged() - (retryNumber.get() * BATCH_SIZE * 2)) + ", total acked = " + queueControl.getMessagesAcknowledged());
                  }
                  TextMessage message = session.createTextMessage("This is blue " + largeString);
                  message.setStringProperty("color", "blue");
                  message.setIntProperty("i", i);
                  mmsFactory.send(message);

                  message = session.createTextMessage("This is red " + largeString);
                  message.setStringProperty("color", "red");
                  message.setIntProperty("i", i);
                  mmsFactory.send(message);

                  if (i % 10 == 0) {
                     logger.debug("Sending " + i + " run = " + run);
                  }

                  if (i == 0) {
                     waitForAckChange(queueControl, lastTime);
                  }
               }

               Session sessionControl = connection.createSession(true, Session.SESSION_TRANSACTED);
               MessageProducer producerControl = sessionControl.createProducer(controlTopic);
               Message controlmessage = sessionControl.createMessage();
               controlmessage.setStringProperty("control", "flush");
               producerControl.send(controlmessage);
               sessionControl.commit();
               sessionControl.close();

               Wait.assertTrue(() -> {
                  // We will wait a bit here until it's a least a bit closer to the whole Batch
                  if ((queueControl.getMessagesAcknowledged() + queueControl.getMessagesKilled() + queueControl.getMessagesExpired()) - (retryNumber.get() * BATCH_SIZE * 2) > (BATCH_SIZE * 2 - 500)) {
                     return true;
                  } else {
                     logger.debug("Received " + queueControl.getMessagesAcknowledged());
                     return false;
                  }
               }, 45_000, 1_000);

               expectedTotalSize += BATCH_SIZE * 2;
               Wait.assertEquals(expectedTotalSize, queueControl::getMessagesAdded);

               Wait.assertEquals(expectedTotalSize, () -> queueControl.getMessagesAcknowledged() + queueControl.getMessagesKilled());
               retryNumber.incrementAndGet();

               for (Process c : consumers) {
                  c.destroyForcibly();
               }
               for (int i = 0; i < consumers.length; i++) {
                  File file = new File(getConsumerLog(i));
                  if (!file.delete()) {
                     logger.debug("not possible to remove " + file);
                  }
               }
               for (int r = 0; r < consumers.length; r++) {
                  consumers[r] = startConsumerProcess(theprotocol, timeForConsumers[r], "MMFactory::MMConsumer", 100, r);
               }
            }

            Thread.sleep(1000);
         } finally {
            deadConsumer.destroyForcibly();
            for (Process c : consumers) {
               c.destroyForcibly();
            }

            dlqProcess.destroyForcibly();

            for (int i = 0; i < consumers.length; i++) {
               File file = new File(getConsumerLog(i));
               if (!file.delete()) {
                  logger.warn("not possible to remove " + file);
               }
            }

            File file = new File(getConsumerLog(1000)); //the DLQ processing ID used
            if (!file.delete()) {
               logger.warn("not possible to remove " + file);
            }
         }
      }

   }

   private void waitForAckChange(QueueControl queueControl, AtomicInteger lastTime) throws Exception {
      Wait.waitFor(() -> {

         if (lastTime.get() == queueControl.getMessagesAcknowledged()) {
            logger.debug("Waiting some change on " + queueControl.getMessagesAcknowledged() + " with messages Added = " + queueControl.getMessagesAdded() + " and killed = " + queueControl.getMessagesKilled());
            return false;
         } else {
            logger.debug("Condition met! with " + queueControl.getMessagesAcknowledged() + " with messages Added = " + queueControl.getMessagesAdded() + " and killed = " + queueControl.getMessagesKilled());
            lastTime.set((int)queueControl.getMessagesAcknowledged());
            return true;
         }
      }, 5_000);
   }

   static class Consumer implements MessageListener {

      boolean clientAck = false;
      volatile int slowTime;
      final String queuename;
      final int credits;
      final String protocol;
      final int id;
      ConnectionFactory factory;
      Connection connection;
      Session session;
      MessageConsumer consumer;
      MessageConsumer controlConsumer;
      Queue queue;

      Session sessionControl;

      PrintStream fileStream;

      Consumer(String protocol, int slowTime, String queueName, int credits, String logOutput, int id) throws Exception {
         this.slowTime = slowTime;
         this.queuename = queueName;
         this.credits = credits;
         this.protocol = protocol;
         fileStream = new PrintStream(new FileOutputStream(logOutput, true), true);
         this.id = id;
      }

      @Override
      public void onMessage(Message message) {
         try {
            String color = message.getStringProperty("color");
            int messageSequence = message.getIntProperty("i");
            if (queuename.equals("DLQ")) {
               logger.debug("Processing DLQ on color=" + color + ", sequence=" + messageSequence);
            } else if (slowTime > 0) {
               Thread.sleep(slowTime);
            }

            if (clientAck) {
               message.acknowledge();
            }

            fileStream.println(color + ";" + messageSequence);
         } catch (Throwable e) {
            e.printStackTrace();
            System.exit(-1);
         }

      }

      class ControlListener implements MessageListener {

         @Override
         public void onMessage(Message message) {
            try {

               // This is received at the client, on a remote machine, System.out is the best option to log here
               System.out.println("Received control message");
               if (message.getStringProperty("control").equals("flush")) {
                  Consumer.this.slowTime = 0;
                  System.out.println("Setting slow time to 0");
               }
               sessionControl.commit();
            } catch (Throwable e) {
               e.printStackTrace();
            }
         }
      }

      public void runListener() {

         //factory = createConnectionFactory(protocol, "tcp://localhost:61616?jms.prefetchPolicy.queuePrefetch=" + credits);
         factory = createConnectionFactory(protocol, "tcp://localhost:61616");

         System.out.println("Starting");
         connect();
         try {
            consumer.setMessageListener(Consumer.this);
         } catch (Exception e) {
            e.printStackTrace();
            System.exit(-1);
         }
      }

      private void connect() {
         try {
            if (connection != null) {
               connection.close();
            }

            connection = factory.createConnection();
            connection.start();
            session = connection.createSession(false, clientAck ? Session.CLIENT_ACKNOWLEDGE : Session.AUTO_ACKNOWLEDGE);
            queue = session.createQueue(queuename);
            consumer = session.createConsumer(queue);

            sessionControl = connection.createSession(true, Session.SESSION_TRANSACTED);
            Topic topic = sessionControl.createTopic("MMControl");
            controlConsumer = sessionControl.createSharedDurableConsumer(topic, "consumer" + id);
            controlConsumer.setMessageListener(new ControlListener());


         } catch (Exception e) {
            e.printStackTrace();
            System.exit(-1);
         }
      }
   }

}
