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

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.transaction.TransactionManager;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.atomic.AtomicReference;

import org.apache.activemq.artemis.api.core.QueueConfiguration;
import org.apache.activemq.artemis.api.core.RoutingType;
import org.apache.activemq.artemis.api.core.TransportConfiguration;
import org.apache.activemq.artemis.api.jms.ActiveMQJMSClient;
import org.apache.activemq.artemis.api.jms.JMSFactoryType;
import org.apache.activemq.artemis.core.remoting.impl.invm.InVMConnectorFactory;
import org.apache.activemq.artemis.core.server.ActiveMQServer;
import org.apache.activemq.artemis.jms.bridge.ConnectionFactoryFactory;
import org.apache.activemq.artemis.jms.bridge.DestinationFactory;
import org.apache.activemq.artemis.jms.bridge.QualityOfServiceMode;
import org.apache.activemq.artemis.jms.bridge.impl.JMSBridgeImpl;
import org.apache.activemq.artemis.jms.client.ActiveMQJMSConnectionFactory;
import org.apache.activemq.artemis.tests.util.ActiveMQTestBase;
import org.apache.activemq.artemis.tests.util.Wait;
import org.apache.activemq.artemis.utils.RandomUtil;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.mockito.Mockito;

public class JMSBridgeImplTest extends ActiveMQTestBase {

   private static final String SOURCE = RandomUtil.randomString();

   private static final String TARGET = RandomUtil.randomString();

   private ActiveMQServer server;

   @Override
   @BeforeEach
   public void setUp() throws Exception {
      super.setUp();
      server = createServer(false, createDefaultInVMConfig());
      server.start();

      server.createQueue(QueueConfiguration.of(SOURCE).setRoutingType(RoutingType.ANYCAST));
      server.createQueue(QueueConfiguration.of(TARGET).setRoutingType(RoutingType.ANYCAST));
   }

   private static ConnectionFactory createConnectionFactory() {
      ActiveMQJMSConnectionFactory cf = (ActiveMQJMSConnectionFactory) ActiveMQJMSClient.createConnectionFactoryWithoutHA(JMSFactoryType.CF, new TransportConfiguration(InVMConnectorFactory.class.getName()));
      cf.setReconnectAttempts(0);
      cf.setBlockOnNonDurableSend(true);
      cf.setBlockOnDurableSend(true);
      return cf;
   }

   @Test
   public void testExceptionOnSourceAndManualRestartSucceeds() throws Exception {
      final AtomicReference<Connection> sourceConn = new AtomicReference<>();
      ActiveMQJMSConnectionFactory failingSourceCF = new ActiveMQJMSConnectionFactory(false, new TransportConfiguration(InVMConnectorFactory.class.getName())) {
         private static final long serialVersionUID = -8866390811966688830L;

         @Override
         public Connection createConnection() throws JMSException {
            sourceConn.set(super.createConnection());
            return sourceConn.get();
         }
      };
      // Note! We disable automatic reconnection on the session factory. The bridge needs to do the reconnection
      failingSourceCF.setReconnectAttempts(0);
      failingSourceCF.setBlockOnNonDurableSend(true);
      failingSourceCF.setBlockOnDurableSend(true);

      ConnectionFactoryFactory sourceCFF = () -> failingSourceCF;
      ConnectionFactoryFactory targetCFF = () -> createConnectionFactory();
      DestinationFactory sourceDF = () -> ActiveMQJMSClient.createQueue(JMSBridgeImplTest.SOURCE);
      DestinationFactory targetDF = () -> ActiveMQJMSClient.createQueue(JMSBridgeImplTest.TARGET);
      TransactionManager tm = Mockito.mock(TransactionManager.class);

      JMSBridgeImpl bridge = new JMSBridgeImpl();
      bridge.setSourceConnectionFactoryFactory(sourceCFF);
      bridge.setSourceDestinationFactory(sourceDF);
      bridge.setTargetConnectionFactoryFactory(targetCFF);
      bridge.setTargetDestinationFactory(targetDF);
      bridge.setFailureRetryInterval(1);
      bridge.setMaxRetries(0);
      bridge.setMaxBatchSize(1);
      bridge.setMaxBatchTime(-1);
      bridge.setTransactionManager(tm);
      bridge.setQualityOfServiceMode(QualityOfServiceMode.AT_MOST_ONCE);

      Assertions.assertFalse(bridge.isStarted());
      bridge.start();
      Assertions.assertTrue(bridge.isStarted());

      // make sure the bridge is actually working first
      Connection targetConn = createConnectionFactory().createConnection();
      Session targetSess = targetConn.createSession(false, Session.AUTO_ACKNOWLEDGE);
      MessageConsumer consumer = targetSess.createConsumer(targetDF.createDestination());
      final List<Message> messages = new LinkedList<>();
      MessageListener listener = message -> messages.add(message);
      consumer.setMessageListener(listener);
      targetConn.start();

      Session sourceSess = sourceConn.get().createSession(false, Session.AUTO_ACKNOWLEDGE);
      MessageProducer producer = sourceSess.createProducer(sourceDF.createDestination());
      producer.send(sourceSess.createTextMessage());

      Wait.assertEquals(1, () -> messages.size(), 2000, 100);

      sourceConn.get().getExceptionListener().onException(new JMSException("exception on the source"));
      Wait.assertTrue(() -> bridge.isFailed(), 2000, 50);
      targetConn.close();
      bridge.stop();
      Assertions.assertFalse(bridge.isStarted());
      bridge.start();
      Assertions.assertTrue(bridge.isStarted());

      // test the bridge again after it's been restarted to ensure it's working
      targetConn = JMSBridgeImplTest.createConnectionFactory().createConnection();
      targetSess = targetConn.createSession(false, Session.AUTO_ACKNOWLEDGE);
      consumer = targetSess.createConsumer(targetDF.createDestination());
      messages.clear();
      consumer.setMessageListener(listener);
      targetConn.start();

      sourceSess = sourceConn.get().createSession(false, Session.AUTO_ACKNOWLEDGE);
      producer = sourceSess.createProducer(sourceDF.createDestination());
      producer.send(sourceSess.createTextMessage());

      Wait.assertEquals(1, () -> messages.size(), 2000, 100);
      targetConn.close();
      sourceConn.get().close();
      bridge.stop();
      Assertions.assertFalse(bridge.isStarted());
   }
}
