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

import static org.apache.activemq.artemis.tests.compatibility.GroovyRun.SNAPSHOT;
import static org.apache.activemq.artemis.tests.compatibility.GroovyRun.TWO_THIRTYTHREE_ZERO;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.MessageConsumer;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.jms.Topic;
import java.io.File;
import java.io.PrintWriter;
import java.io.StringWriter;
import java.lang.invoke.MethodHandles;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;

import org.apache.activemq.artemis.tests.compatibility.base.ClasspathBase;
import org.apache.activemq.artemis.tests.extensions.parameterized.ParameterizedTestExtension;
import org.apache.activemq.artemis.tests.extensions.parameterized.Parameters;
import org.apache.activemq.artemis.utils.FileUtil;
import org.apache.qpid.jms.JmsConnectionFactory;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.TestTemplate;
import org.junit.jupiter.api.extension.ExtendWith;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

@ExtendWith(ParameterizedTestExtension.class)
public class MirroredVersionTest extends ClasspathBase {

   private static final String QUEUE_NAME = "MirroredQueue";
   private static final String TOPIC_NAME = "MirroredTopic";

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

   private final ClassLoader mainClassloader;

   private final ClassLoader backupClassLoader;

   private final boolean useDual;

   @Parameters(name = "BrokerA={0}, BrokerB={1}, dualMirror={2}")
   public static Collection getParameters() {
      List<Object[]> combinations = new ArrayList<>();
      combinations.add(new Object[]{TWO_THIRTYTHREE_ZERO, SNAPSHOT, true});
      combinations.add(new Object[]{SNAPSHOT, TWO_THIRTYTHREE_ZERO, true});
      combinations.add(new Object[]{SNAPSHOT, SNAPSHOT, true});
      return combinations;
   }

   public MirroredVersionTest(String main, String backup, boolean useDual) throws Exception {
      this.mainClassloader = getClasspath(main);

      this.backupClassLoader = getClasspath(backup);

      this.useDual = useDual;
   }

   @BeforeEach
   public void beforeEach() {
      deleteFolders();
   }

   @AfterEach
   public void cleanupServers() {
      try {
         evaluate(mainClassloader, "multiVersionMirror/mainServerStop.groovy");
      } catch (Exception ignored) {
      }
      try {
         evaluate(backupClassLoader, "multiVersionMirror/backupServerStop.groovy");
      } catch (Exception ignored) {
      }

      deleteFolders();
   }

   private void deleteFolders() {
      FileUtil.deleteDirectory(new File(serverFolder.getAbsolutePath(), "1"));
      FileUtil.deleteDirectory(new File(serverFolder.getAbsolutePath(), "2"));
   }

   private String createBody(int size) {
      StringWriter writer = new StringWriter();
      PrintWriter pw = new PrintWriter(writer);
      for (int i = 0; i < size; i++) {
         pw.print("-");
      }
      return writer.toString();
   }


   @TestTemplate
   public void testMirrorReplica() throws Throwable {
      testMirrorReplica(100);
   }

   @TestTemplate
   public void testMirrorReplicaLM() throws Throwable {
      testMirrorReplica(300 * 1024);
   }

   public void testMirrorReplica(int stringSize) throws Throwable {
      String body = createBody(stringSize);
      logger.debug("Starting live");
      startMainBroker();

      ConnectionFactory factoryMain = new JmsConnectionFactory("amqp://localhost:61616");

      try (Connection connection = factoryMain.createConnection()) {
         Session session = connection.createSession(true, Session.SESSION_TRANSACTED);
         MessageProducer producer = session.createProducer(session.createQueue(QUEUE_NAME));
         for (int i = 0; i < 10; i++) {
            TextMessage message = session.createTextMessage("hello " + i + body);
            message.setIntProperty("count", i);
            producer.send(message);
         }
         session.commit();
      }

      logger.debug("restarting main server");
      evaluate(mainClassloader, "multiVersionMirror/mainServerStop.groovy");
      startMainBroker();

      logger.debug("starting backup");
      startBackupBroker();

      ConnectionFactory factoryReplica = new JmsConnectionFactory("amqp://localhost:61617");

      try (Connection connection = factoryReplica.createConnection()) {
         Session session = connection.createSession(true, Session.SESSION_TRANSACTED);
         MessageConsumer consumer = session.createConsumer(session.createQueue(QUEUE_NAME));
         connection.start();
         for (int i = 0; i < 10; i++) {
            TextMessage message = (TextMessage) consumer.receive(5000);
            if (message == null) break;
         }
         session.rollback();
      }

      logger.debug("Restarting backup");

      evaluate(backupClassLoader, "multiVersionMirror/backupServerStop.groovy");
      startBackupBroker();

      try (Connection connection = factoryReplica.createConnection()) {
         Session session = connection.createSession(true, Session.SESSION_TRANSACTED);
         MessageConsumer consumer = session.createConsumer(session.createQueue(QUEUE_NAME));
         connection.start();
         for (int i = 0; i < 10; i++) {
            TextMessage message = (TextMessage) consumer.receive(5000);
            if (message == null) {
               break;
            }
         }
         session.commit();
      }
   }

   private void startMainBroker() throws Exception {
      evaluate(mainClassloader, "multiVersionMirror/mainServer.groovy", serverFolder.getAbsolutePath(), "1", QUEUE_NAME, TOPIC_NAME);
   }

   private void startBackupBroker() throws Exception {
      evaluate(backupClassLoader, "multiVersionMirror/backupServer.groovy", serverFolder.getAbsolutePath(), "2", QUEUE_NAME, TOPIC_NAME, String.valueOf(useDual));
   }

   @TestTemplate
   public void testTopic() throws Throwable {
      int stringSize = 100;
      String body = createBody(stringSize);
      logger.debug("Starting live");
      startMainBroker();
      logger.debug("Starting backup");
      startBackupBroker();

      String clientID1 = "CONNECTION_1";
      String clientID2 = "CONNECTION_2";

      String sub1 = "SUB_1";
      String sub2 = "SUB_2";

      ConnectionFactory factoryMain = new JmsConnectionFactory("amqp://localhost:61616");

      try (javax.jms.Connection connection = factoryMain.createConnection()) {
         connection.setClientID(clientID1);
         Session session = connection.createSession(true, Session.SESSION_TRANSACTED);
         Topic topic = session.createTopic(TOPIC_NAME);
         MessageConsumer consumer = session.createDurableConsumer(topic, sub1);
      }
      try (javax.jms.Connection connection = factoryMain.createConnection()) {
         connection.setClientID(clientID2);
         Session session = connection.createSession(true, Session.SESSION_TRANSACTED);
         Topic topic = session.createTopic(TOPIC_NAME);
         MessageConsumer consumer = session.createDurableConsumer(topic, sub2);
      }

      evaluate(backupClassLoader, "multiVersionMirror/backupServerStop.groovy");

      try (Connection connection = factoryMain.createConnection()) {
         Session session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
         Topic topic = session.createTopic(TOPIC_NAME);
         MessageProducer producer = session.createProducer(null);
         for (int i = 0; i < 10; i++) {
            TextMessage message = session.createTextMessage("hello " + i + body);
            message.setIntProperty("count", i);
            producer.send(topic, message);
         }
         session.commit();
      }

      evaluate(mainClassloader, "multiVersionMirror/mainServerStop.groovy");
      startBackupBroker();
      startMainBroker();

      ConnectionFactory factoryReplica = new JmsConnectionFactory("amqp://localhost:61617");

      try (Connection connection = factoryReplica.createConnection()) {
         connection.setClientID(clientID1);
         Session session = connection.createSession(true, Session.SESSION_TRANSACTED);
         Topic topic = session.createTopic(TOPIC_NAME);
         connection.start();
         MessageConsumer consumer = session.createDurableConsumer(topic, sub1);
         for (int i = 0; i < 10; i++) {
            TextMessage message = (TextMessage)consumer.receive(5000);
            if (message == null) {
               break;
            }
         }
         session.rollback();
      }

      logger.debug("Restarting backup");
      evaluate(backupClassLoader, "multiVersionMirror/backupServerStop.groovy");
      startBackupBroker();

      try (Connection connection = factoryReplica.createConnection()) {
         connection.setClientID(clientID1);
         Session session = connection.createSession(true, Session.SESSION_TRANSACTED);
         Topic topic = session.createTopic(TOPIC_NAME);
         connection.start();
         MessageConsumer consumer = session.createDurableConsumer(topic, sub1);
         for (int i = 0; i < 10; i++) {
            TextMessage message = (TextMessage)consumer.receive(5000);
            if (message == null) {
               break;
            }
         }
         session.commit();
      }

      try (Connection connection = factoryReplica.createConnection()) {
         connection.setClientID(clientID2);
         Session session = connection.createSession(true, Session.SESSION_TRANSACTED);
         Topic topic = session.createTopic(TOPIC_NAME);
         connection.start();
         MessageConsumer consumer = session.createDurableConsumer(topic, sub2);
         for (int i = 0; i < 10; i++) {
            TextMessage message = (TextMessage)consumer.receive(5000);
            if (message == null) {
               break;
            }
         }
         session.commit();
      }
   }

}