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

import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.broker.BrokerService;
import org.apache.activemq.broker.region.RegionBroker;
import org.apache.activemq.broker.region.policy.PolicyEntry;
import org.apache.activemq.broker.region.policy.PolicyMap;
import org.apache.activemq.command.ActiveMQDestination;
import org.apache.activemq.command.ActiveMQQueue;
import org.apache.activemq.store.PersistenceAdapter;
import org.apache.activemq.store.kahadb.KahaDBStore;
import org.apache.activemq.usage.MemoryUsage;
import org.apache.activemq.usage.StoreUsage;
import org.apache.activemq.usage.SystemUsage;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import jakarta.jms.Connection;
import jakarta.jms.Destination;
import jakarta.jms.Message;
import jakarta.jms.MessageConsumer;
import jakarta.jms.MessageProducer;
import jakarta.jms.Session;
import jakarta.jms.TextMessage;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;

public class DuplicateFromStoreTest {
    private static final Logger LOG = LoggerFactory.getLogger(DuplicateFromStoreTest.class);
    private String activemqURL;
    private BrokerService broker;

    private static final String DESTNAME = "TEST";
    private static final int NUM_PRODUCERS = 100;
    private static final int NUM_CONSUMERS = 20;

    private static final int NUM_MSGS = 20000;
    private static final int CONSUMER_SLEEP = 0;
    private static final int PRODUCER_SLEEP = 10;

    private final CountDownLatch producersFinished = new CountDownLatch(NUM_PRODUCERS);
    private final CountDownLatch consumersFinished = new CountDownLatch(NUM_CONSUMERS);

    private final AtomicInteger totalMessagesToSend = new AtomicInteger(NUM_MSGS);
    private final AtomicInteger totalMessagesSent = new AtomicInteger(NUM_MSGS);

    private final AtomicInteger totalReceived = new AtomicInteger(0);


    @Before
    public void startBroker() throws Exception {

        broker = new BrokerService();
        broker.setDeleteAllMessagesOnStartup(true);
        broker.addConnector("tcp://0.0.0.0:0");

        // Create <policyEntry>
        final PolicyEntry policy = new PolicyEntry();
        final ActiveMQDestination dest = new ActiveMQQueue(">");
        policy.setDestination(dest);
        policy.setMemoryLimit(10 * 1024 * 1024); // 10 MB
        policy.setExpireMessagesPeriod(0);
        policy.setEnableAudit(false); // allow any duplicates from the store to bubble up to the q impl
        policy.setQueuePrefetch(100);
        final PolicyMap policies = new PolicyMap();
        policies.put(dest, policy);
        broker.setDestinationPolicy(policies);

        // configure <systemUsage>
        final MemoryUsage memoryUsage = new MemoryUsage();
        memoryUsage.setPercentOfJvmHeap(50);

        final StoreUsage storeUsage = new StoreUsage();
        storeUsage.setLimit(8 * 1024 * 1024 * 1024); // 8 gb

        final SystemUsage memoryManager = new SystemUsage();
        memoryManager.setMemoryUsage(memoryUsage);
        memoryManager.setStoreUsage(storeUsage);
        broker.setSystemUsage(memoryManager);

        // configure KahaDB persistence
        final PersistenceAdapter kahadb = new KahaDBStore();
        ((KahaDBStore) kahadb).setConcurrentStoreAndDispatchQueues(true);
        broker.setPersistenceAdapter(kahadb);

        // start broker
        broker.start();
        broker.waitUntilStarted();

        activemqURL = broker.getTransportConnectorByScheme("tcp").getPublishableConnectString();
    }

    @After
    public void stopBroker() throws Exception {
        if (broker != null) {
            broker.stop();
        }
    }

    @Test(timeout = 120_000)
    public void testDuplicateMessage() throws Exception {
        LOG.info("Testing for duplicate messages.");

        //create producer and consumer threads
        final ExecutorService producers = Executors.newFixedThreadPool(NUM_PRODUCERS);
        final ExecutorService consumers = Executors.newFixedThreadPool(NUM_CONSUMERS);

        createOpenwireClients(producers, consumers);

        LOG.info("All producers and consumers got started. Awaiting their termination");
        assertTrue("producers should finish", producersFinished.await(2, TimeUnit.MINUTES));
        LOG.info("All producers have terminated. remaining to send: " + totalMessagesToSend.get() + ", sent:" + totalMessagesSent.get());

        assertTrue("consumers should finish", consumersFinished.await(2, TimeUnit.MINUTES));
        LOG.info("All consumers have terminated.");

        producers.shutdownNow();
        consumers.shutdownNow();

        assertEquals("no messages pending, i.e. dlq empty", 0l, ((RegionBroker)broker.getRegionBroker()).getDestinationStatistics().getMessages().getCount());

        // validate cache can be enabled if disabled

    }


    protected void createOpenwireClients(final ExecutorService producers, final ExecutorService consumers) {
        for (int i = 0; i < NUM_CONSUMERS; i++) {
            LOG.trace("Creating consumer for destination " + DESTNAME);
            final Consumer consumer = new Consumer(DESTNAME, false);
            consumers.submit(consumer);
            // wait for consumer to signal it has fully initialized
            synchronized (consumer.init) {
                try {
                    consumer.init.wait();
                } catch (InterruptedException e) {
                    LOG.error(e.toString(), e);
                }
            }
        }

        for (int i = 0; i < NUM_PRODUCERS; i++) {
            LOG.trace("Creating producer for destination " + DESTNAME);
            final Producer producer = new Producer(DESTNAME, false, 0);
            producers.submit(producer);
        }
    }

    class Producer implements Runnable {

        private final String destName;
        private final boolean isTopicDest;

        public Producer(final String dest, final boolean isTopic, final int ttl) {
            this.destName = dest;
            this.isTopicDest = isTopic;
        }

        /**
         * Connect to broker and constantly send messages
         */
        public void run() {

            Connection connection = null;

            try {
                final ActiveMQConnectionFactory amq = new ActiveMQConnectionFactory(activemqURL);
                connection = amq.createConnection();

                connection.setExceptionListener(e -> e.printStackTrace());
                connection.start();

                // Create a Session
                final Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
                final Destination destination;
                if (isTopicDest) {
                    // Create the destination (Topic or Queue)
                    destination = session.createTopic(destName);
                } else {
                    destination = session.createQueue(destName);
                }
                // Create a MessageProducer from the Session to the Topic or Queue
                final MessageProducer producer = session.createProducer(destination);

                // Create message
                long counter = 0;
                //enlarge msg to 16 kb
                final int msgSize = 16 * 1024;
                final StringBuilder stringBuilder = new StringBuilder();
                stringBuilder.setLength(msgSize + 15);
                stringBuilder.append("Message: ");
                stringBuilder.append(counter);
                for (int j = 0; j < (msgSize / 10); j++) {
                    stringBuilder.append("XXXXXXXXXX");
                }
                final String text = stringBuilder.toString();
                final TextMessage message = session.createTextMessage(text);

                // send message
                while (totalMessagesToSend.decrementAndGet() >= 0) {
                    producer.send(message);
                    totalMessagesSent.incrementAndGet();
                    LOG.debug("Sent message: " + counter);
                    counter++;

                    if ((counter % 10000) == 0)
                        LOG.info("sent " + counter + " messages");

                    Thread.sleep(PRODUCER_SLEEP);
                }
            } catch (Exception ex) {
                LOG.error(ex.toString());
                return;
            } finally {
                try {
                    if (connection != null) {
                        connection.close();
                    }
                } catch (Exception ignored) {
                } finally {
                    producersFinished.countDown();
                }
            }
            LOG.debug("Closing producer for " + destName);
        }
    }

    class Consumer implements Runnable {

        final Object init = new Object();
        private final String queueName;
        private final boolean isTopic;

        public Consumer(final String destName, final boolean topic) {
            this.isTopic = topic;
            this.queueName = destName;
        }

        /**
         * connect to broker and receive messages
         */
        public void run() {

            Connection connection = null;

            try {
                final ActiveMQConnectionFactory amq = new ActiveMQConnectionFactory(activemqURL);
                connection = amq.createConnection();
                connection.setExceptionListener(e -> e.printStackTrace());
                connection.start();
                // Create a Session
                final Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
                // Create the destination (Topic or Queue)
                final Destination destination;
                if (isTopic)
                    destination = session.createTopic(queueName);
                else
                    destination = session.createQueue(queueName);

                //Create a MessageConsumer from the Session to the Topic or Queue
                final MessageConsumer consumer = session.createConsumer(destination);

                synchronized (init) {
                    init.notifyAll();
                }

                // Wait for a message
                long counter = 0;
                while (totalReceived.get() < NUM_MSGS) {
                    final Message message2 = consumer.receive(5000);

                    if (message2 instanceof TextMessage) {
                        final TextMessage textMessage = (TextMessage) message2;
                        final String text = textMessage.getText();
                        LOG.debug("Received: " + text.substring(0, 50));
                    } else if (totalReceived.get() < NUM_MSGS) {
                        LOG.error("Received message of unsupported type. Expecting TextMessage. count: " + totalReceived.get());
                    } else {
                        // all done
                        break;
                    }
                    if (message2 != null) {
                        counter++;
                        totalReceived.incrementAndGet();
                        if ((counter % 10000) == 0)
                            LOG.info("received " + counter + " messages");

                        Thread.sleep(CONSUMER_SLEEP);
                    }
                }
            } catch (Exception e) {
                LOG.error("Error in Consumer: " + e.getMessage());
                return;
            } finally {
                try {
                    if (connection != null) {
                        connection.close();
                    }
                } catch (Exception ignored) {
                } finally {
                    consumersFinished.countDown();
                }
            }
        }
    }
}
