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

import static org.junit.jupiter.api.Assertions.assertNotNull;
import java.lang.invoke.MethodHandles;
import java.net.URI;
import java.util.concurrent.TimeUnit;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.Session;
import javax.jms.Topic;

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.core.config.amqpBrokerConnectivity.AMQPBrokerConnectConfiguration;
import org.apache.activemq.artemis.core.config.amqpBrokerConnectivity.AMQPBrokerConnectionAddressType;
import org.apache.activemq.artemis.core.config.amqpBrokerConnectivity.AMQPBrokerConnectionElement;
import org.apache.activemq.artemis.core.server.ActiveMQServer;
import org.apache.activemq.artemis.tests.integration.amqp.AmqpClientTestSupport;
import org.apache.activemq.artemis.tests.util.CFUtil;
import org.apache.qpid.proton.amqp.transport.LinkError;
import org.apache.qpid.protonj2.test.driver.ProtonTestServer;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.Timeout;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Test the Receiver functionality on AMQP broker connections
 */
@Timeout(20)
public class AMQPBrokerConnectionReceiverTest extends AmqpClientTestSupport {

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

   @Override
   protected String getConfiguredProtocols() {
      return "AMQP,CORE";
   }

   @Override
   protected ActiveMQServer createServer() throws Exception {
      // Creates the broker used to make the outgoing connection. The port passed is for
      // that brokers acceptor. The test server connected to by the broker binds to a random port.
      return createServer(AMQP_PORT, false);
   }

   @Test
   public void testBrokerConnectionCreatesReceiverOnRemote() throws Exception {
      try (ProtonTestServer peer = new ProtonTestServer()) {
         peer.expectSASLAnonymousConnect();
         peer.expectOpen().respond();
         peer.expectBegin().respond();
         peer.expectAttach().ofReceiver().respondInKind();
         peer.expectFlow();
         peer.start();

         final URI remoteURI = peer.getServerURI();
         LOG.info("Test started, peer listening on: {}", remoteURI);

         final AMQPBrokerConnectionElement element = new AMQPBrokerConnectionElement();
         element.setType(AMQPBrokerConnectionAddressType.RECEIVER);
         element.setName(getTestName());
         element.setMatchAddress("test");

         final AMQPBrokerConnectConfiguration amqpConnection =
            new AMQPBrokerConnectConfiguration(getTestName(), "tcp://" + remoteURI.getHost() + ":" + remoteURI.getPort());
         amqpConnection.setReconnectAttempts(0);
         amqpConnection.addElement(element);
         amqpConnection.setAutostart(true);

         server.getConfiguration().addAMQPConnection(amqpConnection);
         server.start();
         server.createQueue(new QueueConfiguration("test").setRoutingType(RoutingType.ANYCAST)
                                                          .setAddress("test")
                                                          .setAutoCreated(false));

         peer.waitForScriptToComplete();
         peer.expectClose();
         peer.remoteClose().now();
         peer.waitForScriptToComplete();
         peer.close();
      }
   }

   @Test
   public void testIncomingMessageWithNoToFieldArrivesOnConfiguredAddress() throws Exception {
      try (ProtonTestServer peer = new ProtonTestServer()) {
         peer.expectSASLAnonymousConnect();
         peer.expectOpen().respond();
         peer.expectBegin().respond();
         peer.start();

         final URI remoteURI = peer.getServerURI();
         LOG.info("Test started, peer listening on: {}", remoteURI);

         final AMQPBrokerConnectionElement element = new AMQPBrokerConnectionElement();
         element.setType(AMQPBrokerConnectionAddressType.RECEIVER);
         element.setName(getTestName());
         element.setMatchAddress("test");

         final AMQPBrokerConnectConfiguration amqpConnection =
            new AMQPBrokerConnectConfiguration(getTestName(), "tcp://" + remoteURI.getHost() + ":" + remoteURI.getPort());
         amqpConnection.setReconnectAttempts(0);
         amqpConnection.addElement(element);
         amqpConnection.setAutostart(true);

         server.getConfiguration().addAMQPConnection(amqpConnection);
         server.start();

         peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
         peer.expectAttach().ofReceiver().withSource().withAddress("test").and()
                                         .withTarget().withAddress("test").and()
                                         .respondInKind();
         peer.expectFlow();
         peer.remoteTransfer().withDeliveryId(1)
                              .withBody().withString("test-body").also()
                              .queue();
         peer.expectDisposition().withState().accepted();

         final ConnectionFactory factory = CFUtil.createConnectionFactory("AMQP", "tcp://localhost:" + AMQP_PORT);

         try (Connection connection = factory.createConnection()) {
            final Session session = connection.createSession(Session.AUTO_ACKNOWLEDGE);
            final Topic topic = session.createTopic("test");
            final MessageConsumer consumer = session.createConsumer(topic);

            connection.start();

            final Message received = consumer.receive(5_000);
            assertNotNull(received);

            consumer.close();
         }

         peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
         peer.expectClose();
         peer.remoteClose().now();
         peer.waitForScriptToComplete();
         peer.close();
      }
   }

   @Test
   public void testIncomingMessageWithToFieldArrivesOnConfiguredAddress() throws Exception {
      try (ProtonTestServer peer = new ProtonTestServer()) {
         peer.expectSASLAnonymousConnect();
         peer.expectOpen().respond();
         peer.expectBegin().respond();
         peer.start();

         final URI remoteURI = peer.getServerURI();
         LOG.info("Test started, peer listening on: {}", remoteURI);

         final AMQPBrokerConnectionElement element = new AMQPBrokerConnectionElement();
         element.setType(AMQPBrokerConnectionAddressType.RECEIVER);
         element.setName(getTestName());
         element.setMatchAddress("test");

         final AMQPBrokerConnectConfiguration amqpConnection =
            new AMQPBrokerConnectConfiguration(getTestName(), "tcp://" + remoteURI.getHost() + ":" + remoteURI.getPort());
         amqpConnection.setReconnectAttempts(0);
         amqpConnection.addElement(element);
         amqpConnection.setAutostart(true);

         server.getConfiguration().addAMQPConnection(amqpConnection);
         server.start();

         peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
         peer.expectAttach().ofReceiver().withSource().withAddress("test").and()
                                         .withTarget().withAddress("test").and()
                                         .respondInKind();
         peer.expectFlow();
         peer.remoteTransfer().withDeliveryId(1)
                              .withProperties().withTo("should-not-be-used").also()
                              .withBody().withString("test-body").also()
                              .queue();
         peer.expectDisposition().withState().accepted();

         final ConnectionFactory factory = CFUtil.createConnectionFactory("AMQP", "tcp://localhost:" + AMQP_PORT);

         try (Connection connection = factory.createConnection()) {
            final Session session = connection.createSession(Session.AUTO_ACKNOWLEDGE);
            final Topic topic = session.createTopic("test");
            final MessageConsumer consumer = session.createConsumer(topic);

            connection.start();

            final Message received = consumer.receive(5_000);
            assertNotNull(received);

            consumer.close();
         }

         peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
         peer.expectClose();
         peer.remoteClose().now();
         peer.waitForScriptToComplete();
         peer.close();
      }
   }

   @Test
   public void testBrokerConnectionRetriesReceiverOnRemoteIfAttachRejected() throws Exception {
      try (ProtonTestServer peer = new ProtonTestServer()) {
         peer.expectSASLAnonymousConnect();
         peer.expectOpen().respond();
         peer.expectBegin().respond();
         peer.expectAttach().ofReceiver().reject(true, LinkError.DETACH_FORCED.toString(), "Attach refused");
         peer.expectDetach().optional();
         peer.expectClose().optional();
         peer.expectConnectionToDrop();
         peer.expectSASLAnonymousConnect();
         peer.expectOpen().respond();
         peer.expectBegin().respond();
         peer.expectAttach().ofReceiver().respondInKind();
         peer.expectFlow();
         peer.start();

         final URI remoteURI = peer.getServerURI();
         LOG.info("Test started, peer listening on: {}", remoteURI);

         final AMQPBrokerConnectionElement element = new AMQPBrokerConnectionElement();
         element.setType(AMQPBrokerConnectionAddressType.RECEIVER);
         element.setName(getTestName());
         element.setMatchAddress("test");

         final AMQPBrokerConnectConfiguration amqpConnection =
            new AMQPBrokerConnectConfiguration(getTestName(), "tcp://" + remoteURI.getHost() + ":" + remoteURI.getPort());
         amqpConnection.setReconnectAttempts(10);
         amqpConnection.setRetryInterval(100);
         amqpConnection.addElement(element);
         amqpConnection.setAutostart(true);

         server.getConfiguration().addAMQPConnection(amqpConnection);
         server.start();
         server.createQueue(new QueueConfiguration("test").setRoutingType(RoutingType.ANYCAST)
                                                          .setAddress("test")
                                                          .setAutoCreated(false));

         peer.waitForScriptToComplete();
         peer.expectClose();
         peer.remoteClose().now();
         peer.waitForScriptToComplete();
         peer.close();
      }
   }

   @Test
   public void testBrokerConnectionRetriesReceiverOnRemoteIfTargetQueueRemovedAndLaterAddedBack() throws Exception {
      try (ProtonTestServer peer = new ProtonTestServer()) {
         peer.expectSASLAnonymousConnect();
         peer.expectOpen().respond();
         peer.expectBegin().respond();
         peer.execute(() -> {
            try {
               server.createQueue(new QueueConfiguration("test").setRoutingType(RoutingType.ANYCAST)
                                                                .setAddress("test")
                                                                .setAutoCreated(false));
            } catch (Exception e) {
               LOG.warn("Error on creating server address and queue: ", e);
            }
         }).queue();
         peer.expectAttach().ofReceiver();
         peer.execute(() -> {
            try {
               server.removeAddressInfo(SimpleString.of("test"), null, true);
            } catch (Exception e) {
               LOG.warn("Error on removing server address and queue: ", e);
            }
            peer.respondToLastAttach().now();
         }).queue();
         peer.expectDetach().respond();
         peer.execute(() -> {
            try {
               server.createQueue(new QueueConfiguration("test").setRoutingType(RoutingType.ANYCAST)
                                                                .setAddress("test")
                                                                .setAutoCreated(false));
            } catch (Exception e) {
               LOG.warn("Error on creating server address and queue: ", e);
            }
         }).queue();
         peer.expectAttach().ofReceiver().respondInKind();
         peer.expectFlow();
         peer.start();

         final URI remoteURI = peer.getServerURI();
         LOG.info("Test started, peer listening on: {}", remoteURI);

         final AMQPBrokerConnectionElement element = new AMQPBrokerConnectionElement();
         element.setType(AMQPBrokerConnectionAddressType.RECEIVER);
         element.setName(getTestName());
         element.setMatchAddress("test");

         final AMQPBrokerConnectConfiguration amqpConnection =
            new AMQPBrokerConnectConfiguration(getTestName(), "tcp://" + remoteURI.getHost() + ":" + remoteURI.getPort());
         amqpConnection.setReconnectAttempts(10);
         amqpConnection.setRetryInterval(50);
         amqpConnection.addElement(element);
         amqpConnection.setAutostart(true);

         server.getConfiguration().addAMQPConnection(amqpConnection);
         server.start();

         peer.waitForScriptToComplete();
         peer.expectClose();
         peer.remoteClose().now();
         peer.waitForScriptToComplete();
         peer.close();
      }
   }
}
