/*
 * JORAM: Java(TM) Open Reliable Asynchronous Messaging
 * Copyright (C) 2003 - 2026 ScalAgent Distributed Technologies
 * Copyright (C) 2004 France Telecom R&D
 * Copyright (C) 2003 - 2004 Bull SA
 *
 * 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): Frederic Maistre (Bull)
 * Contributor(s): ScalAgent Distributed Technologies
 */
package org.objectweb.joram.mom.proxies; 

import java.io.IOException;
import java.util.Hashtable;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Vector;

import org.objectweb.joram.mom.util.JoramHelper;
import org.objectweb.joram.shared.client.AbstractJmsReply;
import org.objectweb.joram.shared.client.XACnxPrepare;
import org.objectweb.util.monolog.api.BasicLevel;
import org.objectweb.util.monolog.api.Logger;

import fr.dyade.aaa.agent.AgentId;
import fr.dyade.aaa.agent.AgentServer;
import fr.dyade.aaa.common.Debug;
import fr.dyade.aaa.common.encoding.Decoder;
import fr.dyade.aaa.common.encoding.Encodable;
import fr.dyade.aaa.common.encoding.EncodableFactory;
import fr.dyade.aaa.common.encoding.EncodableHelper;
import fr.dyade.aaa.common.encoding.Encoder;

/**
 * The <code>ClientContext</code> class holds the data related to a connection context.
 * Contexts are only reloaded during a broker restart to allow them to be cleaned up, they
 * are then immediately deleted.
 */
public final class ClientContext implements java.io.Serializable, Encodable {
  /** define serialVersionUID for interoperability */
  private static final long serialVersionUID = 1L;
  
  /** logger */
  public final static Logger logger = Debug.getLogger(ClientContext.class.getName());
  private static final boolean DEBUG = logger.isLoggable(BasicLevel.DEBUG);

  /** Context identifier. */
  private int id;
  
  /** Returns the identifier of the context. */
  int getId() {
    return id;
  }
  
  /** Vector of temporary destinations. */
  private Vector<AgentId> tempDestinations;

  /** Adds a temporary destination identifier. */
  void addTemporaryDestination(AgentId destId) {
    tempDestinations.add(destId);
    setModified();
  }
   
  Iterator<AgentId> getTempDestinations() {
    //Creates an enumeration not backed by tempDestinations
    Vector<AgentId> tempDests = new Vector<AgentId>();
    for (Iterator<AgentId> dests = tempDestinations.iterator(); dests.hasNext();) {
      tempDests.addElement(dests.next());
    }
    return tempDests.iterator();
  }

  /** Removes a temporary destination identifier. */
  void removeTemporaryDestination(AgentId destId) {
    deliveringQueues.remove(destId);
    tempDestinations.remove(destId);
    setModified();
  }

  /** Identifiers of queues delivering messages. */
  private Hashtable<AgentId, AgentId> deliveringQueues;
  
  /** Adds, if not already, the identifier of a delivering queue. */ 
  void addDeliveringQueue(AgentId queueId) {
    if (deliveringQueues.get(queueId) == null) {
      deliveringQueues.put(queueId, queueId);
      setModified();
    }
  }

  /** Returns the identifiers of the delivering queues. */
  Iterator<AgentId> getDeliveringQueues() {
    return deliveringQueues.keySet().iterator();
  }

  /** Prepared transactions objects waiting for commit. */
  private Hashtable<Xid, XACnxPrepare> transactionsTable;

  /** <code>true</code> if the context is activated. */
  private transient boolean started;

  /** Sets the activation status of the context. */
  void setActivated(boolean started) {
    if (DEBUG)
      logger.log(BasicLevel.DEBUG,
                 "ClientContext[" + proxyId + ',' + id + "].setActivated(" + started + ')');
    this.started = started;
  }

  /** Returns <code>true</code> if the context is activated. */
  boolean isActivated() {
    return started;
  }
  
  /**
   * Identifier of a cancelled "receive" request, set when a PTP listener has
   * been unset.
   */
  private transient int cancelledRequestId;

  /** Cancels a "receive" request. */
  void cancelReceive(int cancelledRequestId) {
    if (DEBUG)
      logger.log(BasicLevel.DEBUG, 
                 "ClientContext[" + proxyId + ':' + id + "].cancelReceive(" + cancelledRequestId + ')');
    this.cancelledRequestId = cancelledRequestId;
  }

  /** Returns the cancelled "receive" request identifier. */
  int getCancelledReceive() {
    return cancelledRequestId;
  }

  /** Vector of active subscriptions' names. */
  private transient Vector<String> activeSubs;
  
  /** Adds an active subscription name. */
  void addSubName(String subName) {
    activeSubs.add(subName);
  }

  /** Returns the active subscriptions' names. */
  Iterator<String> getActiveSubs() {
    return activeSubs.iterator();
  }

  /** Removes an active subscription name. */
  void removeSubName(String subName) {
    activeSubs.remove(subName);
  }
  
  /** Pending replies waiting for the context to be activated. */
  private transient Vector<AbstractJmsReply> repliesBuffer;

  /** Adds a pending delivery. */
  void addPendingDelivery(AbstractJmsReply reply) {
    if (DEBUG)
      logger.log(BasicLevel.DEBUG,
                 "ClientContext[" + proxyId + ',' + id + "].addPendingDelivery(" + reply + ')');
    repliesBuffer.add(reply);
  }

  /** Returns the pending deliveries. */
  Iterator<AbstractJmsReply> getPendingDeliveries() {
    return repliesBuffer.iterator();
  }

  /** Clears the pending deliveries buffer. */
  void clearPendingDeliveries() {
    repliesBuffer.clear();
  }

  /** Contexts waiting for the replies from some local agents*/
  private transient Hashtable<Integer, MultiReplyContext> commitTable;  
  
  /**
   * Some requests may require to wait for several
   * SendReplyNot notifications before replying to the client.
   * 
   * @param requestId
   * @param asyncReplyCount
   */
  void addMultiReplyContext(int requestId, int asyncReplyCount) {
    if (DEBUG)
      logger.log(BasicLevel.DEBUG, 
                 "ClientContext[" + proxyId + ':' + id + "].addMultiReplyContext(" + requestId + ',' + asyncReplyCount + ')');
    if (commitTable == null)
      commitTable = new Hashtable<Integer, MultiReplyContext>();
    commitTable.put(Integer.valueOf(requestId), new MultiReplyContext(asyncReplyCount));
    setModified();
  }
  
  /** The proxy's agent identifier */
  private AgentId proxyId;
  
  /** An handle to proxy Agent */
  private transient ProxyAgentItf proxy;
  
  // TODO (AF): useless?
  public AgentId getProxyId() {
    return proxyId;
  }

  void setProxyAgent(ProxyAgentItf proxy) {
    this.proxy = proxy;
    this.proxyId = proxy.getId();
  }
  
  private transient String txName;

  // True if the context should be saved.
  private transient boolean modified;
  // True if the context has never been saved
  // This flag allows you to use a create operation during the first backup and to avoid a "need insert"
  // with BD persistence.
  private transient boolean first = true;
  
  private final void setModified() {
    if (! modified) {
      modified = true;
      proxy.modifiedClient(this);
    }
  }
  
  ClientContext() {}

  /**
   * Constructs a <code>ClientContext</code> instance.
   *
   * @param proxyId  The proxy's agent identifier. 
   * @param id  Identifier of the context.
   */
  ClientContext(AgentId proxyId, int id) {
    this.proxyId = proxyId;
    this.id = id;

    tempDestinations = new Vector<AgentId>();
    deliveringQueues = new Hashtable<AgentId, AgentId>();

    started = false;
    cancelledRequestId = -1;
    activeSubs = new Vector<String>();
    repliesBuffer = new Vector<AbstractJmsReply>();
    
    modified = true;
  }
  
  /**
   * Called by UserAgent when a SendReplyNot arrived.
   * 
   * @param requestId
   * @return &gt; 0 if there are still some pending replies, 0 if all the replies arrived (the context is removed)
   * or if the context doesn't exist
   */
  int setReply(int requestId) {
    if (DEBUG)
      logger.log(BasicLevel.DEBUG, 
                 "ClientContext[" + proxyId + ':' + id + "].setReply(" + requestId + ')');
    
    if (commitTable == null) return 0;
    
    Integer ctxKey = Integer.valueOf(requestId);
    MultiReplyContext ctx = (MultiReplyContext) commitTable.get(ctxKey);
    if (ctx == null) return 0;

    ctx.counter--;
    if (ctx.counter == 0) {
      commitTable.remove(ctxKey);
      setModified();
    }
    return ctx.counter;
  }
  
  static final class MultiReplyContext {
    int counter;
    
    MultiReplyContext(int c) {
      counter = c;
    }
  }

  /** Registers a given transaction "prepare". */
  void registerTxPrepare(Object key, XACnxPrepare prepare) throws Exception {
    if (transactionsTable == null)
      transactionsTable = new Hashtable<Xid, XACnxPrepare>();

    if (! transactionsTable.containsKey(key)) {
      transactionsTable.put((Xid) key, prepare);
      setModified();
    } else
      throw new Exception("Prepare request already received by "
                          + "TM for this transaction.");
  }

  /** Returns and deletes a given transaction "prepare". */
  XACnxPrepare getTxPrepare(Object key) {
    XACnxPrepare prepare = null;
    if (transactionsTable != null) {
      prepare = (XACnxPrepare) transactionsTable.remove(key);
      setModified();
    }
    return prepare;
  }

  /**
   * 
   * @param key XID
   * @return true if prepared transation.
   */
  public boolean isPrepared(Object key) {
    return transactionsTable.containsKey(key);
  }
  
  /** Returns the identifiers of the prepared transactions. */
  Iterator<Xid> getTxIds() {
    if (transactionsTable == null)
      return new Hashtable<Xid, Object>().keySet().iterator();
    return transactionsTable.keySet().iterator();
  }

  public String toString() {
    StringBuffer buff = new StringBuffer();
    buff.append("ClientContext (proxyId=");
    buff.append(proxyId);
    buff.append(",id=");
    buff.append(id);
    buff.append(",tempDestinations=");
    buff.append(tempDestinations);
    buff.append(",deliveringQueues=");
    buff.append(deliveringQueues);
    buff.append(",transactionsTable=");
    buff.append(transactionsTable);
    buff.append(",started=");
    buff.append(started);
    buff.append(",cancelledRequestId=");
    buff.append(cancelledRequestId);
    buff.append(",activeSubs=");
    buff.append(activeSubs);
    buff.append(",repliesBuffer=");
    buff.append(repliesBuffer);
    buff.append(')');
    return buff.toString();
  }

  public int getEncodableClassId() {
    return JoramHelper.CLIENT_CONTEXT_CLASS_ID;
  }

  public int getEncodedSize() throws Exception {
    int encodedSize = INT_ENCODED_SIZE;
    for (String activeSub : activeSubs) {
      encodedSize += EncodableHelper.getStringEncodedSize(activeSub);
    }
    encodedSize += INT_ENCODED_SIZE;
    Iterator<Entry<AgentId, AgentId>> deliveringQueueIterator = deliveringQueues.entrySet().iterator();
    while (deliveringQueueIterator.hasNext()) {
      Entry<AgentId, AgentId> deliveringQueue = deliveringQueueIterator.next();
      encodedSize += deliveringQueue.getKey().getEncodedSize();
    }
    encodedSize += INT_ENCODED_SIZE;
    encodedSize += INT_ENCODED_SIZE;
    
    for (AgentId tempDestination : tempDestinations) {
      encodedSize += tempDestination.getEncodedSize();
    }
    
    encodedSize += BOOLEAN_ENCODED_SIZE;
    if (transactionsTable != null) {
      encodedSize += INT_ENCODED_SIZE;
      Iterator<Entry<Xid, XACnxPrepare>> transactionsIterator = transactionsTable
          .entrySet().iterator();
      while (transactionsIterator.hasNext()) {
        Entry<Xid, XACnxPrepare> context = transactionsIterator.next();
        // Not useful to encode the key as it is in the value
        // context.getKey().encode(encoder);
        encodedSize += context.getValue().getEncodedSize();
      }
    }
    
    return encodedSize;
  }

  public void encode(Encoder encoder) throws Exception {
    encoder.encodeUnsignedInt(activeSubs.size());
    for (String activeSub : activeSubs) {
      encoder.encodeString(activeSub);
    }
    encoder.encodeUnsignedInt(deliveringQueues.size());
    Iterator<Entry<AgentId, AgentId>> deliveringQueueIterator = deliveringQueues.entrySet().iterator();
    while (deliveringQueueIterator.hasNext()) {
      Entry<AgentId, AgentId> deliveringQueue = deliveringQueueIterator.next();
      deliveringQueue.getKey().encode(encoder);
      // not useful to encode the value
    }
    encoder.encodeUnsignedInt(id);
    
    encoder.encodeUnsignedInt(tempDestinations.size());
    for (AgentId tempDestination : tempDestinations) {
      tempDestination.encode(encoder);
    }
    
    if (transactionsTable == null) {
      encoder.encodeBoolean(true);
    } else {
      encoder.encodeBoolean(false);
      encoder.encodeUnsignedInt(transactionsTable.size());
      Iterator<Entry<Xid, XACnxPrepare>> transactionsIterator = transactionsTable
          .entrySet().iterator();
      while (transactionsIterator.hasNext()) {
        Entry<Xid, XACnxPrepare> context = transactionsIterator.next();
        // Not useful to encode the key as it is in the value
        // context.getKey().encode(encoder);
        context.getValue().encode(encoder);
      }
    }

  }

  public void decode(Decoder decoder) throws Exception {
    int activeSubsSize = decoder.decodeUnsignedInt();
    activeSubs = new Vector<String>(activeSubsSize);
    for (int i = 0; i < activeSubsSize; i++) {
      String activeSub = decoder.decodeString();
      activeSubs.add(activeSub);
    }
    int deliveringQueuesSize = decoder.decodeUnsignedInt();
    deliveringQueues = new Hashtable<AgentId, AgentId>(deliveringQueuesSize);
    for (int i = 0; i < deliveringQueuesSize; i++) {
      AgentId key = new AgentId((short) 0, (short) 0, 0);
      key.decode(decoder);
      deliveringQueues.put(key, key);
    }
    id = decoder.decodeUnsignedInt();
    
    int tempDestinationsSize = decoder.decodeUnsignedInt();
    tempDestinations = new Vector<AgentId>(tempDestinationsSize);
    for (int i = 0; i < tempDestinationsSize; i++) {
      AgentId tempDestination = new AgentId((short) 0, (short) 0, 0);
      tempDestination.decode(decoder);
      tempDestinations.add(tempDestination);
    }
    
    boolean isNull = decoder.decodeBoolean();
    if (isNull) {
      transactionsTable = null;
    } else {
      int size = decoder.decodeUnsignedInt();
      transactionsTable = new Hashtable<Xid, XACnxPrepare>(size);
      for (int i = 0; i < size; i++) {
        XACnxPrepare ctx = new XACnxPrepare();
        ctx.decode(decoder);
        Xid xid = new Xid(ctx.getBQ(), ctx.getFI(), ctx.getGTI());
        transactionsTable.put(xid, ctx);
      }
    }
  }
  
  public static String getTransactionPrefix(AgentId proxyId) {
    StringBuffer clientContextPrefix = new StringBuffer(19).append("CC").append(proxyId.toString()).append('_');
    return clientContextPrefix.toString();
  }
  
  public final String getTxName() {
    if (txName == null) {
      txName = getTransactionPrefix(proxyId) + id;
    }
    return txName;
  }
  
  public static final ClientContext load(String txName) throws Exception {
    ClientContext cc = (ClientContext) AgentServer.getTransaction().load(txName);
    if (cc != null)
      cc.txName = txName;
    return cc;
  }
  
  public final void save() {
    if (!modified) return;
    try {
      if (first) {
        AgentServer.getTransaction().create(this, getTxName());
        first = false;
      } else {
        AgentServer.getTransaction().save(this, getTxName());
      }
      modified = false;
    } catch (IOException exc) {
      logger.log(BasicLevel.ERROR, "ClientContext named [" + getTxName() + "] could not be saved", exc);
    }
  }
  
  public void delete() {
    AgentServer.getTransaction().delete(getTxName());
  }
  
  List<String> getActiveSubList() {
    return activeSubs;
  }
  
  Map<AgentId, AgentId> getDeliveringQueueTable() {
    return deliveringQueues;
  }
  
  List<AgentId> getTempDestinationList() {
    return tempDestinations; 
  }
  
  Hashtable<Xid, XACnxPrepare> getTransactionsTable() {
    return transactionsTable;
  }

  public static final class ClientContextFactory implements EncodableFactory {
    public Encodable createEncodable() {
      return new ClientContext();
    }
  }
}

final class CCUID {
  @Override
  public int hashCode() {
    final int prime = 31;
    int result = 1;
    result = prime * result + ctxid;
    result = prime * result + ((ua == null) ? 0 : ua.hashCode());
    return result;
  }

  @Override
  public boolean equals(Object obj) {
    if (this == obj)
      return true;
    if (obj == null)
      return false;
    if (getClass() != obj.getClass())
      return false;
    CCUID other = (CCUID) obj;
    if (ctxid != other.ctxid)
      return false;
    if (ua == null) {
      if (other.ua != null)
        return false;
    } else if (!ua.equals(other.ua))
      return false;
    return true;
  }

  // TODO (AF): We could transform AgentId in long to optimize the comparison.
  AgentId ua = null;
  int ctxid = -1;

  CCUID(AgentId ua, int ctxid) {
    this.ua = ua;
    this.ctxid = ctxid;
  }
}
