/*
 * JORAM: Java(TM) Open Reliable Asynchronous Messaging
 * Copyright (C) 2001 - 2025 ScalAgent Distributed Technologies
 * Copyright (C) 1996 - 2000 Dyade
 *
 * 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
 */
package org.objectweb.joram.client.jms.connection;

import java.util.Hashtable;
import java.util.Map;
import java.util.Set;
import java.util.Timer;
import java.util.TimerTask;
import java.util.Vector;

import javax.jms.IllegalStateException;
import javax.jms.InvalidDestinationException;
import javax.jms.JMSException;
import javax.jms.JMSSecurityException;

import org.objectweb.joram.client.jms.Connection;
import org.objectweb.joram.client.jms.MessageProducer;
import org.objectweb.joram.client.jms.Session;
import org.objectweb.joram.shared.client.AbstractJmsReply;
import org.objectweb.joram.shared.client.AbstractJmsRequest;
import org.objectweb.joram.shared.client.CommitRequest;
import org.objectweb.joram.shared.client.ConsumerMessages;
import org.objectweb.joram.shared.client.ConsumerReceiveRequest;
import org.objectweb.joram.shared.client.ConsumerUnsetListRequest;
import org.objectweb.joram.shared.client.MomExceptionReply;
import org.objectweb.joram.shared.client.PingRequest;
import org.objectweb.joram.shared.client.ProducerMessages;
import org.objectweb.joram.shared.client.SessDenyRequest;
import org.objectweb.joram.shared.excepts.MomException;
import org.objectweb.util.monolog.api.BasicLevel;
import org.objectweb.util.monolog.api.Logger;

import fr.dyade.aaa.common.Debug;

public class RequestMultiplexer {
  private static Logger logger = Debug.getLogger(RequestMultiplexer.class.getName());
  private static final boolean DEBUG = logger.isLoggable(BasicLevel.DEBUG);

  private static class Status {
    public static final int OPEN = 0;
    public static final int CLOSE = 1;
    
    private static final String[] names = {"OPEN", "CLOSE"};

    public static String toString(int status) {
      return names[status];
    }
  }
  
  /**
   * Converts a {@link MomExceptionReply} to the corresponding
   * {@link JMSException}.
   * 
   * @param excReply the MOM reply to convert
   * @return the corresponding Exception
   */
  public static JMSException buildJmsException(MomExceptionReply excReply) {
    JMSException jmsExc = null;
    int excType = excReply.getType();
    if (excType == MomExceptionReply.AccessException) {
      jmsExc = new JMSSecurityException(excReply.getMessage());
    } else if (excType == MomExceptionReply.DestinationException) {
      jmsExc = new InvalidDestinationException(excReply.getMessage());
    } else {
      jmsExc = new JMSException(excReply.getMessage());
    }
    return jmsExc;
  }

  private Connection cnx;

  private volatile int status;

  private RequestChannel channel;

  private Map<Integer, ReplyListener> requestsTable;
  
  private Map<Integer, CompletionListener> completionListeners;
  
  private int requestCounter;

  private DemultiplexerDaemon demtpx;
  
  public boolean checkDemultiplexerDaemon() {
    return demtpx.isCurrentThread();
  }
  
  public boolean checkCLSession(Session session) {
    if (DEBUG)
        logger.log(BasicLevel.DEBUG, "checkCLSession runningCL=" + runningCL);
    
    if (checkDemultiplexerDaemon()) {
      // Check if there is a CompletionListener running
      if (runningCL != null)
        return (session == runningCL.session);
      return false;
    } else {
      // The current thread is not the demtpx so there is no
      // CompletionListener running.
      return false;
    }
  }
  
  public boolean checkCLMessageProducer(Session session, MessageProducer mp) {
    if (runningCL != null && mp != null)
      return (checkCLSession(session) && mp.equals(runningCL.messageProducer));
    return false;
  }

  private static Timer timer;
  private static int timerInUse;

  private synchronized static void createTimer() {
    if (timer == null) {
      timer = new Timer("RequestMultiplexer.Timer");
      timerInUse = 0;
    }
    timerInUse++;
    if (DEBUG)
      logger.log(BasicLevel.DEBUG, "createTimer: timerInUse = " + timerInUse);
  }

  private synchronized static void cancelTimer() {
    if (timer != null) {
      timerInUse--;
      if (timerInUse < 1) {
        timer.cancel();
        timer = null;
        timerInUse = 0;
      }
    }
    if (DEBUG)
      logger.log(BasicLevel.DEBUG, "cancelTimer: timerInUse = " + timerInUse);
  }

  private synchronized static Timer getTimer() {
    return timer;
  }

  /**
   * The task responsible for keeping
   * the connection alive.
   */
  private HeartBeatTask heartBeatTask;

  private javax.jms.ExceptionListener exceptionListener;

  /**
   * The date of the last request
   */
  private volatile long lastRequestDate;
  
  public RequestMultiplexer(Connection cnx,
                            RequestChannel channel,
                            long heartBeat) throws JMSException {
    this.channel = channel;
    this.cnx = cnx; 
    requestsTable = new Hashtable<Integer, ReplyListener>();
    completionListeners = new Hashtable<Integer, CompletionListener>();
    requestCounter = 0;
    createTimer();
    channel.setTimer(getTimer());
    try {
      channel.connect();
    } catch (JMSException exc) {
      throw exc;
    } catch (Exception exc) {
      // Wraps the incoming exception
      if (DEBUG)
        logger.log(BasicLevel.DEBUG, "", exc);
      cancelTimer();
      throw new JMSException(exc.toString());
    }
    
    demtpx = new DemultiplexerDaemon();
    demtpx.start();
    setStatus(Status.OPEN);
    
    if (heartBeat > 0) {
      heartBeatTask = new HeartBeatTask(heartBeat);
      lastRequestDate = System.currentTimeMillis();
      try {
        heartBeatTask.start();
      } catch (Exception exc) {
        if (DEBUG)
          logger.log(BasicLevel.DEBUG, "", exc);
        throw new JMSException(exc.toString());
      }
    }
  }

  private void setStatus(int status) {
    if (DEBUG)
      logger.log(BasicLevel.DEBUG, "RequestMultiplexer.setStatus(" + Status.toString(status) + ')');
    this.status = status;
  }
  
  public boolean isClosed() {
    return status == Status.CLOSE;
  }
  
  public void closing() {
    channel.closing();
  }

  public void setExceptionListener(javax.jms.ExceptionListener exceptionListener) {
    this.exceptionListener = exceptionListener;
  }

  public javax.jms.ExceptionListener getExceptionListener() {
    return exceptionListener;
  }

  public void sendRequest(AbstractJmsRequest request) throws JMSException {
    sendRequest(request, null, null);
  }

  public void sendRequest(AbstractJmsRequest request, ReplyListener listener, CompletionListener completionListener) throws JMSException {
    synchronized (this) {
      if (status == Status.CLOSE)
        throw new IllegalStateException("Connection closed");

      if (requestCounter == Integer.MAX_VALUE) {
        requestCounter = 0;
      }

      request.setRequestId(requestCounter++);

      if (listener != null) {
        requestsTable.put(Integer.valueOf(request.getRequestId()), listener);
      }
      
      if (completionListener != null) {
        if ((request instanceof ProducerMessages) || (request instanceof CommitRequest)) {
          // TODO: Uses request.getClassId() == AbstractJmsMessage.PRODUCER_MESSAGES..
          completionListeners.put(request.getRequestId(), completionListener);
        }
      }

      if (heartBeatTask != null) {
        lastRequestDate = System.currentTimeMillis();
      }
    }

    try {
      channel.send(request);
    } catch (Exception exc) {
      if (DEBUG)
        logger.log(BasicLevel.DEBUG, "", exc);
      JMSException jmsExc = new JMSException(exc.toString());
      jmsExc.setLinkedException(exc);
      throw jmsExc;
    }
  }
  
  public void setMultiThreadSync(int delay, int threshold) {
    channel = new MultiThreadSyncChannel(channel, delay, threshold);
  }

  /**
   * Not synchronized because it would possibly
   * deadlock with some reply listeners 
   * (actually requestors).
   */
  public void close() {
    if (DEBUG)
      logger.log(BasicLevel.DEBUG, "RequestMultiplexer.close()");
    
    synchronized (this) {
      if (status == Status.CLOSE)
        return;
      // Immediately set the status as no error
      // can be thrown. This enables to release
      // the lock and avoid any dead lock
      // with the demultiplexer thread that
      // calls close() when interrupted.
      setStatus(Status.CLOSE);
    }

    if (heartBeatTask != null) heartBeatTask.cancel();
    cancelTimer();
    channel.close();
    demtpx.stop();

    if (DEBUG)
      logger.log(BasicLevel.DEBUG, " -> requestsTable=" + requestsTable);
    
    // The requests table can't be accessed
    // either by an external thread (status CLOSE)
    // or by the internal demultiplexer thread (stopped).
    
    cleanup();
  }

  /**
   * Aborts all running requests.
   * Used by:
   * 1- close()
   * 2- the connector layer (OutboundConnection.cleanup())
   */
  public void cleanup() {
    // Create first a copy of the current keys registered into the requests table.
    Integer[] requestIds;
    synchronized (requestsTable) {
      Set<Integer> keySet = requestsTable.keySet();
      requestIds = new Integer[keySet.size()];
      keySet.toArray(requestIds);
    }
    
    for (int i = 0; i < requestIds.length; i++) {
      ReplyListener rl = requestsTable.get(requestIds[i]);
      // The listener may be null because the table
      // may have been modified meanwhile.
      if (rl != null) {
        rl.replyAborted(requestIds[i].intValue());
      }
    }
    requestsTable.clear();
    completionListeners.clear();
  }
  
  public void replyAllError(MomExceptionReply exc) {
    // Create first a copy of the current keys
    // registered into the requests table.
    Integer[] requestIds;
    synchronized (requestsTable) {
      Set<Integer> keySet = requestsTable.keySet();
      requestIds = new Integer[keySet.size()];
      keySet.toArray(requestIds);
    }
    for (int i = 0; i < requestIds.length; i++) {
      ReplyListener rl = requestsTable.get(requestIds[i]);
      // The listener may be null because the table
      // may have been modified meanwhile.
      if (rl != null) {
      	rl.errorReceived(requestIds[i].intValue(), exc);
      }
    }
    
    // completionListeners
    synchronized (completionListeners) {
      Set<Integer> keySet = completionListeners.keySet();
      requestIds = new Integer[keySet.size()];
      keySet.toArray(requestIds);
    }
    for (int i = 0; i < requestIds.length; i++) {
      CompletionListener cl = completionListeners.get(requestIds[i]);
      if (cl != null)
        cl.onException(new Exception(exc.getMessage()));
    }
    completionListeners.clear();
    requestsTable.clear();
  }

  /**
   * Not synchronized because it would possibly
   * deadlock with some reply listeners 
   * (actually requestors).
   */
  public void abortRequest(int requestId) {
    if (DEBUG)
      logger.log(BasicLevel.DEBUG, "RequestMultiplexer.abortRequest(" + requestId + ')');
    
    ReplyListener rl = doAbortRequest(requestId);    
    if (rl != null) {
      rl.replyAborted(requestId);
    }
  }
  
  private synchronized ReplyListener doAbortRequest(int requestId) {
    if (DEBUG)
      logger.log(BasicLevel.DEBUG, "RequestMultiplexer.doAbortRequest(" + requestId + ')');
    
    if (status == Status.CLOSE) return null;
    return requestsTable.remove(Integer.valueOf(requestId));
  }

  /**
   * Not synchronized because it may be called by the
   * demultiplexer during a concurrent close. It would deadlock
   * as the close waits for the demultiplexer to stop.
   */
  private void route(AbstractJmsReply reply, boolean isCompletionListener) {
    if (DEBUG)
      logger.log(BasicLevel.DEBUG, "RequestMultiplexer.route(" + reply + ')');
    
    int requestId = reply.getCorrelationId();
    Integer requestKey = Integer.valueOf(requestId);
    ReplyListener rl = requestsTable.get(requestKey);
    if (reply instanceof MomExceptionReply) {
      MomExceptionReply excReply = (MomExceptionReply) reply;
      if (rl instanceof ErrorListener) {
        ((ErrorListener) rl).errorReceived(requestId, excReply);
      } else {
        // The listener is null or doesn't implement ErrorListener
        if (exceptionListener != null) {
          exceptionListener.onException(buildJmsException(excReply));
        }
      }
    } else {
      if (DEBUG)
        logger.log(BasicLevel.DEBUG, " -> rl = " + rl);
      if (rl != null) {
        try {
          if (!isCompletionListener && rl.replyReceived(reply)) {
            requestsTable.remove(requestKey);
          }
        } catch (AbortedRequestException exc) {
          logger.log(BasicLevel.WARN, " -> Request aborted: " + requestId);
          abortReply(reply);
        }
      } else {
        if (DEBUG)
          logger.log(BasicLevel.DEBUG, " -> Listener not found for the reply: " + requestId);
        abortReply(reply);
      }
    }
  }

  private void abortReply(AbstractJmsReply reply) {
    if (DEBUG)
      logger.log(BasicLevel.DEBUG, "RequestMultiplexer.abortReply(" + reply + ')');
    
    if (reply instanceof ConsumerMessages) {
      deny((ConsumerMessages)reply);
    }
    // Else nothing to do.
  }

  // Denies received messages.
  public void deny(ConsumerMessages messages) {
    if (DEBUG)
      logger.log(BasicLevel.DEBUG, "RequestMultiplexer.deny(" + messages + ')');

    Vector msgList = messages.getMessages();
    Vector<String> ids = new Vector<>();
    for (int i = 0; i < msgList.size(); i++) {
      ids.addElement(((org.objectweb.joram.shared.messages.Message) msgList.elementAt(i)).id);
    }
    SessDenyRequest deny = new SessDenyRequest(messages.comesFrom(),
                                               ids,
                                               messages.getQueueMode());
    try {
      sendRequest(deny);
    } catch (JMSException exc) {
      if (DEBUG)
        logger.log(BasicLevel.DEBUG, "Connection is closed", exc);
      // Connection failure
      // Nothing to do
    }
  }

  // Denies aborted request.
  public void denyRequest(ConsumerReceiveRequest request) {
    if (DEBUG)
      logger.log(BasicLevel.DEBUG, "RequestMultiplexer.denyRequest(" + request.getRequestId() + ')');

    ConsumerUnsetListRequest unsetLR = new ConsumerUnsetListRequest(request.getQueueMode());
    unsetLR.setTarget(request.getTarget());
    unsetLR.setCancelledRequestId(request.getRequestId());

    try {
      sendRequest(unsetLR);
    } catch (JMSException exc) {
      if (DEBUG)
        logger.log(BasicLevel.DEBUG, "Connection is closed", exc);
      // Connection failure
      // Nothing to do
    }
  }

  class onExceptionRunner implements Runnable {
    Exception exc;

    onExceptionRunner(Exception exc) {
      this.exc = exc;
    }

    public void run() {
      onException(exc);
    }
  }

  private void onException(Exception exc) {
    if (DEBUG)
      logger.log(BasicLevel.DEBUG, "RequestMultiplexer.onException(" + exc + ')');
    JMSException jmsExc;
    if (exc instanceof JMSException) {
      jmsExc = (JMSException) exc;
    } else {
      jmsExc = new IllegalStateException(exc.getMessage());
    }
    if (exceptionListener != null)
      exceptionListener.onException(jmsExc);
  }

  public void schedule(TimerTask task, long period) {
    Timer timer = getTimer();
    if (timer != null) {
      try {
        timer.schedule(task, period);
      } catch (Exception exc) {
        if (logger.isLoggable(BasicLevel.ERROR))
          logger.log(BasicLevel.ERROR, "", exc);
      }
    }
  }

  public void setDemultiplexerDaemonName(String name) {
    demtpx.setName(name);
  }
  
  public String getDemultiplexerDaemonName() {
    return demtpx.getName();
  }

  CompletionListener runningCL = null;

  private class DemultiplexerDaemon extends fr.dyade.aaa.common.Daemon {
    DemultiplexerDaemon() {
      // The real name is set later when
      // the proxy id and connection id are known
      // see setDemultiplexerDaemonName()
      super("Connection#?", logger);
    }
    
    public void run() {
      try {
        loop:
        while (running) {
          canStop = true;
          AbstractJmsReply reply;
          try {
            reply = channel.receive();
          } catch (Exception exc) {
            if (DEBUG)
              logger.log(BasicLevel.DEBUG, "Exception during receive", exc);
            // Check if the connection is not already
            // closed (the exception may occur as a consequence
            // of a closure or at the same time as an independant
            // close call).
            if (! isClosed()) {
            	replyAllError(new MomExceptionReply(new MomException(exc.getMessage())));
            	
            	RequestMultiplexer.this.close();
            	
              // The connection close() must be
              // called by another thread. Calling it with
              // this thread (demultiplexer daemon) could
              // lead to a deadlock if another thread called
              // close() just before.
              Closer closer = new Closer(exc);
              new Thread(closer, "RequestMultiplexer.Closer").start();
            } else {
              // Else it means that the connection is already closed
              // Runs the onException in a separate thread in order to avoid
              // deadlock in connector onException (synchronized).
              onExceptionRunner oer = new onExceptionRunner(exc);
              new Thread(oer, "RequestMultiplexer.onException").start();
            }
            
            break loop;
          }
          canStop = false; 
          
          boolean isCompletionListener = false;
          if (!completionListeners.isEmpty()) {
            CompletionListener cl = completionListeners.remove(reply.getCorrelationId());
            if (cl != null) {
              try {
                runningCL = cl;
                if (reply instanceof MomExceptionReply) {
                  cl.onException(new Exception(((MomExceptionReply) reply).getMessage()));
                } else {
                  cl.onCompletion();
                }
              } catch (Throwable t) {
                logger.log(BasicLevel.ERROR,
                           "Error during completion listener execution.", t);
              } finally {
                runningCL = null;
              }
              isCompletionListener = true;
            }
          }
          
          route(reply, isCompletionListener);
          
          if (isCompletionListener)
            requestsTable.remove(reply.getCorrelationId());
          
          if (!running && isClosed()) {
            if (DEBUG)
              logger.log(BasicLevel.DEBUG, "DemultiplexerDaemon ended and Socket closed.");
            onExceptionRunner oer = new onExceptionRunner(new Exception("DemultiplexerDaemon ended and Socket closed."));
            new Thread(oer, "RequestMultiplexer.onException").start();
          }
        }
      } finally {
        finish();
      }
    }
    
    /**
     * Enables the daemon to stop itself.
     */
    public void stop() {
      if (isCurrentThread()) {
        finish();
      } else {
        super.stop();
      }
    }

    protected void shutdown() {}

    protected void close() {}
  }
  
  private class Closer implements Runnable {
    private Exception exc;
    
    Closer(Exception e) {
      exc = e;
    }
    
    public void run() {
      try {
        RequestMultiplexer.this.cnx.close();
      } catch (JMSException exc2) {
        if (logger.isLoggable(BasicLevel.WARN))
          logger.log(BasicLevel.WARN, "Error during close", exc2);
      }
      onException(exc);
    }
  }

  /**
   * Timer task responsible for sending a ping message
   * to the server if no request has been sent during 
   * the specified timeout ('cnxPendingTimer' from the
   * factory parameters).
   */
  private class HeartBeatTask extends TimerTask {    

    private long heartBeat;

    HeartBeatTask(long heartBeat) {
      this.heartBeat = heartBeat;
    }

    public void run() {
      try {
        long date = System.currentTimeMillis();        
        if ((date - lastRequestDate) > heartBeat) {
          sendRequest(new PingRequest());
        }
      } catch (Exception exc) {
        if (logger.isLoggable(BasicLevel.WARN))
          logger.log(BasicLevel.WARN, "HeartBeatTask", exc);
      }
    }

    public void start() throws Exception {
      Timer timer = getTimer();
      if (timer != null)
        timer.schedule(this, heartBeat/4, heartBeat/4);
    }
  }
}