/*
 * JORAM: Java(TM) Open Reliable Asynchronous Messaging
 * Copyright (C) 2011 - 2025 ScalAgent Distributed Technologies
 *
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
 * License as published by the Free Software Foundation; either
 * version 2.1 of the License, or any later version.
 *
 * This library is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * Lesser General Public License for more details.
 *
 * You should have received a copy of the GNU Lesser General Public
 * License along with this library; if not, write to the Free Software
 * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307
 * USA.
 *
 * Initial developer(s): ScalAgent Distributed Technologies
 * Contributor(s): 
 */
package org.objectweb.joram.mom.dest.jms;

import java.util.ArrayList;
import java.util.Hashtable;
import java.util.List;
import java.util.Map;
import java.util.Properties;

import jakarta.jms.Destination;
import jakarta.jms.JMSException;
import jakarta.jms.MessageConsumer;
import jakarta.jms.MessageListener;
import jakarta.jms.Queue;
import jakarta.jms.Session;
import jakarta.jms.Topic;

import org.objectweb.joram.mom.dest.AcquisitionDaemon;
import org.objectweb.joram.mom.dest.ReliableTransmitter;
import org.objectweb.util.monolog.api.BasicLevel;
import org.objectweb.util.monolog.api.Logger;

import fr.dyade.aaa.common.Daemon;
import fr.dyade.aaa.common.Debug;

/**
 * Acquisition daemon for the JMS acquisition bridge.
 */
public class JMSAcquisition implements AcquisitionDaemon {
  private static final Logger logger = Debug.getLogger(JMSAcquisition.class.getName());
  private static final boolean DEBUG = logger.isLoggable(BasicLevel.DEBUG);

  private static final String DESTINATION_NAME_PROP = "jms.DestinationName";

  private static final String UPDATE_PERIOD_PROP = "jms.ConnectionUpdatePeriod";

  private static final String ROUTING_PROP = "jms.Routing";
  
  private static final String DURABLE_SUBSCRIPTION_PROP = "jms.DurableSubscriptionName";
  
  private static final String SELECTOR_PROP = "jms.Selector";

  private static ConnectionUpdater connectionUpdater;

  private ReliableTransmitter transmitter;

  // Use a vector as it is used by 2 different threads
  private Map<String, JmsListener> listeners = new Hashtable<String, JmsListener>();

  /** If routing prop has been set, it defines a list of connection to use. */
  private List<String> connectionNames = null;

  private volatile boolean closing = false;

  /** The JNDI name of the foreign JMS destination. */
  private String destName;

  /** Foreign JMS destination object. */
  private Destination dest = null;

  /** Selector for filtering messages. */
  protected String selector;
  
  private String durableSubscriptionName = null;

  public void start(Properties properties, ReliableTransmitter transmitter) {
    if (DEBUG)
      logger.log(BasicLevel.DEBUG, "Start JMSAcquisition.");
    
    this.transmitter = transmitter;

    destName = properties.getProperty(DESTINATION_NAME_PROP);
    if (destName == null) {
      throw new IllegalArgumentException("Missing Destination JNDI name.");
    }

    long updatePeriod = 5000L;
    try {
      if (properties.containsKey(UPDATE_PERIOD_PROP)) {
        updatePeriod = Long.parseLong(properties.getProperty(UPDATE_PERIOD_PROP));
      }
    } catch (NumberFormatException nfe) {
      logger.log(BasicLevel.ERROR, "Property " + UPDATE_PERIOD_PROP
          + "could not be parsed properly, use default value.", nfe);
    }

    connectionNames = null;
    if (properties.containsKey(ROUTING_PROP)) {
      connectionNames = JMSConnectionService.convertToList(properties.getProperty(ROUTING_PROP));
    }

    durableSubscriptionName = null;
    if (properties.containsKey(DURABLE_SUBSCRIPTION_PROP)) {
      durableSubscriptionName = properties.getProperty(DURABLE_SUBSCRIPTION_PROP);
    }

    selector = null;
    if (properties.containsKey(SELECTOR_PROP)) {
      selector = properties.getProperty(SELECTOR_PROP);
    }

    if (connectionUpdater == null) {
      connectionUpdater = new ConnectionUpdater(updatePeriod);
    }
    connectionUpdater.addUpdateListener(this);
  }

  public void stop() {
    if (DEBUG)
      logger.log(BasicLevel.DEBUG, "Stop JMSAcquisition.");

    connectionUpdater.removeUpdateListener(this);
    
    closing = true;
    synchronized (listeners) {
      for (JmsListener listener : listeners.values()) {
        try {
          if (DEBUG)
            logger.log(BasicLevel.DEBUG, "Close JMS session: " + listener.session);

          listener.session.close();
          
          if (DEBUG)
            logger.log(BasicLevel.DEBUG, "JMS session closed: " + listener.session);
        } catch (JMSException exc) {
          if (logger.isLoggable(BasicLevel.INFO)) {
            logger.log(BasicLevel.INFO, "Error while stopping JmsAcquisition.", exc);
          }
        }
        // Remove ExceptionListener from JMSModule
        listener.connection.removeExceptionListener(listener);
      }
      
      listeners.clear();
      closing = false;
    }
  }

  /**
   * Create a new JMS consumer for each connection available.
   */
  public synchronized void updateConnections() {
    List<JMSModule> connections = JMSConnectionService.getInstance().getConnections();

    for (JMSModule connection : connections) {
      if (!listeners.containsKey(connection.getName())) {
        if (connectionNames == null || connectionNames.contains(connection.getName())) {
          if (DEBUG)
            logger.log(BasicLevel.DEBUG,
                       "Creating a new consumer for connection: " + connection.getName(), new Exception());
          
          // /!\ Be careful, do not close session, or consumer. They are use through the JmsListener
          // object beyond this method call.
          MessageConsumer consumer = null;
          try {
            Session session = connection.getCnx().createSession(true, Session.AUTO_ACKNOWLEDGE);
            dest = (Destination) connection.retrieveJndiObject(destName);
            if (dest instanceof Queue) {
              consumer = session.createConsumer(dest, selector);
            } else {
              if (durableSubscriptionName != null) {
                consumer = session.createDurableSubscriber((Topic) dest, durableSubscriptionName, selector, false);
              } else {
                consumer = session.createConsumer(dest, selector);
              }
            }
            if (DEBUG)
              logger.log(BasicLevel.DEBUG, "setConsumer: consumer=" + consumer);  
            JmsListener listener = new JmsListener(session, connection);
            consumer.setMessageListener(listener);
            connection.getCnx().start();
            connection.addExceptionListener(listener);
            listeners.put(connection.getName(), listener);
          } catch (Exception exc) {
            if (DEBUG)
              logger.log(BasicLevel.ERROR,
                         "Error while starting consumer on connection " + connection.getCnxFactName(), exc);
            else
              logger.log(BasicLevel.ERROR,
                         "Error while starting consumer on connection " + connection.getCnxFactName() + ": " + exc.getMessage());
            if (consumer != null)
              try {
                consumer.close();
              } catch (JMSException e) {}
          }
        }
      }
    }
  }

  class JmsListener implements MessageListener {
    private JMSModule connection;
    private Session session;

    public JmsListener(Session session, JMSModule connection) {
      this.session = session;
      this.connection = connection;
    }

    /**
     * Implements the {@link MessageListener} interface for processing the
     * asynchronous deliveries coming from the foreign JMS server.
     */
    public void onMessage(jakarta.jms.Message jmsMessage) {
      if (DEBUG)
        logger.log(BasicLevel.DEBUG, "JMSAcquisition[" + connection.getName() + "].onMessage(" + jmsMessage + ')');

      try {
        org.ow2.joram.jakarta.jms.Message clientMessage = null;
        try {
          clientMessage = org.ow2.joram.jakarta.jms.Message.convertJMSMessage(jmsMessage);
          org.ow2.joram.jakarta.jms.Message.prepareJMSMessage(clientMessage);
        } catch (JMSException conversionExc) {
          // Conversion error: denying the message.
          if (DEBUG)
            logger.log(BasicLevel.WARN,
                       "JMSAcquisition[" + connection.getName() + "].onMessage, rollback, can not convert message.", conversionExc);
          else
            logger.log(BasicLevel.WARN,
                       "JMSAcquisition[" + connection.getName() + "].onMessage, rollback, can not convert message: " + conversionExc.getMessage());

          session.rollback();
          return;
        }
        transmitter.transmit(clientMessage.getMomMsg(), jmsMessage.getJMSMessageID());

        if (DEBUG)
          logger.log(BasicLevel.DEBUG, "JMSAcquisition[" + connection.getName() + "].onMessage: Try to commit.");

        session.commit();
      } catch (JMSException exc) {
        // Commit or rollback failed: nothing to do.
        if (DEBUG)
          logger.log(BasicLevel.ERROR, "JMSAcquisition[" + connection.getName() + "].onMessage(" + jmsMessage + ')', exc);
        else
          logger.log(BasicLevel.ERROR, "JMSAcquisition[" + connection.getName() + "].onMessage(" + jmsMessage + "): " + exc.getMessage());
      } catch (Throwable t) {
        if (DEBUG)
          logger.log(BasicLevel.ERROR, "JMSAcquisition[" + connection.getName() + "].onMessage(" + jmsMessage + ')', t);
        else
          logger.log(BasicLevel.ERROR, "JMSAcquisition[" + connection.getName() + "].onMessage(" + jmsMessage + "): " + t.getMessage());
      }
    }

    public void onException(JMSException exception) {
      if (DEBUG)
        logger.log(BasicLevel.DEBUG, "JMSAcquisition[" + connection.getName() + "]: Consumer error for session " + session);
      
      if (!closing) {
        listeners.remove(connection.getName());
      }
    }

  }

  /**
   * Daemon used to periodically update the pool of connections known by the
   * acquisition destinations.
   */
  private static class ConnectionUpdater extends Daemon {
    // List of all acquisition modules using to update
    private List<JMSAcquisition> modules = new ArrayList<JMSAcquisition>();

    private long period;

    /** Constructs a <code>ConnectionUpdater</code> thread. */
    protected ConnectionUpdater(long updatePeriod) {
      super("JMS_ConnectionUpdater", logger);
      setDaemon(false);
      period = updatePeriod;
      if (DEBUG) {
        logmon.log(BasicLevel.DEBUG, "ConnectionUpdater<init>");
      }
    }

    private boolean closing = false;
    
    /** The daemon's loop. */
    public void run() {
      if (DEBUG) {
        logmon.log(BasicLevel.DEBUG, "run()");
      }

      try {
        boolean firstTime = true;
        while (running) {
          if (DEBUG)
            logmon.log(BasicLevel.DEBUG, "update connections in " + period + " ms");

          canStop = true;
          if (firstTime) {
            firstTime = false;
          } else {
            try {
              Thread.sleep(period);
            } catch (InterruptedException exc) {
              if (DEBUG) {
                logmon.log(BasicLevel.DEBUG, "Thread interrupted.");
              }
              continue;
            }
          }
          canStop = false;

          if (DEBUG) {
            logmon.log(BasicLevel.DEBUG, "update connections");
          }

          // The ConnectionHandler is closing, modules can be locked by removeUpdateListener method.
          if (! closing) {
            synchronized (modules) {
              if (modules.size() == 0) stop();

              for (JMSAcquisition listener : modules) {
                listener.updateConnections();
              }
            }
          } else {
            break;
          }
        }
      } finally {
        finish();
      }
    }

    /** Shuts the daemon down. */
    public void shutdown() {
      interrupt();
    }

    /** Releases the daemon's resources. */
    public void close() {
    }

    protected void addUpdateListener(JMSAcquisition module) {
      synchronized (modules) {
        modules.add(module);
        if (modules.size() == 1) {
          start();
        }
      }
      module.updateConnections();
    }

    protected void removeUpdateListener(JMSAcquisition module) {
      // The closing field avoid a potential deadlock with daemon during stop.
      closing = true;
      try {
        synchronized (modules) {
          modules.remove(module);
          if (modules.size() == 0) stop();
        }
      } finally {
        closing = false;
      }
    }
  }
}
