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

import static org.junit.jupiter.api.Assertions.assertTrue;

import java.lang.invoke.MethodHandles;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.TimeUnit;

import org.apache.activemq.artemis.api.core.ActiveMQNotConnectedException;
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.MessageHandler;
import org.apache.activemq.artemis.api.core.client.ServerLocator;
import org.apache.activemq.artemis.core.client.impl.ClientSessionInternal;
import org.apache.activemq.artemis.core.protocol.core.impl.RemotingConnectionImpl;
import org.apache.activemq.artemis.core.server.ActiveMQServer;
import org.apache.activemq.artemis.jms.client.ActiveMQTextMessage;
import org.apache.activemq.artemis.tests.util.ActiveMQTestBase;
import org.junit.jupiter.api.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class OrderReattachTest extends ActiveMQTestBase {

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

   final SimpleString ADDRESS = SimpleString.of("address");

   private ActiveMQServer server;



   @Test
   public void testOrderOnSendInVM() throws Throwable {
      doTestOrderOnSend(false);
   }

   public void doTestOrderOnSend(final boolean isNetty) throws Throwable {
      server = createServer(false, isNetty);

      server.start();
      ServerLocator locator = createFactory(isNetty).setReconnectAttempts(15).setConfirmationWindowSize(1024 * 1024).setBlockOnNonDurableSend(false).setBlockOnAcknowledge(false);

      ClientSessionFactory sf = createSessionFactory(locator);

      final ClientSession session = sf.createSession(false, true, true);

      final LinkedBlockingDeque<Boolean> failureQueue = new LinkedBlockingDeque<>();

      final CountDownLatch ready = new CountDownLatch(1);

      // this test will use a queue. Whenever the test wants a failure.. it can just send TRUE to failureQueue
      // This Thread will be reading the queue
      Thread failer = new Thread(() -> {
         ready.countDown();
         while (true) {
            try {
               Boolean poll = false;
               try {
                  poll = failureQueue.poll(60, TimeUnit.SECONDS);
               } catch (InterruptedException e) {
                  e.printStackTrace();
                  break;
               }

               Thread.sleep(1);

               final RemotingConnectionImpl conn = (RemotingConnectionImpl) ((ClientSessionInternal) session).getConnection();

               // True means... fail session
               if (poll) {
                  conn.fail(new ActiveMQNotConnectedException("poop"));
               } else {
                  // false means... finish thread
                  break;
               }
            } catch (Exception e) {
               e.printStackTrace();
            }
         }
      });

      failer.start();

      ready.await();

      try {
         doSend2(1, sf, failureQueue);
      } finally {
         try {
            session.close();
         } catch (Exception e) {
            e.printStackTrace();
         }

         try {
            locator.close();
         } catch (Exception e) {
            //
         }
         try {
            sf.close();
         } catch (Exception e) {
            e.printStackTrace();
         }

         failureQueue.put(false);

         failer.join();
      }

   }

   public void doSend2(final int order,
                       final ClientSessionFactory sf,
                       final LinkedBlockingDeque<Boolean> failureQueue) throws Exception {
      ClientSession s = sf.createSession(false, false, false);

      final int numMessages = 500;

      final int numSessions = 10;

      Set<ClientConsumer> consumers = new HashSet<>();
      Set<ClientSession> sessions = new HashSet<>();

      for (int i = 0; i < numSessions; i++) {
         SimpleString subName = SimpleString.of("sub" + i);

         // failureQueue.push(true);

         ClientSession sessConsume = sf.createSession(false, true, true);

         sessConsume.createQueue(QueueConfiguration.of(subName).setAddress(ADDRESS).setDurable(false));

         ClientConsumer consumer = sessConsume.createConsumer(subName);

         consumers.add(consumer);

         sessions.add(sessConsume);
      }

      ClientSession sessSend = sf.createSession(false, true, true);

      ClientProducer producer = sessSend.createProducer(ADDRESS);

      for (int i = 0; i < numMessages; i++) {
         ClientMessage message = sessSend.createMessage(ActiveMQTextMessage.TYPE, false, 0, System.currentTimeMillis(), (byte) 1);

         if (i % 10 == 0) {
            // failureQueue.push(true);
         }
         message.putIntProperty(SimpleString.of("count"), i);
         producer.send(message);
      }

      for (ClientSession session : sessions) {
         session.start();
      }

      class MyHandler implements MessageHandler {

         final CountDownLatch latch = new CountDownLatch(1);

         int count;

         Exception failure;

         @Override
         public void onMessage(final ClientMessage message) {
            if (count >= numMessages) {
               failure = new Exception("too many messages");
               latch.countDown();
            }

            if (message.getIntProperty("count") != count) {
               failure = new Exception("counter " + count + " was not as expected (" + message.getIntProperty("count") + ")");
               logger.warn("Failure on receiving message ", failure);
               failure.printStackTrace();
               latch.countDown();
            }

            count++;

            if (count % 100 == 0) {
               failureQueue.push(true);
            }

            if (count == numMessages) {
               latch.countDown();
            }
         }
      }

      Set<MyHandler> handlers = new HashSet<>();

      for (ClientConsumer consumer : consumers) {
         MyHandler handler = new MyHandler();

         consumer.setMessageHandler(handler);

         handlers.add(handler);
      }

      for (MyHandler handler : handlers) {
         boolean ok = handler.latch.await(60000, TimeUnit.MILLISECONDS);

         assertTrue(ok);

         if (handler.failure != null) {
            throw handler.failure;
         }
      }

      // failureQueue.push(true);

      sessSend.close();

      for (ClientSession session : sessions) {
         // failureQueue.push(true);
         session.close();
      }

      for (int i = 0; i < numSessions; i++) {

         failureQueue.push(true);

         SimpleString subName = SimpleString.of("sub" + i);

         s.deleteQueue(subName);
      }

      s.close();
   }
}
