/*
 * JORAM: Java(TM) Open Reliable Asynchronous Messaging
 * Copyright (C) 2004 - 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.
 */
package org.ow2.joram.jakarta.connector;

import jakarta.jms.JMSException;
import jakarta.jms.Message;
import jakarta.jms.MessageListener;
import jakarta.jms.Session;
import jakarta.jms.XAConnection;
import jakarta.jms.XASession;
import jakarta.resource.spi.endpoint.MessageEndpoint;
import jakarta.resource.spi.endpoint.MessageEndpointFactory;
import jakarta.resource.spi.work.Work;
import jakarta.resource.spi.work.WorkManager;
import javax.transaction.xa.XAResource;

import org.objectweb.util.monolog.api.BasicLevel;
import org.objectweb.util.monolog.api.Logger;

import fr.dyade.aaa.common.Debug;

/**
 * An <code>InboundSession</code> instance is responsible for processing
 * delivered messages within a <code>jakarta.resource.spi.Work</code> instance,
 * and passing them to a set of application server endpoints.
 */
class InboundSession implements jakarta.jms.ServerSession, Work, jakarta.jms.MessageListener {
  private static final Logger logger = Debug.getLogger(InboundSession.class.getName());
  private static final boolean DEBUG = logger.isLoggable(BasicLevel.DEBUG);
  
  /** <code>InboundConsumer</code> instance this session belongs to. */
  private InboundConsumer consumer;

  /** Application server's <code>WorkManager</code> instance. */
  private WorkManager workManager;
  /** Application's endpoints factory. */
  private MessageEndpointFactory endpointFactory; 

  /**
   * <code>Session</code> instance dedicated to processing
   * the delivered messages.
   */
  private Session session;
  /** <code>XAResource</code> instance, if any. */
  private XAResource xaResource = null;


  /**
   * Constructs an <code>InboundSession</code> instance.
   *
   * @param consumer         InboundConsumer creating this session.
   * @param workManager      Application server's <code>WorkManager</code>
   *                         instance.
   * @param endpointFactory  Application's endpoints factory.
   * @param cnx              Connection to the underlying JORAM server.
   * @param transacted       <code>true</code> if deliveries occur within a 
   *                         XA transaction.
   */
  InboundSession(InboundConsumer consumer, WorkManager workManager, MessageEndpointFactory endpointFactory,
                 XAConnection cnx, boolean transacted, int ackMode) {
    if (DEBUG)
      logger.log(BasicLevel.DEBUG,
                 "InboundSession(" + consumer + "," + workManager + "," + endpointFactory + "," + cnx + "," + transacted + "," + ackMode + ")");
    
    this.consumer = consumer;
    this.workManager = workManager;
    this.endpointFactory = endpointFactory;

    try {
      if (transacted) {
        session = cnx.createXASession();
        xaResource = ((XASession) session).getXAResource();
        if (DEBUG)
          logger.log(BasicLevel.DEBUG, "InboundSession xaResource = " + xaResource);
      } else {
        session = cnx.createSession(false, ackMode);
      }

      if (DEBUG)
        logger.log(BasicLevel.DEBUG, "InboundSession session = " + session);
      session.setMessageListener(this);
    } catch (JMSException exc) {}
  }


  /**
   * Provides the wrapped <code>Session</code> instance for
   * processing delivered messages.
   *
   * @exception JMSException  Never thrown.
   */
  public Session getSession() throws JMSException {
    return session;
  }

  /**
   * Notifies that the messages are ready to be processed.
   *
   * @exception JMSException  If submitting the processing work fails.
   */
  public void start() throws JMSException {
    try {
      if (DEBUG)
        logger.log(BasicLevel.DEBUG, "ServerSession submits Work instance.");
      workManager.scheduleWork(this);
    } catch (Exception exc) {
      throw new JMSException("Can't start the adapter session for processing "
                             + "the delivered messages: " + exc);
    }
  }

  /** <code>jakarta.resource.spi.Work</code> method, not effective. */
  public void release() {
    try {
      session.close();
    } catch (JMSException exc) { }
  }

  /** Runs the wrapped session for processing the messages. */
  public void run() {
    if (DEBUG)
      logger.log(BasicLevel.DEBUG, "ServerSession runs wrapped Session.");
    session.run();
    consumer.releaseSession(this);
  }

  /** Forwards a processed message to an endpoint. */
  public void onMessage(Message message) {
    if (DEBUG)
      logger.log(BasicLevel.DEBUG, this + " onMessage(" + message + ")");

    MessageEndpoint endpoint = null;
    try {
      if (DEBUG)
        logger.log(BasicLevel.DEBUG, "ServerSession passes message to listener.");
      endpoint = endpointFactory.createEndpoint(xaResource);
      ((MessageListener) endpoint).onMessage(message);
      endpoint.release();
    } catch (Exception exc) {
      try {
        // try to clean the context for next invocation
        if (endpoint != null) endpoint.release();
      } catch (Exception e) {
        // ignore the exception
        if (logger.isLoggable(BasicLevel.INFO))
          logger.log(BasicLevel.INFO, "endpoint.release: ", e);
      }
      logger.log(BasicLevel.ERROR, "InboundSession.onMessage: ", exc);
      throw new java.lang.IllegalStateException("Could not get endpoint instance: ", exc);
    }
  }
}
