/*
 * 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 javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.MessageConsumer;
import javax.jms.MessageProducer;
import javax.jms.Queue;
import javax.jms.Session;
import javax.jms.TextMessage;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;

import org.apache.activemq.artemis.jms.client.ActiveMQConnectionFactory;
import org.apache.activemq.artemis.tests.compatibility.base.ClasspathBase;
import org.apache.qpid.jms.JmsConnectionFactory;
import org.junit.After;
import org.junit.Assert;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;

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

@RunWith(Parameterized.class)
public class MultiVersionReplicaTest extends ClasspathBase {

   private static final String QUEUE_NAME = "MultiVersionReplicaTestQueue";

   private final String main;
   private final ClassLoader mainClassloader;

   private final String backup;
   private final ClassLoader backupClassLoader;



   @Parameterized.Parameters(name = "main={0}, backup={1}")
   public static Collection getParameters() {
      List<Object[]> combinations = new ArrayList<>();
      combinations.add(new Object[]{TWO_SEVENTEEN_ZERO, SNAPSHOT});
      combinations.add(new Object[]{SNAPSHOT, TWO_SEVENTEEN_ZERO});
      combinations.add(new Object[]{TWO_EIGHTEEN_ZERO, SNAPSHOT});
      combinations.add(new Object[]{SNAPSHOT, TWO_EIGHTEEN_ZERO});
      // The SNAPSHOT/SNAPSHOT is here as a test validation only, like in other cases where SNAPSHOT/SNAPSHOT is used.
      combinations.add(new Object[]{SNAPSHOT, SNAPSHOT});
      return combinations;
   }

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

      this.backup = backup;
      this.backupClassLoader = getClasspath(backup);
   }

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


   @Test
   public void testReplica() throws Throwable {
      System.out.println("Starting live");
      evaluate(mainClassloader, "multiVersionReplica/mainServer.groovy", serverFolder.getRoot().getAbsolutePath(), "1", "61000", "61001");
      System.out.println("Starting backup");
      evaluate(backupClassLoader, "multiVersionReplica/backupServer.groovy", serverFolder.getRoot().getAbsolutePath(), "2", "61001", "61000");

      evaluate(mainClassloader, "multiVersionReplica/mainServerIsReplicated.groovy");

      send(new ActiveMQConnectionFactory("tcp://localhost:61000"), 2000, 10);
      send(new JmsConnectionFactory("amqp://localhost:61000"), 2000, 10);

      evaluate(mainClassloader, "multiVersionReplica/mainServerStop.groovy");
      evaluate(backupClassLoader, "multiVersionReplica/backupServerIsActive.groovy");

      receive(new ActiveMQConnectionFactory("tcp://localhost:61001"), 2010);
      receive(new JmsConnectionFactory("amqp://localhost:61001"), 2010);

      evaluate(backupClassLoader, "multiVersionReplica/backupServerStop.groovy");
   }


   private void send(ConnectionFactory factory, int numberOfMessagesTx, int numberOfMessagesNonTx) throws Throwable {
      try (Connection connection = factory.createConnection()) {
         Queue queue;

         {
            Session session = connection.createSession(true, Session.SESSION_TRANSACTED);
            queue = session.createQueue(QUEUE_NAME);
            MessageProducer producer = session.createProducer(queue);
            boolean pending = false;
            for (int i = 0; i < numberOfMessagesTx; i++) {
               producer.send(session.createTextMessage("Hello world!!!!!"));
               pending = true;
               if (i > 0 && i % 100 == 0) {
                  session.commit();
                  pending = false;
               }
            }
            if (pending) {
               session.commit();
            }
            session.close();
         }

         {
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            MessageProducer producer = session.createProducer(queue);
            for (int i = 0; i < numberOfMessagesNonTx; i++) {
               producer.send(session.createTextMessage("Hello world!!!!!"));
            }
         }
      }
   }

   private void receive(ConnectionFactory factory, int numberOfMessages) throws Throwable {
      try (Connection connection = factory.createConnection()) {
         Session session = connection.createSession(true, Session.SESSION_TRANSACTED);
         Queue queue = session.createQueue(QUEUE_NAME);
         connection.start();
         MessageConsumer consumer = session.createConsumer(queue);
         boolean pending = false;
         for (int i = 0; i < numberOfMessages; i++) {
            TextMessage message = (TextMessage) consumer.receive(1000);
            Assert.assertNotNull(message);
            pending = true;
            if (i > 0 && i % 100 == 0) {
               session.commit();
               pending = false;
            }
         }
         if (pending) {
            session.commit();
         }
      }
   }

}



