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

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

import java.lang.invoke.MethodHandles;
import java.util.ArrayList;
import java.util.HashMap;

import org.apache.activemq.artemis.api.core.ActiveMQException;
import org.apache.activemq.artemis.core.config.Configuration;
import org.apache.activemq.artemis.core.config.impl.ConfigurationImpl;
import org.apache.activemq.artemis.core.paging.PagingManager;
import org.apache.activemq.artemis.core.persistence.OperationContext;
import org.apache.activemq.artemis.core.persistence.StorageManager;
import org.apache.activemq.artemis.core.persistence.impl.nullpm.NullStorageManager;
import org.apache.activemq.artemis.core.postoffice.PostOffice;
import org.apache.activemq.artemis.core.protocol.ServerPacketDecoder;
import org.apache.activemq.artemis.core.protocol.core.impl.RemotingConnectionImpl;
import org.apache.activemq.artemis.core.remoting.impl.invm.InVMConnection;
import org.apache.activemq.artemis.core.security.SecurityStore;
import org.apache.activemq.artemis.core.server.ActiveMQServer;
import org.apache.activemq.artemis.core.server.impl.ActiveMQServerImpl;
import org.apache.activemq.artemis.core.server.impl.ServerConsumerImpl;
import org.apache.activemq.artemis.core.server.impl.ServerSessionImpl;
import org.apache.activemq.artemis.core.server.management.ManagementService;
import org.apache.activemq.artemis.core.transaction.ResourceManager;
import org.apache.activemq.artemis.protocol.amqp.broker.AMQPSessionCallback;
import org.apache.activemq.artemis.protocol.amqp.broker.ProtonProtocolManager;
import org.apache.activemq.artemis.protocol.amqp.broker.ProtonProtocolManagerFactory;
import org.apache.activemq.artemis.protocol.amqp.proton.AMQPConnectionContext;
import org.apache.activemq.artemis.protocol.amqp.proton.AMQPSessionContext;
import org.apache.activemq.artemis.protocol.amqp.proton.ProtonServerSenderContext;
import org.apache.activemq.artemis.protocol.amqp.proton.handler.ProtonHandler;
import org.apache.activemq.artemis.spi.core.protocol.SessionCallback;
import org.apache.activemq.artemis.spi.core.remoting.BaseConnectionLifeCycleListener;
import org.apache.activemq.artemis.spi.core.remoting.BufferHandler;
import org.apache.activemq.artemis.tests.util.ActiveMQTestBase;
import org.apache.activemq.artemis.utils.ExecutorFactory;
import org.apache.activemq.artemis.utils.RandomUtil;
import org.apache.activemq.artemis.utils.actors.ArtemisExecutor;
import org.apache.qpid.proton.engine.Connection;
import org.apache.qpid.proton.engine.EndpointState;
import org.apache.qpid.proton.engine.Sender;
import org.junit.jupiter.api.Test;
import org.mockito.Mockito;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * as worked through ARTEMIS-4476 we fixed the possibility of a ghost consumer (a term coined by a user),
 * where the connection is gone but the consumer still in memory.
 * <p>
 * The fix involved on calling the disconnect from the proper threads.
 * <p>
 * And as a line of defense the ServerConsumer and AMQP handler are also validating the connection states.
 * If a connection is in destroyed state tries to create a consumer, the system should throw an error.
 */
public class OrphanedConsumerDefenseTest extends ActiveMQTestBase {

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

   @Test
   public void testDefendNPESession() throws Exception {
      ActiveMQServerImpl server = Mockito.mock(ActiveMQServerImpl.class);
      SessionCallback sessionCallback = Mockito.mock(SessionCallback.class);
      ManagementService managementService = Mockito.mock(ManagementService.class);

      try {
         new ServerConsumerImpl(1, null, null, null, 1, true, false, new NullStorageManager(), sessionCallback, true, true, managementService, false, 0, server);
         fail("Exception was expected");
      } catch (NullPointerException e) {
         logger.debug("Expected exception", e);
      }
   }

   @Test
   public void testDefendDestroyedConnection() throws Exception {
      ActiveMQServerImpl server = Mockito.mock(ActiveMQServerImpl.class);

      SessionCallback sessionCallback = Mockito.mock(SessionCallback.class);

      ManagementService managementService = Mockito.mock(ManagementService.class);

      StorageManager storageManager = new NullStorageManager();

      ArtemisExecutor artemisExecutor = Mockito.mock(ArtemisExecutor.class);
      ExecutorFactory executorFactory = Mockito.mock(ExecutorFactory.class);
      Mockito.when(executorFactory.getExecutor()).thenReturn(artemisExecutor);

      Configuration configuration = new ConfigurationImpl();

      ActiveMQServer mockServer = Mockito.mock(ActiveMQServer.class);
      Mockito.when(mockServer.getExecutorFactory()).thenReturn(executorFactory);
      Mockito.when(mockServer.getConfiguration()).thenReturn(configuration);

      BufferHandler bufferHandler = Mockito.mock(BufferHandler.class);
      InVMConnection inVMConnection = new InVMConnection(1, bufferHandler, Mockito.mock(BaseConnectionLifeCycleListener.class), artemisExecutor);

      RemotingConnectionImpl remotingConnection = new RemotingConnectionImpl(new ServerPacketDecoder(storageManager), inVMConnection, new ArrayList<>(), new ArrayList<>(), RandomUtil.randomSimpleString(), artemisExecutor);
      remotingConnection.destroy();
      ServerSessionImpl session = new ServerSessionImpl(RandomUtil.randomString(), RandomUtil.randomString(), RandomUtil.randomString(),
                                                        RandomUtil.randomString(), 1000, true, true, true, true, true,
                                                        remotingConnection, storageManager, Mockito.mock(PostOffice.class), Mockito.mock(ResourceManager.class), Mockito.mock(SecurityStore.class), Mockito.mock(ManagementService.class), mockServer, RandomUtil.randomSimpleString(), RandomUtil.randomSimpleString(), Mockito.mock(SessionCallback.class), Mockito.mock(OperationContext.class), Mockito.mock(PagingManager.class), new HashMap<>(), "securityDomain", false);

      try {
         new ServerConsumerImpl(1, session, null, null, 1, true, false, new NullStorageManager(), sessionCallback, true, true, managementService, false, 0, server);
         fail("Exception was expected");
      } catch (ActiveMQException activeMQException) {
         logger.info("Expected exception", activeMQException);
         assertTrue(activeMQException.getMessage().contains("AMQ229250"));
      }
   }

   @Test
   public void testDefendAMQPConsumerNullConnection() throws Exception {
      testDefendAMQPConsumer(true);
   }

   @Test
   public void testDefendAMQPConsumer() throws Exception {
      testDefendAMQPConsumer(false);
   }

   private void testDefendAMQPConsumer(boolean returnNullConnection) throws Exception {
      ActiveMQServer server = Mockito.mock(ActiveMQServer.class);
      ProtonProtocolManagerFactory protonProtocolManagerFactory = new ProtonProtocolManagerFactory();
      ProtonProtocolManager protonProtocolManager = new ProtonProtocolManager(protonProtocolManagerFactory, server, new ArrayList<>(), new ArrayList<>());
      AMQPConnectionContext connectionContext = Mockito.mock(AMQPConnectionContext.class);

      if (returnNullConnection) {
         Mockito.when(connectionContext.getHandler()).thenReturn(null);
      } else {
         ProtonHandler protonHandler = Mockito.mock(ProtonHandler.class);
         Mockito.when(connectionContext.getHandler()).thenReturn(protonHandler);
         Connection qpidConnection = Mockito.mock(Connection.class);
         Mockito.when(qpidConnection.getRemoteState()).thenReturn(EndpointState.CLOSED);
         Mockito.when(protonHandler.getConnection()).thenReturn(qpidConnection);
      }

      Mockito.when(connectionContext.getProtocolManager()).thenReturn(protonProtocolManager);
      Sender sender = Mockito.mock(Sender.class);
      AMQPSessionContext sessionContext = Mockito.mock(AMQPSessionContext.class);
      AMQPSessionCallback sessionCallback = Mockito.mock(AMQPSessionCallback.class);

      try {
         ProtonServerSenderContext serverSenderContext = new ProtonServerSenderContext(connectionContext, sender, sessionContext, sessionCallback);
         serverSenderContext.initialize();
      } catch (ActiveMQException e) {
         assertTrue(e.getMessage().contains("AMQ119027"));
         logger.warn(e.getMessage(), e);
      }
   }
}