/*
 * 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 static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertTrue;

import java.lang.invoke.MethodHandles;
import java.util.ConcurrentModificationException;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

import org.apache.activemq.artemis.api.core.ActiveMQException;
import org.apache.activemq.artemis.api.core.Message;
import org.apache.activemq.artemis.api.core.QueueConfiguration;
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.MessageReference;
import org.apache.activemq.artemis.core.server.Queue;
import org.apache.activemq.artemis.core.server.impl.LastValueQueue;
import org.apache.activemq.artemis.core.settings.impl.AddressSettings;
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.apache.activemq.artemis.utils.collections.LinkedListIterator;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class LVQTest extends ActiveMQTestBase {

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

   private ActiveMQServer server;

   private ClientSession clientSession;

   private ClientSession clientSessionTxReceives;

   private ClientSession clientSessionTxSends;

   private final SimpleString address = SimpleString.of("LVQTestAddress");

   private final SimpleString qName1 = SimpleString.of("LVQTestQ1");

   @Test
   public void testSimple() throws Exception {
      ClientProducer producer = clientSession.createProducer(address);
      ClientConsumer consumer = clientSession.createConsumer(qName1);
      ClientMessage m1 = createTextMessage(clientSession, "m1");
      SimpleString rh = SimpleString.of("SMID1");
      m1.putStringProperty(Message.HDR_LAST_VALUE_NAME, rh);
      ClientMessage m2 = createTextMessage(clientSession, "m2");
      m2.putStringProperty(Message.HDR_LAST_VALUE_NAME, rh);
      producer.send(m1);
      producer.send(m2);

      Queue lvQueue = server.locateQueue(qName1);
      // delivering is asynchronous. We have to make sure that all messages were sent
      Wait.assertEquals(2L, lvQueue::getMessagesAdded, 5000, 100);
      // messages were replaced by the LVQ so it should have only one
      Wait.assertEquals(1L, lvQueue::getMessageCount, 5000, 100);
      Wait.assertEquals(1L, lvQueue::getMessagesReplaced, 5000, 100);

      clientSession.start();
      ClientMessage m = consumer.receive(1000);
      assertNotNull(m);
      m.acknowledge();
      assertEquals("m2", m.getBodyBuffer().readString());
   }

   @Test
   public void testSimpleExclusive() throws Exception {
      ServerLocator locator = createNettyNonHALocator().setConsumerWindowSize(0);
      ClientSessionFactory sf = createSessionFactory(locator);
      ClientSession clientSession = addClientSession(sf.createSession(false, true, true));
      final String EXCLUSIVE_QUEUE = "exclusiveQueue";

      clientSession.createQueue(QueueConfiguration.of(EXCLUSIVE_QUEUE).setExclusive(true).setLastValue(true));
      ClientProducer producer = clientSession.createProducer(EXCLUSIVE_QUEUE);
      ClientConsumer consumer = clientSession.createConsumer(EXCLUSIVE_QUEUE);
      clientSession.start();
      ClientMessage m1 = createTextMessage(clientSession, "m1");
      SimpleString rh = SimpleString.of("SMID1");
      m1.putStringProperty(Message.HDR_LAST_VALUE_NAME, rh);
      ClientMessage m2 = createTextMessage(clientSession, "m2");
      m2.putStringProperty(Message.HDR_LAST_VALUE_NAME, rh);
      producer.send(m1);
      producer.send(m2);
      ClientMessage m = consumer.receive(1000);
      assertNotNull(m);
      m.acknowledge();
      assertEquals("m2", m.getBodyBuffer().readString());
   }

   @Test
   public void testSimpleRestart() throws Exception {
      ClientProducer producer = clientSession.createProducer(address);
      ClientMessage m1 = createTextMessage(clientSession, "m1");
      SimpleString rh = SimpleString.of("SMID1");
      m1.putStringProperty(Message.HDR_LAST_VALUE_NAME, rh);
      producer.send(m1);
      ClientMessage m2 = createTextMessage(clientSession, "m2");
      m2.putStringProperty(Message.HDR_LAST_VALUE_NAME, rh);
      producer.send(m2);
      Wait.assertEquals(1, () -> server.locateQueue(qName1).getMessageCount());
      clientSession.close();

      server.stop();
      server.start();

      assertEquals(1, server.locateQueue(qName1).getMessageCount());
      ServerLocator locator = createNettyNonHALocator().setBlockOnAcknowledge(true).setAckBatchSize(0);
      ClientSessionFactory sf = createSessionFactory(locator);
      clientSession = addClientSession(sf.createSession(false, true, true));
      producer = clientSession.createProducer(address);
      ClientMessage m3 = createTextMessage(clientSession, "m3");
      m3.putStringProperty(Message.HDR_LAST_VALUE_NAME, rh);
      producer.send(m3);
      // wait b/c prune takes a deliver attempt which is async
      Wait.assertEquals(1, () -> server.locateQueue(qName1).getMessageCount());

      ClientConsumer consumer = clientSession.createConsumer(qName1);
      clientSession.start();
      ClientMessage m = consumer.receive(1000);
      assertNotNull(m);
      m.acknowledge();
      assertEquals("m3", m.getBodyBuffer().readString());
   }

   @Test
   public void testMultipleMessages() throws Exception {
      ClientProducer producer = clientSession.createProducer(address);
      ClientConsumer consumer = clientSession.createConsumer(qName1);
      SimpleString messageId1 = SimpleString.of("SMID1");
      SimpleString messageId2 = SimpleString.of("SMID2");
      ClientMessage m1 = createTextMessage(clientSession, "m1");
      m1.putStringProperty(Message.HDR_LAST_VALUE_NAME, messageId1);
      ClientMessage m2 = createTextMessage(clientSession, "m2");
      m2.putStringProperty(Message.HDR_LAST_VALUE_NAME, messageId2);
      ClientMessage m3 = createTextMessage(clientSession, "m3");
      m3.putStringProperty(Message.HDR_LAST_VALUE_NAME, messageId1);
      ClientMessage m4 = createTextMessage(clientSession, "m4");
      m4.putStringProperty(Message.HDR_LAST_VALUE_NAME, messageId2);
      producer.send(m1);
      producer.send(m2);
      producer.send(m3);
      producer.send(m4);
      Queue serverQueue = server.locateQueue(qName1);
      Wait.assertEquals(2L, serverQueue::getMessageCount, 5000, 100);
      Wait.assertEquals(4L, serverQueue::getMessagesAdded, 5000, 100);
      Wait.assertEquals(2L, serverQueue::getMessagesReplaced, 5000, 100);
      clientSession.start();
      ClientMessage m = consumer.receive(1000);
      assertNotNull(m);
      m.acknowledge();
      assertEquals("m3", m.getBodyBuffer().readString());
      m = consumer.receive(1000);
      assertNotNull(m);
      m.acknowledge();
      assertEquals("m4", m.getBodyBuffer().readString());
      assertNull(consumer.receiveImmediate());
   }

   @Test
   public void testMultipleMessagesWithoutLastValue() throws Exception {
      ClientProducer producer = clientSession.createProducer(address);
      ClientMessage m1 = createTextMessage(clientSession, "message1");
      ClientMessage m2 = createTextMessage(clientSession, "message2");
      producer.send(m1);
      producer.send(m2);

      Wait.assertEquals(2L, () -> server.locateQueue(qName1).getMessageCount(), 2000, 100);

      ClientConsumer consumer = clientSession.createConsumer(qName1);
      clientSession.start();
      ClientMessage m = consumer.receive(1000);
      assertNotNull(m);
      m.acknowledge();
      assertEquals("message1", m.getBodyBuffer().readString());
      m = consumer.receive(1000);
      assertNotNull(m);
      m.acknowledge();
      assertEquals("message2", m.getBodyBuffer().readString());
   }

   @Test
   public void testMultipleRollback() throws Exception {
      AddressSettings qs = new AddressSettings();
      qs.setDefaultLastValueQueue(true);
      qs.setRedeliveryDelay(1);
      server.getAddressSettingsRepository().addMatch(address.toString(), qs);

      ClientProducer producer = clientSessionTxReceives.createProducer(address);
      ClientConsumer consumer = clientSessionTxReceives.createConsumer(qName1);
      SimpleString messageId1 = SimpleString.of("SMID1");
      ClientMessage m1 = createTextMessage(clientSession, "m1");
      m1.putStringProperty(Message.HDR_LAST_VALUE_NAME, messageId1);
      producer.send(m1);
      clientSessionTxReceives.start();
      for (int i = 0; i < 10; i++) {
         logger.debug("#Deliver {}", i);
         ClientMessage m = consumer.receive(5000);
         assertNotNull(m);
         m.acknowledge();
         clientSessionTxReceives.rollback();
      }
      m1 = createTextMessage(clientSession, "m1");
      m1.putStringProperty(Message.HDR_LAST_VALUE_NAME, messageId1);
      producer.send(m1);
      ClientMessage m = consumer.receive(1000);
      assertNotNull(m);
      m.acknowledge();
      assertEquals("m1", m.getBodyBuffer().readString());
      assertNull(consumer.receiveImmediate());
      clientSessionTxReceives.commit();
   }

   @Test
   public void testFirstMessageReceivedButAckedAfter() throws Exception {
      ClientProducer producer = clientSession.createProducer(address);
      ClientConsumer consumer = clientSession.createConsumer(qName1);
      ClientMessage m1 = createTextMessage(clientSession, "m1");
      SimpleString rh = SimpleString.of("SMID1");
      m1.putStringProperty(Message.HDR_LAST_VALUE_NAME, rh);
      ClientMessage m2 = createTextMessage(clientSession, "m2");
      m2.putStringProperty(Message.HDR_LAST_VALUE_NAME, rh);
      producer.send(m1);
      clientSession.start();
      ClientMessage m = consumer.receive(1000);
      assertNotNull(m);
      producer.send(m2);
      m.acknowledge();
      assertEquals("m1", m.getBodyBuffer().readString());
      m = consumer.receive(1000);
      assertNotNull(m);
      m.acknowledge();
      assertEquals("m2", m.getBodyBuffer().readString());
   }

   @Test
   public void testFirstMessageReceivedAndCancelled() throws Exception {
      ClientProducer producer = clientSession.createProducer(address);
      ClientConsumer consumer = clientSession.createConsumer(qName1);
      ClientMessage m1 = createTextMessage(clientSession, "m1");
      SimpleString rh = SimpleString.of("SMID1");
      m1.putStringProperty(Message.HDR_LAST_VALUE_NAME, rh);
      ClientMessage m2 = createTextMessage(clientSession, "m2");
      m2.putStringProperty(Message.HDR_LAST_VALUE_NAME, rh);
      producer.send(m1);
      clientSession.start();
      ClientMessage m = consumer.receive(1000);
      assertNotNull(m);
      assertEquals("m1", m.getBodyBuffer().readString());
      producer.send(m2);
      consumer.close();
      consumer = clientSession.createConsumer(qName1);
      m = consumer.receive(1000);
      assertNotNull(m);
      assertEquals("m2", m.getBodyBuffer().readString());
      m.acknowledge();
      m = consumer.receiveImmediate();
      assertNull(m);
   }

   @Test
   public void testManyMessagesReceivedAndCancelled() throws Exception {
      ClientProducer producer = clientSession.createProducer(address);
      ClientConsumer consumer = clientSession.createConsumer(qName1);

      SimpleString rh = SimpleString.of("SMID1");
      ClientMessage m1 = createTextMessage(clientSession, "m1");
      m1.putStringProperty(Message.HDR_LAST_VALUE_NAME, rh);
      ClientMessage m2 = createTextMessage(clientSession, "m2");
      m2.putStringProperty(Message.HDR_LAST_VALUE_NAME, rh);
      ClientMessage m3 = createTextMessage(clientSession, "m3");
      m3.putStringProperty(Message.HDR_LAST_VALUE_NAME, rh);
      ClientMessage m4 = createTextMessage(clientSession, "m4");
      m4.putStringProperty(Message.HDR_LAST_VALUE_NAME, rh);
      ClientMessage m5 = createTextMessage(clientSession, "m5");
      m5.putStringProperty(Message.HDR_LAST_VALUE_NAME, rh);
      ClientMessage m6 = createTextMessage(clientSession, "m6");
      m6.putStringProperty(Message.HDR_LAST_VALUE_NAME, rh);
      clientSession.start();
      producer.send(m1);
      ClientMessage m = consumer.receive(1000);
      assertNotNull(m);
      assertEquals("m1", m.getBodyBuffer().readString());
      producer.send(m2);
      m = consumer.receive(1000);
      assertNotNull(m);
      assertEquals("m2", m.getBodyBuffer().readString());
      producer.send(m3);
      m = consumer.receive(1000);
      assertNotNull(m);
      assertEquals("m3", m.getBodyBuffer().readString());
      producer.send(m4);
      m = consumer.receive(1000);
      assertNotNull(m);
      assertEquals("m4", m.getBodyBuffer().readString());
      producer.send(m5);
      m = consumer.receive(1000);
      assertNotNull(m);
      assertEquals("m5", m.getBodyBuffer().readString());
      producer.send(m6);
      m = consumer.receive(1000);
      assertNotNull(m);
      assertEquals("m6", m.getBodyBuffer().readString());
      consumer.close();
      consumer = clientSession.createConsumer(qName1);
      m = consumer.receive(1000);
      assertNotNull(m);
      m.acknowledge();
      assertEquals("m6", m.getBodyBuffer().readString());
      m = consumer.receiveImmediate();
      assertNull(m);
   }

   @Test
   public void testSimpleInTx() throws Exception {

      ClientProducer producer = clientSessionTxReceives.createProducer(address);
      ClientConsumer consumer = clientSessionTxReceives.createConsumer(qName1);
      ClientMessage m1 = createTextMessage(clientSession, "m1");
      SimpleString rh = SimpleString.of("SMID1");
      m1.putStringProperty(Message.HDR_LAST_VALUE_NAME, rh);
      ClientMessage m2 = createTextMessage(clientSession, "m2");
      m2.putStringProperty(Message.HDR_LAST_VALUE_NAME, rh);
      producer.send(m1);
      producer.send(m2);
      clientSessionTxReceives.start();
      ClientMessage m = consumer.receive(1000);
      assertNotNull(m);
      m.acknowledge();
      assertEquals("m2", m.getBodyBuffer().readString());
   }

   @Test
   public void testMultipleMessagesInTx() throws Exception {
      ClientProducer producer = clientSessionTxReceives.createProducer(address);
      SimpleString messageId1 = SimpleString.of("SMID1");
      SimpleString messageId2 = SimpleString.of("SMID2");
      ClientMessage m1 = createTextMessage(clientSession, "m1");
      m1.putStringProperty(Message.HDR_LAST_VALUE_NAME, messageId1);
      ClientMessage m2 = createTextMessage(clientSession, "m2");
      m2.putStringProperty(Message.HDR_LAST_VALUE_NAME, messageId2);
      ClientMessage m3 = createTextMessage(clientSession, "m3");
      m3.putStringProperty(Message.HDR_LAST_VALUE_NAME, messageId1);
      ClientMessage m4 = createTextMessage(clientSession, "m4");
      m4.putStringProperty(Message.HDR_LAST_VALUE_NAME, messageId2);
      producer.send(m1);
      producer.send(m2);
      producer.send(m3);
      producer.send(m4);
      ClientConsumer consumer = clientSessionTxReceives.createConsumer(qName1);
      clientSessionTxReceives.start();
      ClientMessage m = consumer.receive(1000);
      assertNotNull(m);
      m.acknowledge();
      assertEquals("m3", m.getBodyBuffer().readString());
      m = consumer.receive(1000);
      assertNotNull(m);
      m.acknowledge();
      assertEquals("m4", m.getBodyBuffer().readString());
      clientSessionTxReceives.commit();
      m = consumer.receiveImmediate();
      assertNull(m);
   }

   @Test
   public void testMultipleMessagesInTxRollback() throws Exception {
      ClientProducer producer = clientSessionTxReceives.createProducer(address);
      ClientConsumer consumer = clientSessionTxReceives.createConsumer(qName1);
      SimpleString messageId1 = SimpleString.of("SMID1");
      SimpleString messageId2 = SimpleString.of("SMID2");
      ClientMessage m1 = createTextMessage(clientSession, "m1");
      m1.putStringProperty(Message.HDR_LAST_VALUE_NAME, messageId1);
      ClientMessage m2 = createTextMessage(clientSession, "m2");
      m2.putStringProperty(Message.HDR_LAST_VALUE_NAME, messageId2);
      ClientMessage m3 = createTextMessage(clientSession, "m3");
      m3.putStringProperty(Message.HDR_LAST_VALUE_NAME, messageId1);
      ClientMessage m4 = createTextMessage(clientSession, "m4");
      m4.putStringProperty(Message.HDR_LAST_VALUE_NAME, messageId2);
      producer.send(m1);
      producer.send(m2);
      clientSessionTxReceives.start();
      ClientMessage m = consumer.receive(1000);
      assertNotNull(m);
      m.acknowledge();
      assertEquals("m1", m.getBodyBuffer().readString());
      m = consumer.receive(1000);
      assertNotNull(m);
      m.acknowledge();
      assertEquals("m2", m.getBodyBuffer().readString());
      producer.send(m3);
      producer.send(m4);
      m = consumer.receive(1000);
      assertNotNull(m);
      m.acknowledge();
      assertEquals("m3", m.getBodyBuffer().readString());
      m = consumer.receive(1000);
      assertNotNull(m);
      m.acknowledge();
      assertEquals("m4", m.getBodyBuffer().readString());
      clientSessionTxReceives.rollback();
      m = consumer.receive(1000);
      assertNotNull(m);
      m.acknowledge();
      assertEquals("m3", m.getBodyBuffer().readString());
      m = consumer.receive(1000);
      assertNotNull(m);
      m.acknowledge();
      assertEquals("m4", m.getBodyBuffer().readString());
   }

   @Test
   public void testSingleTXRollback() throws Exception {
      ClientProducer producer = clientSessionTxReceives.createProducer(address);
      ClientConsumer consumer = clientSessionTxReceives.createConsumer(qName1);
      SimpleString messageId1 = SimpleString.of("SMID1");
      ClientMessage m1 = createTextMessage(clientSession, "m1");
      m1.putStringProperty(Message.HDR_LAST_VALUE_NAME, messageId1);
      producer.send(m1);
      clientSessionTxReceives.start();
      ClientMessage m = consumer.receive(1000);
      assertNotNull(m);
      m.acknowledge();
      clientSessionTxReceives.rollback();
      m = consumer.receive(1000);
      assertNotNull(m);
      m.acknowledge();
      assertEquals("m1", m.getBodyBuffer().readString());
      assertNull(consumer.receiveImmediate());
   }

   @Test
   public void testMultipleMessagesInTxSend() throws Exception {
      ClientProducer producer = clientSessionTxSends.createProducer(address);
      ClientConsumer consumer = clientSessionTxSends.createConsumer(qName1);
      clientSessionTxSends.start();
      SimpleString rh = SimpleString.of("SMID1");
      ClientMessage m1 = createTextMessage(clientSession, "m1");
      m1.putStringProperty(Message.HDR_LAST_VALUE_NAME, rh);
      ClientMessage m2 = createTextMessage(clientSession, "m2");
      m2.putStringProperty(Message.HDR_LAST_VALUE_NAME, rh);
      ClientMessage m3 = createTextMessage(clientSession, "m3");
      m3.putStringProperty(Message.HDR_LAST_VALUE_NAME, rh);
      ClientMessage m4 = createTextMessage(clientSession, "m4");
      m4.putStringProperty(Message.HDR_LAST_VALUE_NAME, rh);
      ClientMessage m5 = createTextMessage(clientSession, "m5");
      m5.putStringProperty(Message.HDR_LAST_VALUE_NAME, rh);
      ClientMessage m6 = createTextMessage(clientSession, "m6");
      m6.putStringProperty(Message.HDR_LAST_VALUE_NAME, rh);
      producer.send(m1);
      producer.send(m2);
      producer.send(m3);
      producer.send(m4);
      producer.send(m5);
      producer.send(m6);
      clientSessionTxSends.commit();
      for (int i = 1; i < 6; i++) {
         ClientMessage m = consumer.receive(1000);
         assertNotNull(m);
         m.acknowledge();
         assertEquals("m" + i, m.getBodyBuffer().readString());
      }
      consumer.close();
      consumer = clientSessionTxSends.createConsumer(qName1);
      ClientMessage m = consumer.receive(1000);
      assertNotNull(m);
      m.acknowledge();
      assertEquals("m6", m.getBodyBuffer().readString());
   }

   @Test
   public void testMultipleMessagesPersistedCorrectly() throws Exception {
      ClientProducer producer = clientSession.createProducer(address);
      ClientConsumer consumer = clientSession.createConsumer(qName1);
      SimpleString rh = SimpleString.of("SMID1");
      ClientMessage m1 = createTextMessage(clientSession, "m1");
      m1.putStringProperty(Message.HDR_LAST_VALUE_NAME, rh);
      m1.setDurable(true);
      ClientMessage m2 = createTextMessage(clientSession, "m2");
      m2.putStringProperty(Message.HDR_LAST_VALUE_NAME, rh);
      m2.setDurable(true);
      ClientMessage m3 = createTextMessage(clientSession, "m3");
      m3.putStringProperty(Message.HDR_LAST_VALUE_NAME, rh);
      m3.setDurable(true);
      ClientMessage m4 = createTextMessage(clientSession, "m4");
      m4.putStringProperty(Message.HDR_LAST_VALUE_NAME, rh);
      m4.setDurable(true);
      ClientMessage m5 = createTextMessage(clientSession, "m5");
      m5.putStringProperty(Message.HDR_LAST_VALUE_NAME, rh);
      m5.setDurable(true);
      ClientMessage m6 = createTextMessage(clientSession, "m6");
      m6.putStringProperty(Message.HDR_LAST_VALUE_NAME, rh);
      m6.setDurable(true);
      producer.send(m1);
      producer.send(m2);
      producer.send(m3);
      producer.send(m4);
      producer.send(m5);
      producer.send(m6);
      clientSession.start();
      ClientMessage m = consumer.receive(1000);
      assertNotNull(m);
      m.acknowledge();
      assertEquals("m6", m.getBodyBuffer().readString());
      m = consumer.receiveImmediate();
      assertNull(m);
   }

   @Test
   public void testMultipleMessagesPersistedCorrectlyInTx() throws Exception {
      ClientProducer producer = clientSessionTxSends.createProducer(address);
      SimpleString rh = SimpleString.of("SMID1");
      ClientMessage m1 = createTextMessage(clientSession, "m1");
      m1.putStringProperty(Message.HDR_LAST_VALUE_NAME, rh);
      m1.setDurable(true);
      ClientMessage m2 = createTextMessage(clientSession, "m2");
      m2.putStringProperty(Message.HDR_LAST_VALUE_NAME, rh);
      m2.setDurable(true);
      ClientMessage m3 = createTextMessage(clientSession, "m3");
      m3.putStringProperty(Message.HDR_LAST_VALUE_NAME, rh);
      m3.setDurable(true);
      ClientMessage m4 = createTextMessage(clientSession, "m4");
      m4.putStringProperty(Message.HDR_LAST_VALUE_NAME, rh);
      m4.setDurable(true);
      ClientMessage m5 = createTextMessage(clientSession, "m5");
      m5.putStringProperty(Message.HDR_LAST_VALUE_NAME, rh);
      m5.setDurable(true);
      ClientMessage m6 = createTextMessage(clientSession, "m6");
      m6.putStringProperty(Message.HDR_LAST_VALUE_NAME, rh);
      m6.setDurable(true);
      producer.send(m1);
      producer.send(m2);
      producer.send(m3);
      producer.send(m4);
      producer.send(m5);
      producer.send(m6);
      clientSessionTxSends.commit();
      clientSessionTxSends.start();
      ClientConsumer consumer = clientSessionTxSends.createConsumer(qName1);
      ClientMessage m = consumer.receive(1000);
      assertNotNull(m);
      m.acknowledge();
      assertEquals("m6", m.getBodyBuffer().readString());
      m = consumer.receiveImmediate();
      assertNull(m);
   }

   @Test
   public void testMultipleAcksPersistedCorrectly() throws Exception {

      Queue queue = server.locateQueue(qName1);
      ClientProducer producer = clientSession.createProducer(address);
      ClientConsumer consumer = clientSession.createConsumer(qName1);
      SimpleString rh = SimpleString.of("SMID1");
      ClientMessage m1 = createTextMessage(clientSession, "m1");
      m1.putStringProperty(Message.HDR_LAST_VALUE_NAME, rh);
      m1.setDurable(true);
      ClientMessage m2 = createTextMessage(clientSession, "m2");
      m2.putStringProperty(Message.HDR_LAST_VALUE_NAME, rh);
      m2.setDurable(true);
      ClientMessage m3 = createTextMessage(clientSession, "m3");
      m3.putStringProperty(Message.HDR_LAST_VALUE_NAME, rh);
      m3.setDurable(true);
      ClientMessage m4 = createTextMessage(clientSession, "m4");
      m4.putStringProperty(Message.HDR_LAST_VALUE_NAME, rh);
      m4.setDurable(true);
      ClientMessage m5 = createTextMessage(clientSession, "m5");
      m5.putStringProperty(Message.HDR_LAST_VALUE_NAME, rh);
      m5.setDurable(true);
      ClientMessage m6 = createTextMessage(clientSession, "m6");
      m6.putStringProperty(Message.HDR_LAST_VALUE_NAME, rh);
      m6.setDurable(true);
      clientSession.start();
      producer.send(m1);
      ClientMessage m = consumer.receive(1000);
      assertNotNull(m);
      m.acknowledge();
      assertEquals("m1", m.getBodyBuffer().readString());
      producer.send(m2);
      m = consumer.receive(1000);
      assertNotNull(m);
      m.acknowledge();
      assertEquals("m2", m.getBodyBuffer().readString());
      producer.send(m3);
      m = consumer.receive(1000);
      assertNotNull(m);
      m.acknowledge();
      assertEquals("m3", m.getBodyBuffer().readString());
      producer.send(m4);
      m = consumer.receive(1000);
      assertNotNull(m);
      m.acknowledge();
      assertEquals("m4", m.getBodyBuffer().readString());
      producer.send(m5);
      m = consumer.receive(1000);
      assertNotNull(m);
      m.acknowledge();
      assertEquals("m5", m.getBodyBuffer().readString());
      producer.send(m6);
      m = consumer.receive(1000);
      assertNotNull(m);
      m.acknowledge();
      assertEquals("m6", m.getBodyBuffer().readString());

      assertEquals(0, queue.getDeliveringCount());
   }

   @Test
   public void testRemoveMessageThroughManagement() throws Exception {

      Queue queue = server.locateQueue(qName1);
      ClientProducer producer = clientSession.createProducer(address);
      ClientConsumer consumer = clientSession.createConsumer(qName1);
      SimpleString rh = SimpleString.of("SMID1");
      ClientMessage m1 = createTextMessage(clientSession, "m1");
      m1.putStringProperty(Message.HDR_LAST_VALUE_NAME, rh);
      m1.setDurable(true);
      producer.send(m1);

      queue.deleteAllReferences();

      producer.send(m1);

      clientSession.start();
      ClientMessage m = consumer.receive(1000);
      assertNotNull(m);
      m.acknowledge();
      assertEquals("m1", m.getBodyBuffer().readString());

      assertEquals(0, queue.getDeliveringCount());
   }

   @Test
   public void testScheduledMessages() throws Exception {
      final long DELAY_TIME = 100;
      final int MESSAGE_COUNT = 5;
      Queue queue = server.locateQueue(qName1);
      ClientProducer producer = clientSession.createProducer(address);
      ClientConsumer consumer = clientSession.createConsumer(qName1);
      SimpleString rh = SimpleString.of("SMID1");
      long timeSent = 0;
      for (int i = 0; i < MESSAGE_COUNT; i++) {
         ClientMessage m = createTextMessage(clientSession, "m" + i);
         m.setDurable(true);
         m.putStringProperty(Message.HDR_LAST_VALUE_NAME, rh);
         timeSent = System.currentTimeMillis();
         m.putLongProperty(Message.HDR_SCHEDULED_DELIVERY_TIME, timeSent + (i * DELAY_TIME));
         producer.send(m);
      }

      // allow schedules to elapse so the messages will be delivered to the queue
      Wait.waitFor(() -> queue.getScheduledCount() == 0);

      Wait.assertEquals((long) MESSAGE_COUNT, queue::getMessagesAdded, 5000, 100);
      Wait.assertEquals((long) (MESSAGE_COUNT - 1), queue::getMessagesReplaced, 5000, 100);
      queue.flushExecutor();
      CountDownLatch latch = new CountDownLatch(1);
      queue.flushOnIntermediate(latch::countDown);
      assertTrue(latch.await(10, TimeUnit.SECONDS));

      clientSession.start();
      ClientMessage m = consumer.receive(5000);
      assertNotNull(m);
      assertEquals("m" + (MESSAGE_COUNT - 1), m.getBodyBuffer().readString());
      assertEquals(0, queue.getScheduledCount());
   }

   @Test
   public void testMultipleAcksPersistedCorrectly2() throws Exception {

      Queue queue = server.locateQueue(qName1);
      ClientProducer producer = clientSession.createProducer(address);
      ClientConsumer consumer = clientSession.createConsumer(qName1);
      SimpleString rh = SimpleString.of("SMID1");
      ClientMessage m1 = createTextMessage(clientSession, "m1");
      m1.putStringProperty(Message.HDR_LAST_VALUE_NAME, rh);
      m1.setDurable(true);
      ClientMessage m2 = createTextMessage(clientSession, "m2");
      m2.putStringProperty(Message.HDR_LAST_VALUE_NAME, rh);
      m2.setDurable(true);
      clientSession.start();
      producer.send(m1);
      ClientMessage m = consumer.receive(1000);
      assertNotNull(m);
      m.acknowledge();
      assertEquals("m1", m.getBodyBuffer().readString());
      producer.send(m2);
      m = consumer.receive(1000);
      assertNotNull(m);
      m.acknowledge();
      assertEquals("m2", m.getBodyBuffer().readString());

      assertEquals(0, queue.getDeliveringCount());
   }

   @Test
   public void testMultipleAcksPersistedCorrectlyInTx() throws Exception {
      ClientProducer producer = clientSessionTxReceives.createProducer(address);
      ClientConsumer consumer = clientSessionTxReceives.createConsumer(qName1);
      SimpleString rh = SimpleString.of("SMID1");
      ClientMessage m1 = createTextMessage(clientSession, "m1");
      m1.putStringProperty(Message.HDR_LAST_VALUE_NAME, rh);
      m1.setDurable(true);
      ClientMessage m2 = createTextMessage(clientSession, "m2");
      m2.putStringProperty(Message.HDR_LAST_VALUE_NAME, rh);
      m2.setDurable(true);
      ClientMessage m3 = createTextMessage(clientSession, "m3");
      m3.putStringProperty(Message.HDR_LAST_VALUE_NAME, rh);
      m3.setDurable(true);
      ClientMessage m4 = createTextMessage(clientSession, "m4");
      m4.putStringProperty(Message.HDR_LAST_VALUE_NAME, rh);
      m4.setDurable(true);
      ClientMessage m5 = createTextMessage(clientSession, "m5");
      m5.putStringProperty(Message.HDR_LAST_VALUE_NAME, rh);
      m5.setDurable(true);
      ClientMessage m6 = createTextMessage(clientSession, "m6");
      m6.putStringProperty(Message.HDR_LAST_VALUE_NAME, rh);
      m6.setDurable(true);
      clientSessionTxReceives.start();
      producer.send(m1);
      ClientMessage m = consumer.receive(1000);
      assertNotNull(m);
      m.acknowledge();
      assertEquals("m1", m.getBodyBuffer().readString());
      producer.send(m2);
      m = consumer.receive(1000);
      assertNotNull(m);
      m.acknowledge();
      assertEquals("m2", m.getBodyBuffer().readString());
      producer.send(m3);
      m = consumer.receive(1000);
      assertNotNull(m);
      m.acknowledge();
      assertEquals("m3", m.getBodyBuffer().readString());
      producer.send(m4);
      m = consumer.receive(1000);
      assertNotNull(m);
      m.acknowledge();
      assertEquals("m4", m.getBodyBuffer().readString());
      producer.send(m5);
      m = consumer.receive(1000);
      assertNotNull(m);
      m.acknowledge();
      assertEquals("m5", m.getBodyBuffer().readString());
      producer.send(m6);
      m = consumer.receive(1000);
      assertNotNull(m);
      m.acknowledge();
      assertEquals("m6", m.getBodyBuffer().readString());
      clientSessionTxReceives.commit();
   }

   @Test
   public void testLargeMessage() throws Exception {
      ClientProducer producer = clientSessionTxReceives.createProducer(address);
      SimpleString rh = SimpleString.of("SMID1");

      for (int i = 0; i < 5; i++) {
         ClientMessage message = clientSession.createMessage(true);
         message.setBodyInputStream(createFakeLargeStream(300 * 1024));
         message.putStringProperty(Message.HDR_LAST_VALUE_NAME, rh);
         producer.send(message);
         clientSession.commit();
      }
      ClientConsumer consumer = clientSessionTxReceives.createConsumer(qName1);
      clientSessionTxReceives.start();
      ClientMessage m = consumer.receive(1000);
      assertNotNull(m);
      m.acknowledge();
      assertNull(consumer.receiveImmediate());
      clientSessionTxReceives.commit();
   }

   @Test
   public void testSizeInReplace() throws Exception {
      ClientProducer producer = clientSession.createProducer(address);
      ClientMessage m1 = createTextMessage(clientSession, "m1");
      SimpleString rh = SimpleString.of("SMID1");
      m1.putStringProperty(Message.HDR_LAST_VALUE_NAME, rh);

      ClientMessage m2 = clientSession.createMessage(true);
      m2.setBodyInputStream(createFakeLargeStream(10 * 1024));
      m2.putStringProperty(Message.HDR_LAST_VALUE_NAME, rh);

      Queue queue = server.locateQueue(qName1);
      producer.send(m1);
      Wait.assertEquals(123, () -> queue.getPersistentSize());
      producer.send(m2);
      // encoded size is a little larger than payload
      Wait.assertTrue(() -> queue.getPersistentSize() > 10 * 1024);
      Wait.assertEquals(0L,  () -> queue.getDeliveringSize(), 5000, 100);
   }

   @Test
   public void testDeleteReference() throws Exception {
      ClientProducer producer = clientSession.createProducer(address);
      ClientMessage m1 = createTextMessage(clientSession, "m1");
      SimpleString rh = SimpleString.of("SMID1");
      m1.putStringProperty(Message.HDR_LAST_VALUE_NAME, rh);
      m1.setBodyInputStream(createFakeLargeStream(2 * 1024));

      ClientMessage m2 = clientSession.createMessage(true);
      m2.setBodyInputStream(createFakeLargeStream(10 * 1024));
      m2.putStringProperty(Message.HDR_LAST_VALUE_NAME, rh);

      Queue queue = server.locateQueue(qName1);
      producer.send(m1);
      LinkedListIterator<MessageReference> browserIterator = queue.browserIterator();
      // Wait for message delivered to queue
      Wait.assertTrue(() -> browserIterator.hasNext(), 10_000, 2);
      long messageId = browserIterator.next().getMessage().getMessageID();
      browserIterator.close();

      queue.deleteReference(messageId);
      // Wait for delete tx's afterCommit called
      Wait.assertEquals(0L, () -> queue.getDeliveringSize(), 10_000, 2);
      assertEquals(0, queue.getPersistentSize());
      assertTrue(((LastValueQueue)queue).getLastValueKeys().isEmpty());

      producer.send(m2);
      // Wait for message delivered to queue
      Wait.assertTrue(() -> queue.getPersistentSize() > 10 * 1024, 10_000, 2);
      assertEquals(0, queue.getDeliveringSize());
   }

   @Test
   public void testChangeReferencePriority() throws Exception {
      ClientProducer producer = clientSession.createProducer(address);
      ClientMessage m1 = createTextMessage(clientSession, "m1");
      SimpleString rh = SimpleString.of("SMID1");
      m1.putStringProperty(Message.HDR_LAST_VALUE_NAME, rh);

      Queue queue = server.locateQueue(qName1);
      producer.send(m1);
      Wait.assertEquals(1, queue::getMessageCount);
      LinkedListIterator<MessageReference> browserIterator = queue.browserIterator();
      // Wait for message delivered to queue
      assertTrue(browserIterator.hasNext());
      long messageId = browserIterator.next().getMessage().getMessageID();
      browserIterator.close();
      long oldSize = queue.getPersistentSize();

      assertTrue(queue.changeReferencePriority(messageId, (byte) 1));
      // Wait for message delivered to queue
      Wait.assertEquals(oldSize, () -> queue.getPersistentSize(), 10_000, 2);
      assertEquals(0, queue.getDeliveringSize());
   }

   @Test
   public void testConcurrency() throws Exception {
      AtomicBoolean cme = new AtomicBoolean(false);

      AtomicBoolean hash = new AtomicBoolean(true);
      Queue lvq = server.locateQueue(qName1);
      Thread hashCodeThread = new Thread(() -> {
         while (hash.get()) {
            try {
               int hashCode = lvq.hashCode();
            } catch (ConcurrentModificationException e) {
               cme.set(true);
               return;
            }
         }
      });
      hashCodeThread.start();

      AtomicBoolean consume = new AtomicBoolean(true);
      ClientConsumer consumer = clientSessionTxReceives.createConsumer(qName1);
      clientSessionTxReceives.start();
      Thread consumerThread = new Thread(() -> {
         while (consume.get()) {
            try {
               ClientMessage m = consumer.receive();
               m.acknowledge();
               clientSessionTxReceives.commit();
            } catch (ActiveMQException e) {
               e.printStackTrace();
               return;
            }
         }
      });
      consumerThread.start();

      ClientProducer producer = clientSessionTxSends.createProducer(address);
      SimpleString lastValue = RandomUtil.randomUUIDSimpleString();
      AtomicBoolean produce = new AtomicBoolean(true);
      Thread producerThread = new Thread(() -> {
         for (int i = 0; !cme.get() && produce.get(); i++) {
            ClientMessage m = createTextMessage(clientSession, "m" + i, false);
            m.putStringProperty(Message.HDR_LAST_VALUE_NAME, lastValue);
            try {
               producer.send(m);
               clientSessionTxSends.commit();
            } catch (ActiveMQException e) {
               e.printStackTrace();
               return;
            }
         }
      });
      producerThread.start();
      producerThread.join(500);

      try {
         assertFalse(cme.get());
      } finally {
         produce.set(false);
         producerThread.join();
         consume.set(false);
         consumerThread.join();
         hash.set(false);
         hashCodeThread.join();
      }
   }

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

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

      server.getAddressSettingsRepository().addMatch(address.toString(), new AddressSettings().setDefaultLastValueQueue(true));
      // then we create a client as normalServer
      ServerLocator locator = createNettyNonHALocator().setBlockOnAcknowledge(true).setAckBatchSize(0);

      ClientSessionFactory sf = createSessionFactory(locator);
      clientSession = addClientSession(sf.createSession(false, true, true));
      clientSessionTxReceives = addClientSession(sf.createSession(false, true, false));
      clientSessionTxSends = addClientSession(sf.createSession(false, false, true));
      clientSession.createQueue(QueueConfiguration.of(qName1).setAddress(address));
   }
}
