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

import javax.jms.Connection;
import javax.jms.MessageProducer;
import javax.jms.Session;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.activemq.artemis.api.config.ActiveMQDefaultConfiguration;
import org.apache.activemq.artemis.api.core.Message;
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.api.core.client.ClientConsumer;
import org.apache.activemq.artemis.api.core.client.ClientMessage;
import org.apache.activemq.artemis.api.core.client.ClientProducer;
import org.apache.activemq.artemis.api.core.client.ClientSession;
import org.apache.activemq.artemis.api.core.client.ClientSessionFactory;
import org.apache.activemq.artemis.api.core.client.ServerLocator;
import org.apache.activemq.artemis.core.server.ActiveMQServer;
import org.apache.activemq.artemis.core.server.ActiveMQServers;
import org.apache.activemq.artemis.core.server.Queue;
import org.apache.activemq.artemis.core.server.impl.QueueImpl;
import org.apache.activemq.artemis.core.settings.impl.AddressSettings;
import org.apache.activemq.artemis.jms.client.ActiveMQConnectionFactory;
import org.apache.activemq.artemis.tests.util.ActiveMQTestBase;
import org.apache.activemq.artemis.tests.util.Wait;
import org.apache.activemq.artemis.utils.RandomUtil;
import org.junit.Before;
import org.junit.Test;

public class RingQueueTest extends ActiveMQTestBase {

   private ActiveMQServer server;

   private final SimpleString address = new SimpleString("RingQueueTestAddress");

   private final SimpleString qName = new SimpleString("RingQueueTestQ1");

   @Test
   public void testSimple() throws Exception {
      ServerLocator locator = createNettyNonHALocator().setBlockOnAcknowledge(true).setAckBatchSize(0);
      ClientSessionFactory sf = createSessionFactory(locator);
      ClientSession clientSession = addClientSession(sf.createSession(false, true, true));
      clientSession.createQueue(new QueueConfiguration(qName).setAddress(address).setRingSize(1L));
      clientSession.start();
      final Queue queue = server.locateQueue(qName);
      assertEquals(1, queue.getRingSize());

      ClientProducer producer = clientSession.createProducer(address);

      for (int i = 0, j = 0; i < 500; i += 2, j++) {
         ClientMessage m0 = createTextMessage(clientSession, "hello" + i);
         producer.send(m0);
         Wait.assertTrue(() -> queue.getMessageCount() == 1);
         ClientMessage m1 = createTextMessage(clientSession, "hello" + (i + 1));
         producer.send(m1);
         int expectedMessagesReplaced = j + 1;
         Wait.assertTrue(() -> queue.getMessagesReplaced() == expectedMessagesReplaced);
         Wait.assertTrue(() -> queue.getMessageCount() == 1);
         ClientConsumer consumer = clientSession.createConsumer(qName);
         ClientMessage message = consumer.receiveImmediate();
         message.acknowledge();
         consumer.close();
         assertEquals("hello" + (i + 1), message.getBodyBuffer().readString());
      }
   }

   @Test
   public void testRollback() throws Exception {
      ServerLocator locator = createNettyNonHALocator().setBlockOnAcknowledge(true).setAckBatchSize(0);
      ClientSessionFactory sf = createSessionFactory(locator);
      ClientSession clientSession = addClientSession(sf.createSession(false, true, false));
      clientSession.createQueue(new QueueConfiguration(qName).setAddress(address).setRingSize(1L));
      clientSession.start();
      final Queue queue = server.locateQueue(qName);

      assertEquals(1, queue.getRingSize());

      ClientProducer producer = clientSession.createProducer(address);

      ClientMessage m0 = createTextMessage(clientSession, "hello0");
      producer.send(m0);
      Wait.assertTrue(() -> queue.getMessageCount() == 1);

      ClientConsumer consumer = clientSession.createConsumer(qName);

      ClientMessage message = consumer.receiveImmediate();
      assertNotNull(message);
      Wait.assertTrue(() -> queue.getDeliveringCount() == 1);

      message.acknowledge();
      assertEquals("hello0", message.getBodyBuffer().readString());

      ClientMessage m1 = createTextMessage(clientSession, "hello1");
      producer.send(m1);
      Wait.assertTrue(() -> queue.getDeliveringCount() == 2);
      Wait.assertTrue(() -> queue.getMessagesReplaced() == 0);
      Wait.assertTrue(() -> queue.getMessageCount() == 2);

      clientSession.rollback();
      consumer.close();
      Wait.assertTrue(() -> queue.getDeliveringCount() == 0);
      Wait.assertTrue(() -> queue.getMessagesReplaced() == 1);
      Wait.assertTrue(() -> queue.getMessageCount() == 1);

      consumer = clientSession.createConsumer(qName);
      message = consumer.receiveImmediate();
      assertNotNull(message);
      Wait.assertTrue(() -> queue.getDeliveringCount() == 1);

      message.acknowledge();

      clientSession.commit();

      Wait.assertTrue(() -> queue.getMessagesAcknowledged() == 1);
      assertEquals("hello1", message.getBodyBuffer().readString());
   }

   @Test
   public void testConsumerCloseWithDirectDeliver() throws Exception {
      ServerLocator locator = createNettyNonHALocator().setBlockOnAcknowledge(true).setAckBatchSize(0);
      ClientSessionFactory sf = createSessionFactory(locator);
      ClientSession clientSession = addClientSession(sf.createSession(false, true, false));
      clientSession.createQueue(new QueueConfiguration(qName).setAddress(address).setRingSize(1L));
      clientSession.start();
      final Queue queue = server.locateQueue(qName);
      assertEquals(1, queue.getRingSize());

      ClientConsumer consumer = clientSession.createConsumer(qName);
      ClientProducer producer = clientSession.createProducer(address);

      ClientMessage message = createTextMessage(clientSession, "hello0");
      producer.send(message);
      message = createTextMessage(clientSession, "hello1");
      producer.send(message);
      Wait.assertEquals(2, queue::getMessageCount);
      Wait.assertEquals(2, queue::getDeliveringCount);
      consumer.close();
      Wait.assertEquals(1, queue::getMessageCount);
      Wait.assertEquals(0,  queue::getDeliveringCount);
      Wait.assertEquals(1, queue::getMessagesReplaced);
      consumer = clientSession.createConsumer(qName);
      message = consumer.receiveImmediate();
      assertNotNull(message);
      Wait.assertTrue(() -> queue.getDeliveringCount() == 1);
      message.acknowledge();
      clientSession.commit();
      Wait.assertTrue(() -> queue.getMessagesAcknowledged() == 1);
      assertEquals("hello1", message.getBodyBuffer().readString());
      consumer.close();
      Wait.assertTrue(() -> queue.getMessageCount() == 0);
      Wait.assertTrue(() -> queue.getDeliveringCount() == 0);
      Wait.assertTrue(() -> queue.getMessagesReplaced() == 1);
   }

   @Test
   public void testScheduled() throws Exception {
      ServerLocator locator = createNettyNonHALocator().setBlockOnAcknowledge(true).setAckBatchSize(0);
      ClientSessionFactory sf = createSessionFactory(locator);
      ClientSession clientSession = addClientSession(sf.createSession(false, true, false));
      clientSession.createQueue(new QueueConfiguration(qName).setAddress(address).setRingSize(1L));
      clientSession.start();
      final Queue queue = server.locateQueue(qName);
      assertEquals(1, queue.getRingSize());

      ClientProducer producer = clientSession.createProducer(address);

      ClientMessage m0 = createTextMessage(clientSession, "hello0");
      long time = System.currentTimeMillis();
      time += 500;
      m0.putLongProperty(Message.HDR_SCHEDULED_DELIVERY_TIME, time);
      producer.send(m0);
      Wait.assertTrue(() -> queue.getScheduledCount() == 1);
      Wait.assertTrue(() -> ((QueueImpl) queue).getMessageCountForRing() == 0);
      time = System.currentTimeMillis();
      time += 500;
      m0.putLongProperty(Message.HDR_SCHEDULED_DELIVERY_TIME, time);
      producer.send(m0);
      Wait.assertTrue(() -> queue.getScheduledCount() == 2);
      Wait.assertTrue(() -> ((QueueImpl) queue).getMessageCountForRing() == 0);
      Wait.assertTrue(() -> queue.getMessagesReplaced() == 1);
      Wait.assertTrue(() -> ((QueueImpl) queue).getMessageCountForRing() == 1);
   }

   @Test
   public void testDefaultAddressSetting() throws Exception {
      SimpleString random = RandomUtil.randomSimpleString();
      server.getAddressSettingsRepository().addMatch(address.toString(), new AddressSettings().setDefaultRingSize(100));

      ActiveMQConnectionFactory cf = new ActiveMQConnectionFactory("vm://0");
      Connection c = cf.createConnection();
      Session s = c.createSession();
      MessageProducer producer = s.createProducer(s.createQueue(address.toString()));
      producer.send(s.createMessage());
      Wait.assertTrue(() -> server.locateQueue(address) != null);
      assertEquals(100, server.locateQueue(address).getRingSize());
      producer.close();
      producer = s.createProducer(s.createQueue(random.toString()));
      producer.send(s.createMessage());
      Wait.assertTrue(() -> server.locateQueue(random) != null);
      assertEquals(ActiveMQDefaultConfiguration.getDefaultRingSize(), server.locateQueue(random).getRingSize());
   }

   @Test
   public void testUpdate() throws Exception {
      ServerLocator locator = createNettyNonHALocator().setBlockOnAcknowledge(true).setAckBatchSize(0);
      ClientSessionFactory sf = createSessionFactory(locator);
      ClientSession clientSession = addClientSession(sf.createSession(false, true, true));
      clientSession.createQueue(new QueueConfiguration(qName).setAddress(address).setRoutingType(RoutingType.ANYCAST));
      clientSession.start();
      final Queue queue = server.locateQueue(qName);
      assertEquals(-1, queue.getRingSize());

      ClientProducer producer = clientSession.createProducer(address);
      for (int i = 0; i < 100; i++) {
         producer.send(clientSession.createMessage(true));
      }
      Wait.assertTrue(() -> queue.getMessageCount() == 100);

      queue.setRingSize(10);

      ClientConsumer consumer = clientSession.createConsumer(qName);
      ClientMessage message;
      for (int j = 0; j < 95; j++) {
         message = consumer.receiveImmediate();
         message.acknowledge();
      }
      consumer.close();
      Wait.assertEquals(5, queue::getMessageCount);

      for (int i = 0; i < 5; i++) {
         producer.send(clientSession.createMessage(true));
      }
      Wait.assertEquals(10, queue::getMessageCount);

      // these sends will be replacing the old values
      for (int i = 0; i < 5; i++) {
         producer.send(clientSession.createMessage(true));
         Wait.assertEquals(10, queue::getMessageCount);
      }

      Wait.assertEquals(5, queue::getMessagesReplaced);
      consumer = clientSession.createConsumer(qName);
      message = consumer.receiveImmediate();
      assertNotNull(message);
      message.acknowledge();
      consumer.close();
      Wait.assertTrue(() -> queue.getMessageCount() == 9);

      queue.setRingSize(5);

      consumer = clientSession.createConsumer(qName);
      for (int j = 0; j < 4; j++) {
         message = consumer.receiveImmediate();
         message.acknowledge();
      }
      consumer.close();
      Wait.assertTrue(() -> queue.getMessageCount() == 5);
      producer.send(clientSession.createMessage(true));
      Wait.assertTrue(() -> queue.getMessagesReplaced() == 6);

      queue.setRingSize(10);

      for (int i = 0; i < 5; i++) {
         producer.send(clientSession.createMessage(true));
      }
      Wait.assertTrue(() -> queue.getMessageCount() == 10);
      producer.send(clientSession.createMessage(true));
      Wait.assertTrue(() -> queue.getMessagesReplaced() == 7);
      Wait.assertTrue(() -> queue.getMessageCount() == 10);
   }

   @Test
   public void testNonDestructive() throws Exception {
      ServerLocator locator = createNettyNonHALocator().setBlockOnAcknowledge(true).setAckBatchSize(0);
      ClientSessionFactory sf = createSessionFactory(locator);
      ClientSession clientSession = addClientSession(sf.createSession(false, true, true));
      clientSession.createQueue(new QueueConfiguration(qName).setAddress(address).setRingSize(1L).setNonDestructive(true));
      clientSession.start();
      final Queue queue = server.locateQueue(qName);
      assertEquals(1, queue.getRingSize());

      ClientProducer producer = clientSession.createProducer(address);

      ClientMessage message = createTextMessage(clientSession, "hello" + 0);
      producer.send(message);
      for (int i = 0; i < 5; i++) {
         Wait.assertTrue(() -> queue.getMessageCount() == 1);
         message = createTextMessage(clientSession, "hello" + (i + 1));
         producer.send(message);
         final int finalI = i + 1;
         Wait.assertTrue(() -> queue.getMessagesReplaced() == finalI);
         Wait.assertTrue(() -> queue.getMessageCount() == 1);
         ClientConsumer consumer = clientSession.createConsumer(qName);
         message = consumer.receiveImmediate();
         assertNotNull(message);
         message.acknowledge(); // non-destructive!
         consumer.close();
         assertEquals("hello" + (i + 1), message.getBodyBuffer().readString());
      }
   }

   @Test
   public void testNonDestructiveWithConsumerClose() throws Exception {
      ServerLocator locator = createNettyNonHALocator().setBlockOnAcknowledge(true).setAckBatchSize(0);
      ClientSessionFactory sf = createSessionFactory(locator);
      ClientSession clientSession = addClientSession(sf.createSession(false, true, true));
      clientSession.createQueue(new QueueConfiguration(qName).setAddress(address).setRingSize(1L).setNonDestructive(true));
      clientSession.start();
      final Queue queue = server.locateQueue(qName);
      assertEquals(1, queue.getRingSize());

      ClientProducer producer = clientSession.createProducer(address);

      ClientMessage m0 = createTextMessage(clientSession, "hello" + 0);
      producer.send(m0);
      Wait.assertTrue(() -> queue.getMessageCount() == 1);
      ClientConsumer consumer = clientSession.createConsumer(qName);
      Wait.assertTrue(() -> queue.getDeliveringCount() == 1);
      consumer.close();
      Wait.assertTrue(() -> queue.getDeliveringCount() == 0);
      Wait.assertTrue(() -> queue.getMessageCount() == 1);
   }

   @Test
   public void testMultipleConcurrentProducers() throws Exception {
      final long RING_SIZE = 25;
      ServerLocator locator = createNettyNonHALocator().setBlockOnAcknowledge(true).setAckBatchSize(0);
      ClientSessionFactory sf = createSessionFactory(locator);
      ClientSession clientSession = addClientSession(sf.createSession(false, true, true));
      clientSession.createQueue(new QueueConfiguration(qName).setAddress(address).setRingSize(RING_SIZE));
      clientSession.start();
      final Queue queue = server.locateQueue(qName);
      assertEquals(RING_SIZE, queue.getRingSize());
      final int nThreads = 25;
      final long numberOfMessages = RING_SIZE;

      SomeProducer[] producers = new SomeProducer[nThreads];

      try {
         for (int i = 0; i < nThreads; i++) {
            producers[i] = new SomeProducer(numberOfMessages, nThreads, address);
         }

         for (int i = 0; i < nThreads; i++) {
            producers[i].start();
         }

         for (SomeProducer producer : producers) {
            producer.join();
            assertEquals(0, producer.errors.get());
         }
      } catch (Exception e) {
         e.printStackTrace();
         fail(e.getMessage());
      }

      Wait.assertTrue("message count should be " + RING_SIZE + " but it's actually " + queue.getMessageCount(), () -> queue.getMessageCount() == RING_SIZE);
   }

   @Override
   @Before
   public void setUp() throws Exception {
      super.setUp();

      server = addServer(ActiveMQServers.newActiveMQServer(createDefaultNettyConfig(), true));
      // start the server
      server.start();
   }

   class SomeProducer extends Thread {

      final ClientSessionFactory factory;
      final ServerLocator locator;
      final ClientSession prodSession;
      public final AtomicInteger errors = new AtomicInteger(0);
      final long numberOfMessages;
      final int nThreads;
      final SimpleString address;

      SomeProducer(long numberOfMessages, int nThreads, SimpleString address) throws Exception {
         locator = createNettyNonHALocator();
         factory = locator.createSessionFactory();
         prodSession = factory.createSession(true, false);
         this.numberOfMessages = numberOfMessages;
         this.nThreads = nThreads;
         this.address = address;
      }

      @Override
      public void run() {
         try {
            ClientProducer producer = prodSession.createProducer(address);
            for (int i = 0; i < numberOfMessages; i++) {
               ClientMessage message = prodSession.createMessage(true);
               message.putIntProperty("prodNR", i % nThreads);
               producer.send(message);
            }

         } catch (Throwable e) {
            e.printStackTrace();
            errors.incrementAndGet();
         } finally {
            try {
               prodSession.close();
               locator.close();
            } catch (Throwable ignored) {
               ignored.printStackTrace();
            }
         }
      }
   }
}
