content: /**
   * 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.ra;
  
  import java.lang.reflect.Method;
  
  import java.util.concurrent.atomic.AtomicBoolean;
  import jakarta.jms.Connection;
  import jakarta.jms.ConnectionConsumer;
  import jakarta.jms.ExceptionListener;
  import jakarta.jms.JMSException;
  import jakarta.jms.Message;
  import jakarta.jms.MessageListener;
  import jakarta.jms.Session;
  import jakarta.jms.Topic;
  import javax.naming.InitialContext;
  import javax.naming.NamingException;
  import jakarta.resource.ResourceException;
  import jakarta.resource.spi.endpoint.MessageEndpointFactory;
  import jakarta.resource.spi.work.Work;
  import jakarta.resource.spi.work.WorkException;
  import jakarta.resource.spi.work.WorkManager;
  
  import org.apache.activemq.ActiveMQConnection;
  import org.apache.activemq.ActiveMQConnectionConsumer;
  import org.apache.activemq.command.ActiveMQDestination;
  import org.apache.activemq.command.ActiveMQQueue;
  import org.apache.activemq.command.ActiveMQTopic;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  /**
   *  $Date$
   */
  public class ActiveMQEndpointWorker {
  
      public static final Method ON_MESSAGE_METHOD;
      private static final Logger LOG = LoggerFactory.getLogger(ActiveMQEndpointWorker.class);
  
      private static final long INITIAL_RECONNECT_DELAY = 1000; // 1 second.
      private static final long MAX_RECONNECT_DELAY = 1000 * 30; // 30 seconds.
      private static final ThreadLocal<Session> THREAD_LOCAL = new ThreadLocal<Session>();
  
      static {
          try {
              ON_MESSAGE_METHOD = MessageListener.class.getMethod("onMessage", new Class[] {
                  Message.class
              });
          } catch (Exception e) {
              throw new ExceptionInInitializerError(e);
          }
      }
  
      protected final ActiveMQEndpointActivationKey endpointActivationKey;
      protected final MessageEndpointFactory endpointFactory;
      protected final WorkManager workManager;
      protected final boolean transacted;
  
      private final ActiveMQDestination dest;
      private final Work connectWork;
      private final AtomicBoolean connecting = new AtomicBoolean(false);    
      private final Object shutdownMutex = new String("shutdownMutex");
      
      private ActiveMQConnection connection;
      private ActiveMQConnectionConsumer consumer;
      private ServerSessionPoolImpl serverSessionPool;
      private boolean running;
  
      protected ActiveMQEndpointWorker(final MessageResourceAdapter adapter, ActiveMQEndpointActivationKey key) throws ResourceException {
          this.endpointActivationKey = key;
          this.endpointFactory = endpointActivationKey.getMessageEndpointFactory();
          this.workManager = adapter.getBootstrapContext().getWorkManager();
          try {
              this.transacted = endpointFactory.isDeliveryTransacted(ON_MESSAGE_METHOD);
          } catch (NoSuchMethodException e) {
              throw new ResourceException("Endpoint does not implement the onMessage method.");
          }
  
          connectWork = new Work() {
              long currentReconnectDelay = INITIAL_RECONNECT_DELAY;
  
              public void release() {
                  //
              }
  
              public void run() {
                  currentReconnectDelay = INITIAL_RECONNECT_DELAY;
                  MessageActivationSpec activationSpec = endpointActivationKey.getActivationSpec();
                  if (LOG.isInfoEnabled()) {
                      LOG.info("Establishing connection to broker [" + adapter.getInfo().getServerUrl() + "]");
                  }
  
                  while (connecting.get() && running) {
                      try {
                          connection = adapter.makeConnection(activationSpec);
                          connection.setExceptionListener(new ExceptionListener() {
                              public void onException(JMSException error) {
                                  if (!serverSessionPool.isClosing()) {
                                      // initiate reconnection only once, i.e. on initial exception
                                      // and only if not already trying to connect
                                      LOG.error("Connection to broker failed: " + error.getMessage(), error);
                                      if (connecting.compareAndSet(false, true)) {
                                          synchronized (connectWork) {
                                              disconnect();
                                              serverSessionPool.closeSessions();
                                              connect();
                                          }
                                      } else {
                                          // connection attempt has already been initiated
                                          LOG.info("Connection attempt already in progress, ignoring connection exception");
                                      }
                                  }
                              }
                          });
                          connection.start();
  
                          if (activationSpec.isDurableSubscription()) {
                              consumer = (ActiveMQConnectionConsumer) connection.createDurableConnectionConsumer(
                                      (Topic) dest,
                                      activationSpec.getSubscriptionName(),
                                      emptyToNull(activationSpec.getMessageSelector()),
                                      serverSessionPool,
                                      connection.getPrefetchPolicy().getDurableTopicPrefetch(),
                                      activationSpec.getNoLocalBooleanValue());
                          } else {
                              consumer = (ActiveMQConnectionConsumer) connection.createConnectionConsumer(
                                      dest,
                                      emptyToNull(activationSpec.getMessageSelector()),
                                      serverSessionPool,
                                      getPrefetch(activationSpec, connection, dest),
                                      activationSpec.getNoLocalBooleanValue());
                          }
  
  
                          if (connecting.compareAndSet(true, false)) {
                              if (LOG.isInfoEnabled()) {
                                  LOG.info("Successfully established connection to broker [" + adapter.getInfo().getServerUrl() + "]");
                              }
                          } else {
                              LOG.error("Could not release connection lock");
                          }
  
                          if (consumer.getConsumerInfo().getCurrentPrefetchSize() == 0) {
                              LOG.error("Endpoint " + endpointActivationKey.getActivationSpec() + " will not receive any messages due to broker 'zero prefetch' configuration for: " + dest);
                          }
  
                      } catch (JMSException error) {
                          if (LOG.isDebugEnabled()) {
                              LOG.debug("Failed to connect: " + error.getMessage(), error);
                          }
                          disconnect();
                          pause(error);
                      }
                  }
              }
  
              private int getPrefetch(MessageActivationSpec activationSpec, ActiveMQConnection connection, ActiveMQDestination destination) {
                  if (destination.isTopic()) {
                      return connection.getPrefetchPolicy().getTopicPrefetch();
                  } else if (destination.isQueue()) {
                      return connection.getPrefetchPolicy().getQueuePrefetch();
                  } else {
                      return activationSpec.getMaxMessagesPerSessionsIntValue() * activationSpec.getMaxSessionsIntValue();
                  }
              }
              
              private void pause(JMSException error) {
                  if (currentReconnectDelay == MAX_RECONNECT_DELAY) {
                      LOG.error("Failed to connect to broker [" + adapter.getInfo().getServerUrl() + "]: " 
                              + error.getMessage(), error);
                      LOG.error("Endpoint will try to reconnect to the JMS broker in " + (MAX_RECONNECT_DELAY / 1000) + " seconds");
                  }
                  try {
                      synchronized ( shutdownMutex ) {
                          // shutdownMutex will be notified by stop() method in
                          // order to accelerate shutdown of endpoint
                          shutdownMutex.wait(currentReconnectDelay);
                      }
                  } catch ( InterruptedException e ) {
                      Thread.interrupted();
                  }
                  currentReconnectDelay *= 2;
                  if (currentReconnectDelay > MAX_RECONNECT_DELAY) {
                      currentReconnectDelay = MAX_RECONNECT_DELAY;
                  }                
              }
          };
  
          MessageActivationSpec activationSpec = endpointActivationKey.getActivationSpec();
          if (activationSpec.isUseJndi()) {
              try {
                  InitialContext initialContext = new InitialContext();
                  dest = (ActiveMQDestination) initialContext.lookup(activationSpec.getDestination());
              }
              catch (NamingException exc) {
                  throw new ResourceException("JNDI lookup failed for "
                      + activationSpec.getDestination());
              }
          }
          else {
              if ("jakarta.jms.Queue".equals(activationSpec.getDestinationType())) {
                  dest = new ActiveMQQueue(activationSpec.getDestination());
              }
              else if ("jakarta.jms.Topic".equals(activationSpec.getDestinationType())) {
                  dest = new ActiveMQTopic(activationSpec.getDestination());
              }
              else {
                  throw new ResourceException("Unknown destination type: "
                      + activationSpec.getDestinationType());
              }
          }
      }
  
      /**
       * @param c
       */
      public static void safeClose(Connection c) {
          try {
              if (c != null) {
                  LOG.debug("Closing connection to broker");
                  c.close();
              }
          } catch (JMSException e) {
              LOG.trace("failed to close c {}", c, e);
          }
      }
  
      /**
       * @param cc
       */
      public static void safeClose(ConnectionConsumer cc) {
          try {
              if (cc != null) {
                  LOG.debug("Closing ConnectionConsumer");
                  cc.close();
              }
          } catch (JMSException e) {
              LOG.trace("failed to close cc {}", cc, e);
          }
      }
  
      /**
       * 
       */
      public void start() throws ResourceException {
          synchronized (connectWork) {
              if (running)
              return;
          running = true;
  
              if ( connecting.compareAndSet(false, true) ) {
                  LOG.info("Starting");
          serverSessionPool = new ServerSessionPoolImpl(this, endpointActivationKey.getActivationSpec().getMaxSessionsIntValue());
          connect();
              } else {
                  LOG.warn("Ignoring start command, EndpointWorker is already trying to connect");
      }
          }
      }
  
      /**
       * 
       */
      public void stop() throws InterruptedException {
          synchronized (shutdownMutex) {
              if (!running)
                  return;
              running = false;
              LOG.info("Stopping");
              // wake up pausing reconnect attempt
              shutdownMutex.notifyAll();
              try {
                  serverSessionPool.close();
              } catch (Throwable ignored) {
                  LOG.debug("Unexpected error on server session pool close", ignored);
              }
          }
          disconnect();
      }
  
      private boolean isRunning() {
          return running;
      }
  
      private void connect() {
          synchronized ( connectWork ) {
              if (!running) {
                  return;
              }
  
              try {
                  workManager.scheduleWork(connectWork, WorkManager.INDEFINITE, null, null);
              } catch (WorkException e) {
                  running = false;
                  LOG.error("Work Manager did not accept work: ", e);
              }
          }
      }
  
      /**
       * 
       */
      private void disconnect() {
          synchronized ( connectWork ) {
          safeClose(consumer);
          consumer = null;
          safeClose(connection);
          connection = null;
      }
              }
  
      protected void registerThreadSession(Session session) {
          THREAD_LOCAL.set(session);
      }
  
      protected void unregisterThreadSession(Session session) {
          THREAD_LOCAL.set(null);
      }
  
      // for testing
      public void setConnection(ActiveMQConnection activeMQConnection) {
          this.connection = activeMQConnection;
      }
  
      protected ActiveMQConnection getConnection() {
          // make sure we only return a working connection
          // in particular make sure that we do not return null
          // after the resource adapter got disconnected from
          // the broker via the disconnect() method
          synchronized ( connectWork ) {
              return connection;
          }
      }
  
      private String emptyToNull(String value) {
          if (value == null || value.length() == 0) {
              return null;
          }
          return value;
      }
  
  }
  