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

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 java.lang.invoke.MethodHandles;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.activemq.artemis.tests.soak.SoakTestBase;
import org.apache.activemq.artemis.tests.util.CFUtil;
import org.apache.activemq.artemis.utils.SpawnedVMSupport;
import org.junit.Assert;
import org.junit.Assume;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import static org.apache.activemq.artemis.tests.soak.TestParameters.intMandatoryProperty;
import static org.apache.activemq.artemis.tests.soak.TestParameters.testProperty;

/**
 * Refer to ./scripts/parameters.sh for suggested parameters
 *
 * Even though this test is not testing Paging, it will use Page just to generate enough load to the server to compete for resources in Native Buffers.
 *
 */
@RunWith(Parameterized.class)
public class OWLeakTest extends SoakTestBase {

   private static final int OK = 33; // arbitrary code. if the spawn returns this the test went fine

   public static final String SERVER_NAME_0 = "openwire-leaktest";
   private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
   private static final String TEST_NAME = "OW_LEAK";
   private static final boolean TEST_ENABLED = Boolean.parseBoolean(testProperty(TEST_NAME, "TEST_ENABLED", "true"));
   private static final String PROTOCOL_LIST = testProperty(TEST_NAME, "PROTOCOL_LIST", "OPENWIRE");
   private static final int TEST_TIMEOUT_MINUTES = testProperty(TEST_NAME, "TIMETOUT_MINUTES", 10);
   private final String protocol;
   private final int NUMBER_OF_MESSAGES;
   private final int PRODUCERS;
   private final int MESSAGE_SIZE;
   Process serverProcess;

   public OWLeakTest(String protocol) {
      this.protocol = protocol;
      NUMBER_OF_MESSAGES = intMandatoryProperty(TEST_NAME, protocol + "_NUMBER_OF_MESSAGES");
      PRODUCERS = intMandatoryProperty(TEST_NAME, protocol + "_PRODUCERS");
      MESSAGE_SIZE = intMandatoryProperty(TEST_NAME, protocol + "_MESSAGE_SIZE");
   }

   @Parameterized.Parameters(name = "protocol={0}")
   public static Collection<Object[]> parameters() {
      String[] protocols = PROTOCOL_LIST.split(",");

      ArrayList<Object[]> parameters = new ArrayList<>();
      for (String str : protocols) {
         logger.debug("Adding {} to the list for the test", str);
         parameters.add(new Object[]{str});
      }

      return parameters;
   }

   @Before
   public void before() throws Exception {
      Assume.assumeTrue(TEST_ENABLED);
      cleanupData(SERVER_NAME_0);

      serverProcess = startServer(SERVER_NAME_0, 0, 10_000);
   }


   private static String createLMBody(int messageSize, int producer, int sequence) {
      StringBuffer buffer = new StringBuffer();
      String baseString = "A Large body from producer " + producer + ", sequence " + sequence;

      while (buffer.length() < messageSize) {
         buffer.append(baseString);
      }
      return buffer.toString();
   }


   public static void main(String[] arg) {
      int PRODUCERS = Integer.parseInt(arg[0]);
      int NUMBER_OF_MESSAGES = Integer.parseInt(arg[1]);
      int MESSAGE_SIZE = Integer.parseInt(arg[2]);
      String protocol = arg[3];
      ExecutorService service = Executors.newFixedThreadPool(PRODUCERS + 1 + 1);

      String QUEUE_NAME = "some_queue";

      Semaphore semaphore = new Semaphore(PRODUCERS);

      CountDownLatch latch = new CountDownLatch(PRODUCERS + 1 + 1);

      AtomicBoolean running = new AtomicBoolean(true);

      AtomicInteger errors = new AtomicInteger(0);

      try {

         for (int i = 0; i < PRODUCERS; i++) {
            final int producerID = i;
            ConnectionFactory factory = CFUtil.createConnectionFactory(protocol, "tcp://localhost:61616");
            service.execute(() -> {
               try {
                  for (int msg = 0; msg < NUMBER_OF_MESSAGES; msg++) {
                     Connection connection = factory.createConnection();
                     Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
                     MessageProducer producer = session.createProducer(session.createQueue(QUEUE_NAME));
                     TextMessage message = session.createTextMessage(createLMBody(MESSAGE_SIZE, producerID, msg));
                     message.setIntProperty("producerID", producerID);
                     message.setIntProperty("sequence", msg);
                     semaphore.acquire();
                     producer.send(message);
                     logger.debug("Thread {} Sent message with size {} with the total number of {} messages of {}", producerID, MESSAGE_SIZE, msg, NUMBER_OF_MESSAGES);
                     producer.close();
                     session.close();
                     connection.close();
                  }
               } catch (Exception e) {
                  errors.incrementAndGet();
                  e.printStackTrace();
                  logger.warn(e.getMessage(), e);
               } finally {
                  latch.countDown();
               }
            });
         }


         service.execute(() -> {

            int[] producerSequence = new int[PRODUCERS];

            try {
               ConnectionFactory factory = CFUtil.createConnectionFactory(protocol, "tcp://localhost:61616");
               Connection connection = factory.createConnection();
               Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
               MessageConsumer consumer = session.createConsumer(session.createQueue(QUEUE_NAME));
               connection.start();

               for (int i = 0; i < NUMBER_OF_MESSAGES * PRODUCERS; i++) {
                  TextMessage message = (TextMessage) consumer.receive(60_000);
                  Assert.assertNotNull(message);
                  int producerID = message.getIntProperty("producerID");
                  int sequence = message.getIntProperty("sequence");
                  logger.debug("Received message {} from producer {}", sequence, producerID);
                  Assert.assertEquals(producerSequence[producerID], sequence);
                  producerSequence[producerID]++;
                  Assert.assertEquals(createLMBody(MESSAGE_SIZE, producerID, sequence), message.getText());
                  semaphore.release();
               }

            } catch (Throwable e) {
               errors.incrementAndGet();
               logger.warn(e.getMessage(), e);
            } finally {
               running.set(false);
               latch.countDown();
            }
         });

         service.execute(() -> {
            // this is just creating enough loading somewhere else to compete for resources
            ConnectionFactory factory = CFUtil.createConnectionFactory("core", "tcp://localhost:61616");
            try {
               Connection connection = factory.createConnection();
               Session session = connection.createSession(true, Session.SESSION_TRANSACTED);
               MessageProducer producer = session.createProducer(session.createQueue("fastQueue"));
               MessageConsumer consumer = session.createConsumer(session.createQueue("fastQueue"));
               connection.start();
               long msg = 0;
               char[] msgStr = new char[1024];
               String buffer = new String(msgStr);
               Arrays.fill(msgStr, 'a');
               while (running.get()) {
                  TextMessage message = session.createTextMessage(buffer);
                  producer.send(message);
                  if (++msg % 10000L == 0L) {
                     logger.debug("Sent and receive {} fast messages", msg);
                  }

                  if (msg > 5000L) {
                     message = (TextMessage) consumer.receive(10000);
                     Assert.assertNotNull(message);
                  }

                  if (msg % 100L == 0L) {
                     session.commit();
                  }
               }
               session.commit();
               producer.close();
               consumer.close();
               session.close();
               connection.close();
            } catch (Exception e) {
               errors.incrementAndGet();
               e.printStackTrace();
               logger.warn(e.getMessage(), e);
            } finally {
               latch.countDown();
               running.set(false);
            }
         });


         Assert.assertTrue(latch.await(TEST_TIMEOUT_MINUTES, TimeUnit.MINUTES));

         Assert.assertEquals(0, errors.get());

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

   @Test
   public void testValidateLeaks() throws Exception {
      // I am using a spawn for the test client, as this test will need a big VM for the client.
      // so I need control over the memory size for the VM.
      Process process = SpawnedVMSupport.spawnVM(OWLeakTest.class.getName(), new String[]{"-Xmx3G"}, "" + PRODUCERS, "" + NUMBER_OF_MESSAGES, "" + MESSAGE_SIZE, protocol);
      logger.debug("Process PID::{}", process.pid());
      Assert.assertTrue(process.waitFor(TEST_TIMEOUT_MINUTES, TimeUnit.MINUTES));
      Assert.assertEquals(OK, process.exitValue());

   }

}
