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

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 java.util.Collections;

import org.apache.activemq.artemis.api.core.QueueConfiguration;
import org.apache.activemq.artemis.api.core.RoutingType;
import org.apache.activemq.artemis.core.config.FederationConfiguration;
import org.apache.activemq.artemis.core.config.federation.FederationQueuePolicyConfiguration;
import org.apache.activemq.artemis.core.config.federation.FederationUpstreamConfiguration;
import org.apache.activemq.artemis.core.server.ActiveMQServer;
import org.apache.activemq.artemis.core.settings.impl.AddressSettings;
import org.apache.activemq.artemis.jms.client.ActiveMQConnectionFactory;
import org.apache.activemq.artemis.tests.util.Wait;
import org.junit.Before;
import org.junit.Test;

public class FederatedQueuePullConsumerTest extends FederatedTestBase {

   @Override
   @Before
   public void setUp() throws Exception {
      super.setUp();
      getServer(0).getConfiguration().addConnectorConfiguration("server-pull-1", "tcp://localhost:" + 61617 + "?consumerWindowSize=0;ackBatchSize=10");
   }

   @Override
   protected boolean isNetty() {
      // such that url params can be used for the server-pull-1 connector, vm urls don't propagate url params!
      return true;
   }

   @Override
   protected void configureQueues(ActiveMQServer server) throws Exception {
      server.getAddressSettingsRepository().addMatch("#", new AddressSettings().setAutoCreateAddresses(false).setAutoCreateQueues(false)
         .setDefaultConsumerWindowSize(20 * 300));
      createSimpleQueue(server, getName());
   }

   protected ConnectionFactory getCF(int i) throws Exception {
      ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("vm://" + i);
      // pull consumers to allow deterministic message consumption
      factory.setConsumerWindowSize(0);
      return factory;
   }

   @Test
   public void testFederatedQueuePullFromUpstream() throws Exception {
      String queueName = getName();

      FederationConfiguration federationConfiguration = FederatedTestUtil.createQueueUpstreamFederationConfiguration("server-pull-1", queueName);
      getServer(0).getConfiguration().getFederationConfigurations().add(federationConfiguration);
      getServer(0).getFederationManager().deploy();

      testFederatedQueuePullFromUpstream(queueName);
   }

   @Test
   public void testMultipleFederatedQueueRemoteConsumersUpstream() throws Exception {
      String connector = "server-pull-1";

      getServer(0).getAddressSettingsRepository().getMatch("#").setAutoCreateAddresses(true).setAutoCreateQueues(true);
      getServer(1).getAddressSettingsRepository().getMatch("#").setAutoCreateAddresses(true).setAutoCreateQueues(true);

      getServer(1).createQueue(new QueueConfiguration("Test.Q.1").setRoutingType(RoutingType.ANYCAST));
      getServer(1).createQueue(new QueueConfiguration("Test.Q.2").setRoutingType(RoutingType.ANYCAST));

      getServer(0).getConfiguration().getFederationConfigurations().add(new FederationConfiguration()
                                                                           .setName("default")
                                                                           .addFederationPolicy(new FederationQueuePolicyConfiguration()
                                                                                                   .setName("myQueuePolicy")
                                                                                                   .addInclude(new FederationQueuePolicyConfiguration.Matcher()
                                                                                                                  .setQueueMatch("#")
                                                                                                                  .setAddressMatch("Test.#")))
                                                                           .addUpstreamConfiguration(new FederationUpstreamConfiguration()
                                                                                                        .setName("server1-upstream")
                                                                                                        .addPolicyRef("myQueuePolicy")
                                                                                                        .setStaticConnectors(Collections.singletonList(connector))));
      getServer(0).getFederationManager().deploy();

      ConnectionFactory cf1 = getCF(0);
      ConnectionFactory cf2 = getCF(0);
      ConnectionFactory cf3 = getCF(1);
      try (Connection consumer1Connection = cf1.createConnection();
           Connection consumer2Connection = cf2.createConnection();
           Connection producerConnection = cf3.createConnection()) {
         consumer1Connection.start();
         Session session1 = consumer1Connection.createSession();
         Queue queue1 = session1.createQueue("Test.Q.1");
         MessageConsumer consumer1 = session1.createConsumer(queue1);

         consumer2Connection.start();
         Session session2 = consumer2Connection.createSession();
         Queue queue2 = session2.createQueue("Test.Q.2");
         MessageConsumer consumer2 = session2.createConsumer(queue2);

         Session session3 = producerConnection.createSession();
         MessageProducer producer = session3.createProducer(queue2);
         producer.send(session3.createTextMessage("hello"));

         assertNotNull(consumer2.receive(1000));

         consumer1Connection.close();

         producer.send(session3.createTextMessage("hello"));

         assertNotNull(consumer2.receive(1000));
      }
   }

   private void testFederatedQueuePullFromUpstream(final String queueName) throws Exception {

      ConnectionFactory cf1 = getCF(1);
      ConnectionFactory cf0 = getCF(0);
      try (Connection connection1 = cf1.createConnection();
           Connection connection0 = cf0.createConnection()) {
         connection1.start();
         Session session1 = connection1.createSession();
         Queue queue1 = session1.createQueue(queueName);
         MessageProducer producer = session1.createProducer(queue1);
         producer.send(session1.createTextMessage("1"));

         connection0.start();
         Session session0 = connection0.createSession();
         Queue queue0 = session0.createQueue(queueName);
         MessageProducer producer0 = session0.createProducer(queue1);
         producer0.send(session0.createTextMessage("0"));

         // no consumer upstream, messages locally, no pull
         MessageConsumer consumer0 = session0.createConsumer(queue0);

         // verify federated
         waitForBindings(getServer(1), queueName, true, 1, 1, 2000);

         // verify no federation of messages
         Wait.assertEquals(1, () -> getMessageCount(getServer(0), queueName), 2000, 100);
         Wait.assertEquals(1, () -> getMessageCount(getServer(1), queueName), 2000, 100);

         // drain local queue
         assertNotNull(consumer0.receive(1000));

         // no messages locally, expect message federation now of a batch
         // 4s b/c local queue size check is in seconds
         assertNotNull(consumer0.receive(4000));

         // verify all messages consumed
         Wait.assertEquals(0L, () -> getMessageCount(getServer(0), queueName), 2000, 100);
         Wait.assertEquals(0L, () -> getMessageCount(getServer(1), queueName), 2000, 100);

         assertNull(consumer0.receiveNoWait());

         // verify batch end
         final int mumMessages = 150;
         for (int i = 0; i < mumMessages; i++ ) {
            producer.send(session1.createTextMessage("1-" + i));
         }

         // upstream has most
         Wait.assertTrue(() -> getMessageCount(getServer(1), queueName) > 100, 2000, 200);
         Wait.assertTrue(() -> getMessageCount(getServer(1), queueName) < mumMessages, 2000, 200);

         // only a batch has been federated
         Wait.assertTrue(() -> getMessageCount(getServer(0), queueName) > 10, 2000, 100);
         Wait.assertTrue(() -> getMessageCount(getServer(0), queueName) < 100, 2000, 100);

         // verify all available
         for (int i = 0; i < mumMessages; i++ ) {
            assertNotNull(consumer0.receive(4000));
         }

         assertNull(consumer0.receiveNoWait());
         consumer0.close();
      }
   }
}
