/**
 * 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
 * <p>
 * http://www.apache.org/licenses/LICENSE-2.0
 * <p>
 * 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.amqp.connect;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.MessageConsumer;
import javax.jms.MessageProducer;
import javax.jms.Queue;
import javax.jms.Session;
import javax.jms.TextMessage;
import java.io.PrintStream;
import java.net.URI;
import java.util.concurrent.TimeUnit;

import org.apache.activemq.artemis.api.core.QueueConfiguration;
import org.apache.activemq.artemis.api.core.RoutingType;
import org.apache.activemq.artemis.cli.commands.tools.PrintData;
import org.apache.activemq.artemis.core.config.amqpBrokerConnectivity.AMQPBrokerConnectConfiguration;
import org.apache.activemq.artemis.core.config.amqpBrokerConnectivity.AMQPMirrorBrokerConnectionElement;
import org.apache.activemq.artemis.core.server.ActiveMQServer;
import org.apache.activemq.artemis.core.server.impl.AddressInfo;
import org.apache.activemq.artemis.logs.AssertionLoggerHandler;
import org.apache.activemq.artemis.tests.integration.amqp.AmqpClientTestSupport;
import org.apache.activemq.artemis.tests.util.CFUtil;
import org.apache.activemq.artemis.utils.StringPrintStream;
import org.apache.activemq.artemis.utils.Wait;
import org.apache.activemq.transport.amqp.client.AmqpClient;
import org.apache.activemq.transport.amqp.client.AmqpConnection;
import org.apache.activemq.transport.amqp.client.AmqpMessage;
import org.apache.activemq.transport.amqp.client.AmqpReceiver;
import org.apache.activemq.transport.amqp.client.AmqpSender;
import org.apache.activemq.transport.amqp.client.AmqpSession;
import org.jboss.logging.Logger;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;

public class BrokerInSyncTest extends AmqpClientTestSupport {

   public static final int TIME_BEFORE_RESTART = 1000;
   protected static final int AMQP_PORT_2 = 5673;
   protected static final int AMQP_PORT_3 = 5674;
   private static final Logger logger = Logger.getLogger(BrokerInSyncTest.class);
   ActiveMQServer server_2;

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

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

   @Before
   public void startLogging() {
      AssertionLoggerHandler.startCapture();
   }

   @After
   public void stopLogging() {
      try {
         Assert.assertFalse(AssertionLoggerHandler.findText("AMQ222214"));
      } finally {
         AssertionLoggerHandler.stopCapture();
      }
   }

   @Override
   protected ActiveMQServer createServer() throws Exception {
      return createServer(AMQP_PORT, false);
   }

   @Test
   public void testSyncOnCreateQueues() throws Exception {
      server.setIdentity("Server1");
      {
         AMQPBrokerConnectConfiguration amqpConnection = new AMQPBrokerConnectConfiguration("connectTowardsServer2", "tcp://localhost:" + AMQP_PORT_2).setReconnectAttempts(3).setRetryInterval(100);
         amqpConnection.addElement(new AMQPMirrorBrokerConnectionElement().setDurable(true));
         server.getConfiguration().addAMQPConnection(amqpConnection);
      }
      server.start();

      server_2 = createServer(AMQP_PORT_2, false);
      server_2.setIdentity("Server2");

      {
         AMQPBrokerConnectConfiguration amqpConnection = new AMQPBrokerConnectConfiguration("connectTowardsServer1", "tcp://localhost:" + AMQP_PORT).setReconnectAttempts(-1).setRetryInterval(100);
         amqpConnection.addElement(new AMQPMirrorBrokerConnectionElement().setDurable(true));
         server_2.getConfiguration().addAMQPConnection(amqpConnection);
      }

      server_2.start();

      server_2.addAddressInfo(new AddressInfo("sometest").setAutoCreated(false));
      server_2.createQueue(new QueueConfiguration("sometest").setDurable(true));

      Wait.assertTrue(() -> server_2.locateQueue("sometest") != null);
      Wait.assertTrue(() -> server.locateQueue("sometest") != null);

      server.addAddressInfo(new AddressInfo("OnServer1").setAutoCreated(false));
      server.createQueue(new QueueConfiguration("OnServer1").setDurable(true));

      Wait.assertTrue(() -> server.locateQueue("OnServer1") != null);
      Wait.assertTrue("Sync is not working on the way back", () -> server_2.locateQueue("OnServer1") != null, 2000);

      Wait.assertTrue(() -> server_2.locateQueue("sometest") != null);
      Wait.assertTrue(() -> server.locateQueue("sometest") != null);

      for (int i = 0; i < 10; i++) {
         final int queueID = i;
         server_2.createQueue(new QueueConfiguration("test2_" + i).setDurable(true));
         server.createQueue(new QueueConfiguration("test1_" + i).setDurable(true));
         Wait.assertTrue(() -> server.locateQueue("test2_" + queueID) != null);
         Wait.assertTrue(() -> server.locateQueue("test1_" + queueID) != null);
         Wait.assertTrue(() -> server_2.locateQueue("test2_" + queueID) != null);
         Wait.assertTrue(() -> server_2.locateQueue("test1_" + queueID) != null);
      }

      server_2.stop();
      server.stop();
   }


   @Test
   public void testSingleMessage() throws Exception {
      server.setIdentity("Server1");
      {
         AMQPBrokerConnectConfiguration amqpConnection = new AMQPBrokerConnectConfiguration("connectTowardsServer2", "tcp://localhost:" + AMQP_PORT_2).setReconnectAttempts(3).setRetryInterval(100);
         amqpConnection.addElement(new AMQPMirrorBrokerConnectionElement().setDurable(true));
         server.getConfiguration().addAMQPConnection(amqpConnection);
      }
      server.start();

      server_2 = createServer(AMQP_PORT_2, false);
      server_2.setIdentity("Server2");

      {
         AMQPBrokerConnectConfiguration amqpConnection = new AMQPBrokerConnectConfiguration("connectTowardsServer1", "tcp://localhost:" + AMQP_PORT).setReconnectAttempts(-1).setRetryInterval(100);
         amqpConnection.addElement(new AMQPMirrorBrokerConnectionElement().setDurable(true));
         server_2.getConfiguration().addAMQPConnection(amqpConnection);
      }

      server_2.start();

      server_2.addAddressInfo(new AddressInfo(getQueueName()).setAutoCreated(false).addRoutingType(RoutingType.ANYCAST));
      server_2.createQueue(new QueueConfiguration(getQueueName()).setDurable(true).setRoutingType(RoutingType.ANYCAST));

      Wait.assertTrue(() -> server_2.locateQueue(getQueueName()) != null);
      Wait.assertTrue(() -> server.locateQueue(getQueueName()) != null);

      ConnectionFactory cf1 = CFUtil.createConnectionFactory("AMQP", "tcp://localhost:" + AMQP_PORT);
      Connection connection1 = cf1.createConnection();
      Session session1 = connection1.createSession(true, Session.SESSION_TRANSACTED);
      connection1.start();

      ConnectionFactory cf2 = CFUtil.createConnectionFactory("AMQP", "tcp://localhost:" + AMQP_PORT_2);
      Connection connection2 = cf2.createConnection();
      Session session2 = connection2.createSession(true, Session.SESSION_TRANSACTED);
      connection2.start();

      Queue queue = session1.createQueue(getQueueName());

      MessageProducer producerServer1 = session1.createProducer(queue);
      MessageProducer producerServer2 = session2.createProducer(queue);

      TextMessage message = session1.createTextMessage("test");
      message.setIntProperty("i", 0);
      message.setStringProperty("server", server.getIdentity());
      producerServer1.send(message);
      session1.commit();

      org.apache.activemq.artemis.core.server.Queue queueOnServer1 = server.locateQueue(getQueueName());
      org.apache.activemq.artemis.core.server.Queue queueOnServer2 = server_2.locateQueue(getQueueName());
      Assert.assertNotNull(queueOnServer1);
      Assert.assertNotNull(queueOnServer2);

      Wait.assertEquals(1, queueOnServer1::getMessageCount);
      Wait.assertEquals(1, queueOnServer2::getMessageCount);

      message = session1.createTextMessage("test");
      message.setIntProperty("i", 1);
      message.setStringProperty("server", server_2.getIdentity());
      producerServer2.send(message);
      session2.commit();

      if (logger.isDebugEnabled() && !Wait.waitFor(() -> queueOnServer1.getMessageCount() == 2)) {
         debugData();
      }

      Wait.assertEquals(2, queueOnServer1::getMessageCount);
      Wait.assertEquals(2, queueOnServer2::getMessageCount);

      connection1.close();
      connection2.close();

      server_2.stop();
      server.stop();
   }


   @Test
   public void testSyncData() throws Exception {
      int NUMBER_OF_MESSAGES = 100;
      server.setIdentity("Server1");
      {
         AMQPBrokerConnectConfiguration amqpConnection = new AMQPBrokerConnectConfiguration("connectTowardsServer2", "tcp://localhost:" + AMQP_PORT_2).setReconnectAttempts(3).setRetryInterval(100);
         amqpConnection.addElement(new AMQPMirrorBrokerConnectionElement().setDurable(true));
         server.getConfiguration().addAMQPConnection(amqpConnection);
      }
      server.start();

      server_2 = createServer(AMQP_PORT_2, false);
      server_2.setIdentity("Server2");

      {
         AMQPBrokerConnectConfiguration amqpConnection = new AMQPBrokerConnectConfiguration("connectTowardsServer1", "tcp://localhost:" + AMQP_PORT).setReconnectAttempts(-1).setRetryInterval(100);
         amqpConnection.addElement(new AMQPMirrorBrokerConnectionElement().setDurable(true));
         server_2.getConfiguration().addAMQPConnection(amqpConnection);
      }

      server_2.start();

      server_2.addAddressInfo(new AddressInfo(getQueueName()).setAutoCreated(false).addRoutingType(RoutingType.ANYCAST));
      server_2.createQueue(new QueueConfiguration(getQueueName()).setDurable(true).setRoutingType(RoutingType.ANYCAST));

      Wait.assertTrue(() -> server_2.locateQueue(getQueueName()) != null);
      Wait.assertTrue(() -> server.locateQueue(getQueueName()) != null);

      ConnectionFactory cf1 = CFUtil.createConnectionFactory("AMQP", "tcp://localhost:" + AMQP_PORT);
      Connection connection1 = cf1.createConnection();
      Session session1 = connection1.createSession(true, Session.SESSION_TRANSACTED);
      connection1.start();

      ConnectionFactory cf2 = CFUtil.createConnectionFactory("AMQP", "tcp://localhost:" + AMQP_PORT_2);
      Connection connection2 = cf2.createConnection();
      Session session2 = connection2.createSession(true, Session.SESSION_TRANSACTED);
      connection2.start();

      Queue queue = session1.createQueue(getQueueName());

      MessageProducer producerServer1 = session1.createProducer(queue);
      MessageProducer producerServer2 = session2.createProducer(queue);

      for (int i = 0; i < NUMBER_OF_MESSAGES; i++) {
         TextMessage message = session1.createTextMessage("test " + i);
         message.setIntProperty("i", i);
         message.setStringProperty("server", server.getIdentity());
         producerServer1.send(message);
      }
      session1.commit();

      org.apache.activemq.artemis.core.server.Queue queueOnServer1 = server.locateQueue(getQueueName());
      org.apache.activemq.artemis.core.server.Queue queueOnServer2 = server_2.locateQueue(getQueueName());
      Assert.assertNotNull(queueOnServer1);
      Assert.assertNotNull(queueOnServer2);

      Wait.assertEquals(NUMBER_OF_MESSAGES, queueOnServer1::getMessageCount);
      Wait.assertEquals(NUMBER_OF_MESSAGES, queueOnServer2::getMessageCount);

      for (int i = NUMBER_OF_MESSAGES; i < NUMBER_OF_MESSAGES * 2; i++) {
         TextMessage message = session1.createTextMessage("test " + i);
         message.setIntProperty("i", i);
         message.setStringProperty("server", server_2.getIdentity());
         producerServer2.send(message);
      }
      session2.commit();

      if (logger.isDebugEnabled() && !Wait.waitFor(() -> queueOnServer1.getMessageCount() == NUMBER_OF_MESSAGES * 2)) {
         debugData();
      }

      Wait.assertEquals(NUMBER_OF_MESSAGES * 2, queueOnServer1::getMessageCount);
      Wait.assertEquals(NUMBER_OF_MESSAGES * 2, queueOnServer2::getMessageCount);

      MessageConsumer consumerOn1 = session1.createConsumer(queue);
      for (int i = 0; i < NUMBER_OF_MESSAGES * 2; i++) {
         TextMessage message = (TextMessage) consumerOn1.receive(5000);
         logger.debug("### Client acking message(" + i + ") on server 1, a message that was original sent on " + message.getStringProperty("server") + " text = " + message.getText());
         Assert.assertNotNull(message);
         Assert.assertEquals(i, message.getIntProperty("i"));
         Assert.assertEquals("test " + i, message.getText());
         session1.commit();
      }

      boolean bothConsumed = Wait.waitFor(() -> {
         long q1 = queueOnServer1.getMessageCount();
         long q2 = queueOnServer2.getMessageCount();
         logger.debug("Queue on Server 1 = " + q1);
         logger.debug("Queue on Server 2 = " + q2);
         return q1 == 0 && q2 == 0;
      }, 5_000, 1000);

      if (logger.isDebugEnabled() && !bothConsumed) {
         debugData();
         Assert.fail("q1 = " + queueOnServer1.getMessageCount() + ", q2 = " + queueOnServer2.getMessageCount());
      }

      Assert.assertEquals(0, queueOnServer1.getMessageCount());
      Assert.assertEquals(0, queueOnServer2.getConsumerCount());

      System.out.println("Queue on Server 1 = " + queueOnServer1.getMessageCount());
      System.out.println("Queue on Server 2 = " + queueOnServer2.getMessageCount());

      server_2.stop();
      server.stop();
   }

   private void debugData() throws Exception {
      StringPrintStream stringPrintStream = new StringPrintStream();
      PrintStream out = stringPrintStream.newStream();
      org.apache.activemq.artemis.core.server.Queue queueToDebugOn1 = server.locateQueue(getQueueName());
      org.apache.activemq.artemis.core.server.Queue queueToDebugOn2 = server_2.locateQueue(getQueueName());
      out.println("*******************************************************************************************************************************");
      out.println("Queue on Server 1 with count = " + queueToDebugOn1.getMessageCount());
      queueToDebugOn1.forEach((r) -> out.println("Server1 has reference " + r.getMessage()));
      out.println("*******************************************************************************************************************************");
      out.println("Queue on Server 2 with count = " + queueToDebugOn2.getMessageCount());
      queueToDebugOn2.forEach((r) -> out.println("Server2 has reference " + r.getMessage()));
      out.println("*******************************************************************************************************************************");
      out.println("PrintData Server 1");
      PrintData.printMessages(server.getConfiguration().getJournalLocation(), out, false, false, true, false);
      out.println("*******************************************************************************************************************************");
      out.println("PrintData Server 2");
      PrintData.printMessages(server_2.getConfiguration().getJournalLocation(), out, false, false, true, false);
      logger.debug("Data Available on Servers:\n" + stringPrintStream.toString());
   }

   @Test
   public void testSyncDataNoSuppliedID() throws Exception {
      int NUMBER_OF_MESSAGES = 100;
      server.setIdentity("Server1");

      {
         AMQPBrokerConnectConfiguration amqpConnection = new AMQPBrokerConnectConfiguration("connectTowardsServer2", "tcp://localhost:" + AMQP_PORT_2).setReconnectAttempts(3).setRetryInterval(100);
         amqpConnection.addElement(new AMQPMirrorBrokerConnectionElement().setDurable(true));
         server.getConfiguration().addAMQPConnection(amqpConnection);
      }
      server.start();

      server_2 = createServer(AMQP_PORT_2, false);
      server_2.setIdentity("Server2");

      {
         AMQPBrokerConnectConfiguration amqpConnection = new AMQPBrokerConnectConfiguration("connectTowardsServer1", "tcp://localhost:" + AMQP_PORT).setReconnectAttempts(-1).setRetryInterval(100);
         amqpConnection.addElement(new AMQPMirrorBrokerConnectionElement().setDurable(true));
         server_2.getConfiguration().addAMQPConnection(amqpConnection);
      }

      server_2.start();

      server_2.addAddressInfo(new AddressInfo(getQueueName()).setAutoCreated(false).addRoutingType(RoutingType.ANYCAST));
      server_2.createQueue(new QueueConfiguration(getQueueName()).setDurable(true).setRoutingType(RoutingType.ANYCAST));

      Wait.assertTrue(() -> server_2.locateQueue(getQueueName()) != null);
      Wait.assertTrue(() -> server.locateQueue(getQueueName()) != null);

      AmqpClient cf1 = new AmqpClient(new URI("tcp://localhost:" + AMQP_PORT), null, null);
      AmqpConnection connection1 = cf1.createConnection();
      connection1.connect();
      AmqpSession session1 = connection1.createSession();

      AmqpClient cf2 = new AmqpClient(new URI("tcp://localhost:" + AMQP_PORT_2), null, null);
      AmqpConnection connection2 = cf2.createConnection();
      connection2.connect();
      AmqpSession session2 = connection2.createSession();

      AmqpSender producerServer1 = session1.createSender(getQueueName());
      AmqpSender producerServer2 = session2.createSender(getQueueName());

      for (int i = 0; i < NUMBER_OF_MESSAGES; i++) {
         AmqpMessage message = new AmqpMessage();
         message.setDurable(true);
         message.setApplicationProperty("i", i);
         producerServer1.send(message);
      }

      org.apache.activemq.artemis.core.server.Queue queueOnServer1 = server.locateQueue(getQueueName());
      org.apache.activemq.artemis.core.server.Queue queueOnServer2 = server_2.locateQueue(getQueueName());
      Assert.assertNotNull(queueOnServer1);
      Assert.assertNotNull(queueOnServer2);

      Wait.assertEquals(NUMBER_OF_MESSAGES, queueOnServer1::getMessageCount);
      Wait.assertEquals(NUMBER_OF_MESSAGES, queueOnServer2::getMessageCount);

      for (int i = NUMBER_OF_MESSAGES; i < NUMBER_OF_MESSAGES * 2; i++) {
         AmqpMessage message = new AmqpMessage();
         message.setDurable(true);
         message.setApplicationProperty("i", i);
         producerServer2.send(message);
      }

      Wait.assertEquals(NUMBER_OF_MESSAGES * 2, queueOnServer1::getMessageCount);
      Wait.assertEquals(NUMBER_OF_MESSAGES * 2, queueOnServer2::getMessageCount);

      AmqpReceiver consumerOn1 = session1.createReceiver(getQueueName());
      consumerOn1.flow(NUMBER_OF_MESSAGES * 2 + 1);
      for (int i = 0; i < NUMBER_OF_MESSAGES * 2; i++) {
         AmqpMessage message = consumerOn1.receive(5, TimeUnit.SECONDS);
         Assert.assertNotNull(message);
         message.accept();
         Assert.assertEquals(i, (int) message.getApplicationProperty("i"));
      }

      Wait.assertEquals(0, queueOnServer1::getMessageCount);
      Wait.assertEquals(0, () -> {
         System.out.println(queueOnServer2.getMessageCount());
         return queueOnServer2.getMessageCount();
      });

      connection1.close();
      connection2.close();
      server_2.stop();
      server.stop();
   }

   @Test
   public void testLargeMessageInSync() throws Exception {
      String queueName = "testSyncLargeMessage";
      server.setIdentity("Server1");
      {
         AMQPBrokerConnectConfiguration amqpConnection = new AMQPBrokerConnectConfiguration("connectTowardsServer2", "tcp://localhost:" + AMQP_PORT_2).setReconnectAttempts(3).setRetryInterval(100);
         amqpConnection.addElement(new AMQPMirrorBrokerConnectionElement().setDurable(true));
         server.getConfiguration().addAMQPConnection(amqpConnection);
      }
      server.start();

      server_2 = createServer(AMQP_PORT_2, false);
      server_2.setIdentity("Server2");

      {
         AMQPBrokerConnectConfiguration amqpConnection = new AMQPBrokerConnectConfiguration("connectTowardsServer1", "tcp://localhost:" + AMQP_PORT).setReconnectAttempts(-1).setRetryInterval(100);
         amqpConnection.addElement(new AMQPMirrorBrokerConnectionElement().setDurable(true));
         server_2.getConfiguration().addAMQPConnection(amqpConnection);
      }

      server_2.start();

      server_2.addAddressInfo(new AddressInfo(queueName).setAutoCreated(false).addRoutingType(RoutingType.ANYCAST));
      server_2.createQueue(new QueueConfiguration(queueName).setDurable(true).setRoutingType(RoutingType.ANYCAST));

      Wait.assertTrue(() -> server_2.locateQueue(queueName) != null);
      Wait.assertTrue(() -> server.locateQueue(queueName) != null);

      String bigString;
      {
         StringBuffer bigStringBuffer = new StringBuffer();
         while (bigStringBuffer.length() < 200 * 1024) {
            bigStringBuffer.append("This is a big string ");
         }
         bigString = bigStringBuffer.toString();
      }

      ConnectionFactory factory1 = CFUtil.createConnectionFactory("AMQP", "tcp://localhost:" + AMQP_PORT);
      ConnectionFactory factory2 = CFUtil.createConnectionFactory("AMQP", "tcp://localhost:" + AMQP_PORT_2);

      try (Connection connection = factory1.createConnection()) {
         Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
         Queue queue = session.createQueue(queueName);
         MessageProducer producer = session.createProducer(queue);
         producer.send(session.createTextMessage(bigString));
      }

      try (Connection connection = factory2.createConnection()) {
         org.apache.activemq.artemis.core.server.Queue serverQueue = server.locateQueue(queueName);
         Wait.assertEquals(1, serverQueue::getMessageCount);
         Wait.assertEquals(1, () -> getNumberOfFiles(server_2.getConfiguration().getLargeMessagesLocation()), 5000, 100);
         Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
         Queue queue = session.createQueue(queueName);
         MessageConsumer consumer = session.createConsumer(queue);
         connection.start();
         TextMessage message = (TextMessage)consumer.receive(5000);
         Assert.assertNotNull(message);
         Assert.assertEquals(bigString, message.getText());
         Wait.assertEquals(0, () -> getNumberOfFiles(server_2.getConfiguration().getLargeMessagesLocation()), 5000, 100);
      }

      try (Connection connection = factory1.createConnection()) {
         org.apache.activemq.artemis.core.server.Queue serverQueue = server.locateQueue(queueName);

         Wait.assertEquals(0, serverQueue::getMessageCount);

         Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
         Queue queue = session.createQueue(queueName);
         MessageConsumer consumer = session.createConsumer(queue);
         connection.start();
         TextMessage message = (TextMessage)consumer.receiveNoWait();
         Assert.assertNull(message);
         server.stop();
         server_2.stop();
         Wait.assertEquals(0, () -> getNumberOfFiles(server.getConfiguration().getLargeMessagesLocation()), 1000, 100);
      }
   }

}