/*
 * JORAM: Java(TM) Open Reliable Asynchronous Messaging
 * Copyright (C) 2016 - 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.tools.rest.jms;

import java.lang.reflect.Constructor;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Date;
import java.util.Enumeration;
import java.util.HashMap;
import java.util.Hashtable;
import java.util.Map;
import java.util.Properties;
import java.util.Random;
import java.util.concurrent.atomic.AtomicLong;

import jakarta.jms.ConnectionFactory;
import jakarta.jms.Destination;
import jakarta.jms.JMSContext;
import jakarta.jms.JMSProducer;
import jakarta.jms.MapMessage;
import jakarta.jms.Message;
import jakarta.jms.ResourceAllocationException;
import jakarta.jms.Topic;
import javax.naming.InitialContext;
import javax.naming.NamingException;
import javax.naming.RefAddr;
import javax.naming.Reference;
import javax.naming.StringRefAddr;

import org.ow2.joram.jakarta.jms.admin.AdministeredObject;
import org.objectweb.util.monolog.api.BasicLevel;
import org.objectweb.util.monolog.api.Logger;
import org.osgi.framework.BundleContext;

import fr.dyade.aaa.common.Debug;

public final class Helper implements HelperMBean {
  private static final Logger logger = Debug.getLogger(Helper.class.getName());
  
  private static final String BYTES_CLASS_NAME = byte[].class.getName();
  public static final String BUNDLE_CF_PROP = "rest.jms.connectionFactory";
  public static final String BUNDLE_JNDI_FACTORY_INITIAL_PROP = "rest.jndi.factory.initial";
  public static final String BUNDLE_JNDI_FACTORY_HOST_PROP = "rest.jndi.factory.host";
  public static final String BUNDLE_JNDI_FACTORY_PORT_PROP = "rest.jndi.factory.port";
  public static final String BUNDLE_IDLE_TIMEOUT_PROP = "rest.idle.timeout";
  public static final String BUNDLE_CLEANER_PERIOD_PROP = "rest.cleaner.period";
  
  public static final String BUNDLE_JMS_USER = "rest.jms.user";
  public static final String BUNDLE_JMS_PASS = "rest.jms.password";
  public static final String BUNDLE_JMS_IP_ALLOWED = "rest.jms.ipallowed";

  public static final String BUNDLE_JNDI_FACTORY_TIMEOUT_PROP = "rest.jndi.socketTimeout";
  public static final String BUNDLE_JNDI_FACTORY_LINGER_PROP = "rest.jndi.socketLinger";
  public static final String BUNDLE_JNDI_FACTORY_REUSE_ADDRESS_PROP = "rest.jndi.socketReuseAddress";
  
  public static final String BUNDLE_JNDI_CACHE_UPDATE_PERIOD_PROPERTY = "rest.jndi.cacheUpdatePeriod";
  public static final int JNDI_CACHE_UPDATE_PERIOD_DEFAULT = 0;
  
  public static final String BUNDLE_MAX_NUMBER_OF_CONTEXT_PROPERTY = "rest.jms.maxNumberOfContext";
  public static final int MAX_NUMBER_OF_CONTEXT_DEFAULT = -1;
  public static int maxNumberOfContext = MAX_NUMBER_OF_CONTEXT_DEFAULT;
  
  public static final String BUNDLE_CLIENTID_PREFIX_PROPERTY = "rest.jms.clientIdPrefix";
  public static final String CLIENTID_PREFIX_DEFAULT = "clientID";
  public static String clientIdPrefix = CLIENTID_PREFIX_DEFAULT;
  
  /**
   * JNDI cache update period in ms. 
   * If 0, the cache is not use, if it is less than 0 it is never updated.
   */
  private static int jndiUpdatePeriod = JNDI_CACHE_UPDATE_PERIOD_DEFAULT;

  public static final int DFLT_CLEANER_PERIOD = 15;
  
  /**
   * This random generator is used to built random identifiers of producer and consumer.
   * May be we should use SecureRandom.
   */
  private Random rand = new Random();
  private static final AtomicLong counter = new AtomicLong(1);
  // Singleton
  private static Helper helper = null;
  
  private InitialContext ictx;
  
  // Be careful, this map is also used in CleanerTask!! 
  private Map<String, RestClientContext> restClientCtxs;
  
  /**
   * Only used by CleanerTask to remove unused contexts.
   * 
   * @return the restClientCtxs
   */
  Map<String, RestClientContext> getRestClientCtxs() {
    return restClientCtxs;
  }

  private Map<String, SessionContext> sessionCtxs;
  private String cfName;
  private long globalIdleTimeout = 0;
  private Properties jndiProps;
  
  private String restUser;
  private String restPass;
  
  private String IPAllowed;
  private IPFilter ipfilter;

  private long lastCleanTime = 0L;
  
  private Helper() {
    restClientCtxs = Collections.synchronizedMap(new HashMap<String, RestClientContext>());
    sessionCtxs = Collections.synchronizedMap(new HashMap<String, SessionContext>());
  }
  
  static public synchronized Helper getInstance() {
    if (helper == null)
      helper = new Helper();
    return helper;
  }
  
  /**
   * @return the Rest/Jms user
   */
  public String getRestUser() {
    return restUser;
  }

  /**
   * @return the Rest/Jms password
   */
  public String getRestPass() {
    return restPass;
  }

  public void setGlobalProperties(BundleContext bundleContext) throws NamingException {
    restUser = bundleContext.getProperty(BUNDLE_JMS_USER);
    restPass = bundleContext.getProperty(BUNDLE_JMS_PASS);
    
    IPAllowed = bundleContext.getProperty(BUNDLE_JMS_IP_ALLOWED);
    if (logger.isLoggable(BasicLevel.INFO))
      logger.log(BasicLevel.INFO, "Set configuration property, IPFilter allowedList=" + IPAllowed);
    ipfilter = new IPFilter(IPAllowed);
    
    // set the connection factory
    setConnectionFactoryName(bundleContext.getProperty(BUNDLE_CF_PROP));
    
    // set the jndi properties
    jndiProps = new Properties();
    jndiProps.setProperty("java.naming.factory.initial", "fr.dyade.aaa.jndi2.client.NamingContextFactory");
    if (bundleContext.getProperty(BUNDLE_JNDI_FACTORY_INITIAL_PROP) != null)
      jndiProps.setProperty("java.naming.factory.initial", bundleContext.getProperty(BUNDLE_JNDI_FACTORY_INITIAL_PROP));
    jndiProps.setProperty("java.naming.factory.host", "localhost");
    if (bundleContext.getProperty(BUNDLE_JNDI_FACTORY_HOST_PROP) != null)
      jndiProps.setProperty("java.naming.factory.host", bundleContext.getProperty(BUNDLE_JNDI_FACTORY_HOST_PROP));
    jndiProps.setProperty("java.naming.factory.port", "16400");
    if (bundleContext.getProperty(BUNDLE_JNDI_FACTORY_PORT_PROP) != null)
      jndiProps.setProperty("java.naming.factory.port", bundleContext.getProperty(BUNDLE_JNDI_FACTORY_PORT_PROP));

    if (bundleContext.getProperty(BUNDLE_JNDI_FACTORY_TIMEOUT_PROP) != null)
      jndiProps.setProperty("fr.dyade.aaa.jndi2.socketTimeOut", bundleContext.getProperty(BUNDLE_JNDI_FACTORY_TIMEOUT_PROP));
    if (bundleContext.getProperty(BUNDLE_JNDI_FACTORY_LINGER_PROP) != null)
      jndiProps.setProperty("fr.dyade.aaa.jndi2.socketLinger", bundleContext.getProperty(BUNDLE_JNDI_FACTORY_LINGER_PROP));
    if (bundleContext.getProperty(BUNDLE_JNDI_FACTORY_REUSE_ADDRESS_PROP) != null)
      jndiProps.setProperty("fr.dyade.aaa.jndi2.socketReuseAddress", bundleContext.getProperty(BUNDLE_JNDI_FACTORY_REUSE_ADDRESS_PROP));

    if (bundleContext.getProperty(BUNDLE_JNDI_CACHE_UPDATE_PERIOD_PROPERTY) != null)
      jndiUpdatePeriod = Integer.valueOf(bundleContext.getProperty(BUNDLE_JNDI_CACHE_UPDATE_PERIOD_PROPERTY));
    
    // TODO: use the osgi service jndi?
    //    ServiceReference<ObjectFactory> ref = context.getServiceReference(javax.naming.spi.ObjectFactory.class);
    //    ObjectFactory jndiFactory = bundleContext.getService(ref);
    
    if (logger.isLoggable(BasicLevel.INFO))
      logger.log(BasicLevel.INFO, "jndiProperties = " + jndiProps);
    
    // set default idle timeout
    String value = bundleContext.getProperty(BUNDLE_IDLE_TIMEOUT_PROP);
    if (value != null && !value.isEmpty()) {
      try {
        globalIdleTimeout = Long.parseLong(value);
        logger.log(BasicLevel.INFO,
                   "Set configuration property, " + BUNDLE_IDLE_TIMEOUT_PROP + '=' + value);
      } catch (NumberFormatException exc) {
          logger.log(BasicLevel.WARN,
              "Bad configuration property " + BUNDLE_IDLE_TIMEOUT_PROP + ", should be a number: " + value);
      }
    }
    
    value = bundleContext.getProperty(BUNDLE_MAX_NUMBER_OF_CONTEXT_PROPERTY);
    if (value != null && !value.isEmpty()) {
      try {
        maxNumberOfContext = Integer.parseInt(value);
        logger.log(BasicLevel.INFO,
                   "Set configuration property, " + BUNDLE_MAX_NUMBER_OF_CONTEXT_PROPERTY + '=' + value);
      } catch (NumberFormatException exc) {
        logger.log(BasicLevel.WARN,
                   "Bad configuration property " + BUNDLE_MAX_NUMBER_OF_CONTEXT_PROPERTY + ", should be a number: " + value);
      }
    }
    
    value = bundleContext.getProperty(BUNDLE_CLIENTID_PREFIX_PROPERTY);
    if (value != null && !value.isEmpty()) {
      clientIdPrefix = value;
      logger.log(BasicLevel.INFO,
                 "Set configuration property, " + BUNDLE_CLIENTID_PREFIX_PROPERTY + '=' + value);
    }
  }
  
  /**
   * @return the IPAllowed
   */
  public String getIPAllowed() {
    return IPAllowed;
  }

  /**
   * Check if the addr is authorized (all local address is authorized).
   * 
   * @param addr The ip address to check
   * @return true if authorized
   * @throws UnknownHostException
   * @throws SocketException
   */
  public boolean checkIPAddress(String addr) {
    return ipfilter.checkIpAllowed(addr);
  }
  
  public boolean authenticationRequired() {
    return restUser != null && !restUser.isEmpty() &&
        restPass != null && !restPass.isEmpty();
  }

  /**
   * @param cfName the connection factory name
   */
  public void setConnectionFactoryName(String cfName) {
    if (cfName != null) {
      this.cfName = cfName;
      if (logger.isLoggable(BasicLevel.DEBUG))
        logger.log(BasicLevel.DEBUG, "Helper.setConnectionFactoryName = " + this.cfName);
    } else {
      this.cfName = "cf";
    }
  }

  private long cleanerPeriod;
  
  public void setCleanerPeriod(long cleanerPeriod) {
    this.cleanerPeriod = cleanerPeriod;
  }
  
  public long getCleanerPeriod() {
    return cleanerPeriod;
  }
  public long getDefaultIdleTimeout() {
    return globalIdleTimeout;
  }
  
  public void setLastCleanTime() {
    lastCleanTime = System.currentTimeMillis();
  }
  
  public String getLastCleanTime() {
    return new Date(lastCleanTime).toString();
  }

  public int getNbContexts() {
    return restClientCtxs.size();
  }
  
  public String dumpContexts() {
    StringBuilder builder = new StringBuilder();
    ArrayList<RestClientContext> values = new ArrayList<RestClientContext>(restClientCtxs.values());
    for (RestClientContext ctx : values) {
      builder.append(ctx).append('\n');
    }
    return builder.toString();
  }
  
  public String dumpSessions() {
    StringBuilder builder = new StringBuilder();
    ArrayList<SessionContext> values = new ArrayList<SessionContext>(sessionCtxs.values());
    for (SessionContext ctx : values) {
      builder.append(ctx).append('\n');
    }
    return builder.toString();
  }
  
  /**
   * Should only be called in stop.
   */
  public void closeAll()  {
    ArrayList<RestClientContext> values = new ArrayList<RestClientContext>(restClientCtxs.values());
    if (logger.isLoggable(BasicLevel.DEBUG))
      logger.log(BasicLevel.DEBUG, "Helper.closeAll " + values);
    for (RestClientContext restClientCtx : values) {
      close(restClientCtx.getClientId());
    }
    // close jndi
    if (ictx != null)
      try {
        ictx.close();
      } catch (NamingException exc) {
        logger.log(BasicLevel.DEBUG, "Cannot close JNDI context: " + exc.getMessage());
      }
  }
  
  /**
   * @param clientId the client id
   */
  public void close(String clientId) {
    if (logger.isLoggable(BasicLevel.DEBUG))
      logger.log(BasicLevel.DEBUG, "Helper.close " + clientId);
    RestClientContext restClientCtx = restClientCtxs.get(clientId);
    if (restClientCtx != null) {
      ArrayList<String> names = new ArrayList<String>(restClientCtx.getSessionCtxNames());
      for (String name : names) {
        closeSessionCtx(name);
      }
      restClientCtxs.remove(clientId);
    }
  }

  /**
   * @param ctxName the producer or consumer name
   */
  public void closeSessionCtx(String ctxName) {
    if (logger.isLoggable(BasicLevel.DEBUG))
      logger.log(BasicLevel.DEBUG, "Helper.closeSessionCtx " + ctxName);
    SessionContext ctx = sessionCtxs.get(ctxName);
    if (ctx != null) {
      RestClientContext restClientCtx = ctx.getClientCtx();
      ctx.getJmsContext().close();
      sessionCtxs.remove(ctxName);
      restClientCtx.removeSessionCtxNames(ctxName);
      if (restClientCtx.getSessionCtxNames().isEmpty()) {
        restClientCtx.getJmsContext().close();
        restClientCtxs.remove(restClientCtx.getClientId());
      }
    }
  }
  
  // Do not use directly, always use access methods
  private long cfhtLastUpdate = 0L;
  private Hashtable<String, ConnectionFactory> cfht = new Hashtable<>();

  /**
   * Returns the value in cache, cleans the cache if needed.
   * @param cfName
   * @return
   */
  private final ConnectionFactory getJNDIcf(String cfName) {
    if (jndiUpdatePeriod == 0) return null;
    
    if (jndiUpdatePeriod > 0) {
      // Cleans the cache if the update period has passed.
      // TODO: handles each entry individually.
      long now = System.currentTimeMillis();
      if ((now - cfhtLastUpdate) > jndiUpdatePeriod) {
        if (logger.isLoggable(BasicLevel.DEBUG))
          logger.log(BasicLevel.DEBUG, "Cleans JNDI cache for ConnectionFactory.");
        
        cfht.clear();
        cfhtLastUpdate = now;
        return null;
      }
    }
    return cfht.get(cfName);
  }
  
  /**
   * Lookup the ConnectionFactory
   * 
   * @param cfName the ConnectionFactory name
   * @return the ConnectionFactory if found
   * @throws NamingException
   */
  public final ConnectionFactory lookupConnectionFactory(String cfName) throws NamingException {
    if (logger.isLoggable(BasicLevel.DEBUG))
      logger.log(BasicLevel.DEBUG, "Helper.lookupConnectionFactory " + cfName);
    
    ConnectionFactory cf = getJNDIcf(cfName);
    if (cf != null) return cf;
    
    Object obj = lookup(cfName);
    if (obj instanceof ConnectionFactory) {
      cf = (ConnectionFactory) obj;
      cfht.put(cfName, cf);
      return cf;
    }
    if (obj != null)
      logger.log(BasicLevel.WARN,
                 "Helper.lookupConnectionFactory, " + cfName + " does not correspond to a ConnectionFactory: " + obj.getClass());
    return null;
  }
  
  // Do not use directly, always use access methods
  private long desthtLastUpdate = 0L;
  private Hashtable<String, Destination> destht = new Hashtable<>();

  /**
   * Returns the value in cache, cleans the cache if needed.
   * @param destName
   * @return
   */
  private final Destination getJNDIdest(String destName) {
    if (jndiUpdatePeriod == 0) return null;
    
    if (jndiUpdatePeriod > 0) {
      // Cleans the cache if the update period has passed.
      // TODO: handles each entry individually.
      long now = System.currentTimeMillis();
      if ((now - desthtLastUpdate) > jndiUpdatePeriod) {
        if (logger.isLoggable(BasicLevel.DEBUG))
          logger.log(BasicLevel.DEBUG, "Cleans JNDI cache for Destination.");
        
        destht.clear();
        desthtLastUpdate = now;
        return null;
      }
    }
    
    return destht.get(destName);
  }
  
  /**
   * Lookup the destination
   * 
   * @param destName the destination name
   * @return the destination if found
   * @throws NamingException
   */
  public final Destination lookupDestination(String destName) throws NamingException {
    if (logger.isLoggable(BasicLevel.DEBUG))
      logger.log(BasicLevel.DEBUG, "Helper.lookupDestination " + destName);

    Destination dest = getJNDIdest(destName);
    if (dest != null) return dest;

    Object obj = lookup(destName);
    if (obj instanceof Destination) {
      dest = (Destination) obj;
      destht.put(destName, dest);
      return dest;
    }
    if (obj != null)
      logger.log(BasicLevel.WARN,
                 "Helper.lookupDestination, " + destName + " does not correspond to a destination: " + obj.getClass());
    return null;
  }
  
  /**
   * Retrieves ConnectionFactory and Destination objects from JNDI.
   * This method converts 'old' javax.jms references to the new Jakarta JMS API.
   * 
   * TODO (AF): We should replace this code by a call to Helper.retrieveJakartaJMSObjectFromJNDI method (org.ow2.joram.jndi package).
   * 
   * @param jndiName
   * @return
   * @throws Exception
   */
  private synchronized Object lookup(String name) {
    try {
      if (logger.isLoggable(BasicLevel.INFO))
        logger.log(BasicLevel.INFO, "Helper.lookup: " + name);
      
      ClassLoader originalContextClassLoader = Thread.currentThread().getContextClassLoader();
      try {
        Thread.currentThread().setContextClassLoader(getClass().getClassLoader());

        if (ictx == null)
          ictx = new InitialContext(jndiProps);
      } finally {
        Thread.currentThread().setContextClassLoader(originalContextClassLoader);
      }

      // Note (AF): Lookup should be done during the classloader change.
      // However it seems that Jakarta/JMS object are not found anyway.
      // Should be tested carefully
      Object obj = ictx.lookup(name);
      if (logger.isLoggable(BasicLevel.INFO))
        logger.log(BasicLevel.INFO, "Helper.lookup(" + name + ") gets " + obj + "\n => " + obj.getClass().getName());

      if (obj instanceof org.objectweb.joram.client.jms.ConnectionFactory) {
        org.objectweb.joram.client.jms.ConnectionFactory cf1 = (org.objectweb.joram.client.jms.ConnectionFactory) obj;
        javax.naming.Reference ref1 = cf1.getReference();

        javax.naming.Reference ref2 = new Reference(ref1.getClassName().replace("objectweb.joram.client", "ow2.joram.jakarta"), 
                                                    org.objectweb.joram.client.jms.admin.ObjectFactory.class.getName(), null);

        Enumeration<RefAddr> refAddrs = ref1.getAll();
        while (refAddrs.hasMoreElements()) {
          RefAddr ra = refAddrs.nextElement();
          if ("cf.reliableClass".equals(ra.getType())) {
            if (ra.getContent() != null)
              ref2.add(new StringRefAddr(ra.getType(), ((String) ra.getContent()).replace("objectweb.joram.client", "ow2.joram.jakarta")));
            else
              ref2.add(new StringRefAddr(ra.getType(), null));
          } else {
            ref2.add(ra);
          }
        }

        String cn = ref1.getClassName().replace("objectweb.joram.client", "ow2.joram.jakarta");
        org.ow2.joram.jakarta.jms.ConnectionFactory cf2 = null;
        try {
          cf2 = (org.ow2.joram.jakarta.jms.ConnectionFactory) Class.forName(cn).newInstance();
        } catch (Exception exc) {
          logger.log(BasicLevel.WARN, "Helper.lookup(" + name + ") Can't transform to Javax/JMS: " + obj);
          return obj;
        }
        cf2.fromReference(ref2);

        if (logger.isLoggable(BasicLevel.INFO))
          logger.log(BasicLevel.INFO, "Helper.lookup(" + name + ") returns " + cf2);

        return cf2;
      } else if (obj instanceof org.objectweb.joram.client.jms.Destination) {
        org.objectweb.joram.client.jms.Destination dest1 = (org.objectweb.joram.client.jms.Destination) obj;
        javax.naming.Reference ref = dest1.getReference();
        String cn = ref.getClassName().replace("objectweb.joram.client", "ow2.joram.jakarta");
        org.ow2.joram.jakarta.jms.Destination dest2 = null;
        try {
          dest2 = (org.ow2.joram.jakarta.jms.Destination) Class.forName(cn).newInstance();
        } catch (Exception exc) {
          logger.log(BasicLevel.WARN, "Helper.lookup(" + name + ") Can't transform to Javax/JMS: " + obj);
          return obj;
        }
        dest2.fromReference(ref);

        if (logger.isLoggable(BasicLevel.INFO))
          logger.log(BasicLevel.INFO, "Helper.lookup(" + name + ") returns " + dest2);

        return dest2;
      } else if (obj instanceof Reference) {
        Reference ref1 = (Reference) obj;

        javax.naming.Reference ref2 = new Reference(ref1.getClassName().replace("objectweb.joram.client", "ow2.joram.jakarta"), 
                                                    org.objectweb.joram.client.jms.admin.ObjectFactory.class.getName(), null);

        Enumeration<RefAddr> refAddrs = ref1.getAll();
        while (refAddrs.hasMoreElements()) {
          RefAddr ra = refAddrs.nextElement();
          if ("cf.reliableClass".equals(ra.getType())) {
            if (ra.getContent() != null)
              ref2.add(new StringRefAddr(ra.getType(), ((String) ra.getContent()).replace("objectweb.joram.client", "ow2.joram.jakarta")));
            else
              ref2.add(new StringRefAddr(ra.getType(), null));
          } else {
            ref2.add(ra);
          }
        }

        AdministeredObject ao = null;
        try {
          Class<?> clazz = Class.forName(ref2.getClassName());
          ao = (AdministeredObject) clazz.newInstance();
          ao.fromReference(ref2);
        } catch (Exception exc) {
          if (logger.isLoggable(BasicLevel.DEBUG))
            logger.log(BasicLevel.ERROR, "Cannot handle Administered object:", exc);
          else
            logger.log(BasicLevel.ERROR, "Cannot handle Administered object: " + exc.getMessage());
        }
        return ao;
      }

      return obj;
    } catch (Throwable t) {
      if (logger.isLoggable(BasicLevel.DEBUG))
        logger.log(BasicLevel.WARN, "Helper.lookup " + name, t);
      else
        logger.log(BasicLevel.WARN, "Helper.lookup " + name + ": " + t.getMessage());
      return null;
    }
  }
  
  /**
   * Create a producer
   * 
   * @param userName the user name
   * @param password the password
   * @param clientId the client id
   * @param prodName the producer name
   * @param dest the JMS destination
   * @param sessionMode the session mode
   * @param deliveryMode the delivery mode
   * @param deliveryDelay the delivery delay
   * @param correlationID the correlation id
   * @param priority the priority
   * @param timeToLive the time to live
   * @param destName need for the jms create
   * @param isQueue
   * @param idleTimeout 
   * @return the producer name
   * @throws Exception
   */
  public synchronized String createProducer(
      String userName,
      String password,
      String clientId,
      String prodName,
      Destination dest, 
      int sessionMode, 
      int deliveryMode,
      long deliveryDelay,
      String correlationID, 
      int priority, 
      long timeToLive,
      String destName,
      boolean isQueue, 
      long idleTimeout) throws Exception {
    
    String prodId = prodName;
    if (prodId == null) {
      // create the new producer Id
      prodId = createProducerId();
      while (sessionCtxs != null && sessionCtxs.get(prodId) != null) {
        prodId = createProducerId();
      }
      if (logger.isLoggable(BasicLevel.DEBUG))
        logger.log(BasicLevel.DEBUG, "Helper.createProducer: create the new producer Id " + prodId);
    }
    
    // Get the rest client context
    RestClientContext restClientCtx = getClientContext(clientId);
    if (restClientCtx == null)
      throw new ResourceAllocationException("Too many JMS connections.");
    if (restClientCtx.getJmsContext() == null) {
      if (restClientCtx.getConnectionFactory() == null) {
        ConnectionFactory cf = lookupConnectionFactory(cfName);
        if (cf == null)
          throw new NamingException("Cannot get ConnectionFactory");
        restClientCtx.setConnectionFactory(cf);
        
        if (idleTimeout != 0)
          restClientCtx.setIdleTimeout(idleTimeout);
        else if (globalIdleTimeout > 0)
          restClientCtx.setIdleTimeout(globalIdleTimeout);
      }
      if (logger.isLoggable(BasicLevel.DEBUG))
        logger.log(BasicLevel.DEBUG, "Helper.createProducer cf = " + restClientCtx.getConnectionFactory());
      JMSContext jmsContext;
      if (userName != null && !userName.isEmpty())
        jmsContext = restClientCtx.getConnectionFactory().createContext(userName, password);
      else
        jmsContext = restClientCtx.getConnectionFactory().createContext();
      jmsContext.setClientID(restClientCtx.getClientId());
      jmsContext.setAutoStart(false);//the default is true.
      restClientCtx.setJmsContext(jmsContext);
      if (logger.isLoggable(BasicLevel.DEBUG))
        logger.log(BasicLevel.DEBUG, "Helper.createProducer jmsContext = " + restClientCtx.getJmsContext());
    }
    
    ProducerContext prodContext = (ProducerContext) sessionCtxs.get(prodId);
    if (prodContext == null) {
      // create a new producer context
      prodContext = new ProducerContext(restClientCtx);
      prodContext.setJmsContext(restClientCtx.getJmsContext().createContext(sessionMode));
      
      prodContext.setDefaultDeliveryMode(deliveryMode);
      prodContext.setDefaultJMSCorrelationID(correlationID);
      prodContext.setDefaultPriority(priority);
      prodContext.setDefaultTimeToLive(timeToLive);
      prodContext.setDefaultDeliveryDelay(deliveryDelay);
      
      JMSProducer  producer = prodContext.getJmsContext().createProducer();      
      ((ProducerContext) prodContext).setProducer(producer);
      sessionCtxs.put(prodId, prodContext);
      restClientCtx.addSessionCtxNames(prodId);

      // Note: In JndiService the destination comes from JNDI.
      Destination destination = dest;
      // set the destination
      if (destination == null) {
        // create the jms destination (Needed in JmsService).
        if (isQueue)
          destination = restClientCtx.getJmsContext().createQueue(destName);
        else
          destination = restClientCtx.getJmsContext().createTopic(destName);
      }
      prodContext.setDest(destination);
    }
    
    return prodId;
  }
  
  /**
   * Create a consumer
   * 
   * @param userName the user name
   * @param password the password
   * @param clientId the client id
   * @param consName the consumer name
   * @param dest the JMS destination
   * @param sessionMode the session mode
   * @param messageSelector
   * @param noLocal
   * @param durable
   * @param shared
   * @param name
   * @param destName need for the jms create
   * @param isQueue
   * @param idleTimeout 
   * @return the consumer name
   * @throws Exception
   */
  public synchronized String createConsumer(
      String userName,
      String password,
      String clientId,
      String consName,
      Destination dest, 
      int sessionMode,
      String messageSelector,
      boolean noLocal,
      boolean durable,
      boolean shared,
      String name,
      String destName,
      boolean isQueue, 
      long idleTimeout) throws Exception {
    
    String consId = consName;
    if (consId == null) {
      // create the new consumer Id
      consId = createConsumerId();
      while (sessionCtxs != null && sessionCtxs.get(consId) != null) {
        consId = createConsumerId();
      }
      if (logger.isLoggable(BasicLevel.DEBUG))
        logger.log(BasicLevel.DEBUG, "Helper.createConsumer: create the new consumer Id " + consId);
    }
    
    // Get the rest client context
    RestClientContext restClientCtx = getClientContext(clientId);
    if (restClientCtx == null)
      throw new ResourceAllocationException("Too many JMS connections.");
    if (restClientCtx.getJmsContext() == null) {
      if (restClientCtx.getConnectionFactory() == null) {
        ConnectionFactory cf = lookupConnectionFactory(cfName);
        if (cf == null)
          throw new NamingException("Cannot get ConnectionFactory");
        restClientCtx.setConnectionFactory(cf);
        
        if (idleTimeout != 0)
          restClientCtx.setIdleTimeout(idleTimeout);
        else if (globalIdleTimeout > 0)
          restClientCtx.setIdleTimeout(globalIdleTimeout);
      }
      if (logger.isLoggable(BasicLevel.DEBUG))
        logger.log(BasicLevel.DEBUG, "Helper.createConsumer cf = " + restClientCtx.getConnectionFactory());
      JMSContext jmsContext;
      if (userName != null && !userName.isEmpty())
        jmsContext = restClientCtx.getConnectionFactory().createContext(userName, password);
      else
        jmsContext = restClientCtx.getConnectionFactory().createContext();
      restClientCtx.setJmsContext(jmsContext);
      restClientCtx.getJmsContext().setClientID(restClientCtx.getClientId());
      restClientCtx.getJmsContext().setAutoStart(false);//the defaul is true.
      if (logger.isLoggable(BasicLevel.DEBUG))
        logger.log(BasicLevel.DEBUG, "Helper.createConsumer jmsContext = " + restClientCtx.getJmsContext());
    }

    SessionContext consumerContext = sessionCtxs.get(consId);
    if (consumerContext == null) {
      // create the consumer context
      consumerContext = new ConsumerContext(restClientCtx);

      // Note: In JndiService the destination comes from JNDI.
      Destination destination = dest;
      // set the destination
      if (destination == null) {
        // create the jms destination (Needed in JmsService).
        if (isQueue)
          destination = restClientCtx.getJmsContext().createQueue(destName);
        else
          destination = restClientCtx.getJmsContext().createTopic(destName);
      }
      consumerContext.setDest(destination);

      consumerContext.setJmsContext(restClientCtx.getJmsContext().createContext(sessionMode));
      //TODO : test if dest == TOPIC for durable and shared
      if (durable && !shared) {
        if (messageSelector == null) {
          ((ConsumerContext) consumerContext).setConsumer(consumerContext.getJmsContext().createDurableConsumer((Topic) consumerContext.getDest(), name));
        } else {
          ((ConsumerContext) consumerContext).setConsumer(consumerContext.getJmsContext().createDurableConsumer((Topic) consumerContext.getDest(), name, messageSelector, noLocal));
        }
      } else if (durable && shared) {
        if (messageSelector == null) {
          ((ConsumerContext) consumerContext).setConsumer(consumerContext.getJmsContext().createSharedDurableConsumer((Topic) consumerContext.getDest(), name));
        } else {
          ((ConsumerContext) consumerContext).setConsumer(consumerContext.getJmsContext().createSharedDurableConsumer((Topic) consumerContext.getDest(), name, messageSelector));
        }
      } else if (shared) {
        if (messageSelector == null) {
          ((ConsumerContext) consumerContext).setConsumer(consumerContext.getJmsContext().createSharedConsumer((Topic) consumerContext.getDest(), name));
        } else {
          ((ConsumerContext) consumerContext).setConsumer(consumerContext.getJmsContext().createSharedConsumer((Topic) consumerContext.getDest(), name, messageSelector));
        }
      } else {
        if (messageSelector == null) {
          ((ConsumerContext) consumerContext).setConsumer(consumerContext.getJmsContext().createConsumer(consumerContext.getDest()));
        } else {
          ((ConsumerContext) consumerContext).setConsumer(consumerContext.getJmsContext().createConsumer(consumerContext.getDest(), messageSelector, noLocal));
        } 
      }

      // put the consumer context in consumers map
      sessionCtxs.put(consId, consumerContext);
      restClientCtx.addSessionCtxNames(consId);

      // start the connection if autostart != true
      if (!consumerContext.getJmsContext().getAutoStart())
        consumerContext.getJmsContext().start();
    }
    
    return consId;
  }

  static final void setMapMessage(Map<String, Object> jsonMap, MapMessage msg) throws Exception {
    if (jsonMap == null)
      return;

    // parse the json map
    for (String key : jsonMap.keySet()) {
      Object value = jsonMap.get(key);
      if (value instanceof ArrayList) {
        ArrayList<String> array = (ArrayList<String>) value; 
        try {
          if (array.size() == 2) {
            String className = array.get(1);
            if (Character.class.getName().equals(className)) {
              value =  array.get(0).charAt(0);
            } else if (BYTES_CLASS_NAME.equals(className)) {
              value = array.get(0).getBytes("UTF-8"); 
            } else {
              Constructor<?> constructor = Class.forName(className).getConstructor(String.class);
              value = constructor.newInstance(array.get(0));
            }
          }
        } catch (Exception e) {
          logger.log(BasicLevel.ERROR, "setMapMessage: ignore map entry " + key + ", " + value + " : " + e.getMessage());
          continue;
        }
      }
      if (logger.isLoggable(BasicLevel.DEBUG))
        logger.log(BasicLevel.DEBUG, "setMapMessage: " + key + ", value = " + value + ", " + value.getClass().getSimpleName());

      switch (value.getClass().getSimpleName()) {
      case "String":
        msg.setString(key, (String) value);
        break;
      case "Boolean":
        msg.setBoolean(key, (Boolean)value);
        break;
      case "Integer":
        msg.setInt(key, (Integer)value);
        break;
      case "Long":
        msg.setLong(key, (Long)value);
        break;
      case "Double":
        msg.setDouble(key, (Double)value);
        break;
      case "Float":
        msg.setFloat(key, (Float)value);
        break;
      case "Short":
        msg.setShort(key, (Short)value);
        break;
      case "Character":
        msg.setChar(key, (char)value);
        break;
      case "Byte":
        msg.setByte(key, (Byte)value);
        break;
      case "byte[]":
        msg.setBytes(key, (byte[])value);
        break;

      default:
        try {
          msg.setObjectProperty(key, value);
        } catch (Exception e) {
          logger.log(BasicLevel.ERROR, "ignore jms setObjectProperties(" + key + ", " + value + ") : " + e.getMessage());
        }
        break;
      }
    }
  }
 
  static final Object getValue(Map map, String key) throws Exception {
    Object value = map.get(key);
    if (value instanceof ArrayList) {
      ArrayList<String> array = (ArrayList<String>) value; 
      try {
        if (array.size() == 2) {
          String className = array.get(1);
          Constructor<?> constructor = Class.forName(className).getConstructor(String.class);
          value = constructor.newInstance(array.get(0));
        }
      } catch (Exception e) {
        logger.log(BasicLevel.ERROR, "getValue(" + key + ", " + value + "): " + e.getMessage());
        throw e;
      }
    }
    return value;
  }
  
  /**
   * @param prodName
   * @param type
   * @param jmsHeaders
   * @param jmsProps
   * @param jmsBody
   * @param deliveryMode
   * @param deliveryDelay
   * @param priority
   * @param timeToLive
   * @param correlationID
   * @return
   * @throws Exception
   */
  public long send(
      String prodName,
      String type, 
      Map<String, Object> jmsHeaders, 
      Map<String, Object> jmsProps, 
      Object jmsBody, 
      int deliveryMode, 
      long deliveryDelay, 
      int priority, 
      long timeToLive, 
      String correlationID) throws Exception {

    try {
      ProducerContext producerCtx = (ProducerContext) sessionCtxs.get(prodName);
      if (producerCtx == null)
        throw new Exception(prodName + " not found.");
      
      return producerCtx.send(type, 
          jmsHeaders, jmsProps, 
          jmsBody, deliveryMode, deliveryDelay, priority, timeToLive,
          correlationID);
    } catch (Exception e) {
      if (logger.isLoggable(BasicLevel.DEBUG))
        logger.log(BasicLevel.WARN, "Cannot send message:", e);
      else
        logger.log(BasicLevel.WARN, "Cannot send message: " + e.getMessage());
      throw e;
    }
  }
  
  public Message consume(
      String consName,
      long timeout,
      boolean noLocal,
      boolean durable,
      boolean shared,
      String name,
      long msgId) throws Exception {
    
    ConsumerContext consumerCtx = (ConsumerContext) sessionCtxs.get(consName);
    if (consumerCtx == null)
      throw new Exception(consName + " not found.");
    
    return consumerCtx.receive(timeout, msgId);
  }

  public String createClientId() {
    return clientIdPrefix + counter.getAndIncrement();
  }
  
  public String createProducerId() {
    return "prod_" + (rand.nextLong() & 0x7FFFFFFFFFFFFFFFL) + "_" + counter.getAndIncrement();
  }
  
  public String createConsumerId() {
    return "cons_" + (rand.nextLong() & 0x7FFFFFFFFFFFFFFFL) + "_" + counter.getAndIncrement();
  }
  
  public SessionContext getSessionCtx(String name) {
    return sessionCtxs.get(name);
  }
  
  /**
   * Returns the context corresponding to the specified identifier.
   * If no matching context exists, try to create one.
   * 
   * Be careful, if there is too many existing contexts returns null.
   * 
   * @param id
   * @return
   */
  public RestClientContext getClientContext(String id) {
    RestClientContext ret = restClientCtxs.get(id);
    if (ret == null) {
      if ((maxNumberOfContext >0) && (restClientCtxs.size() >= maxNumberOfContext)) {
        return null;
      }
      
      String clientID = id;
      if (clientID == null) {
        clientID = createClientId();
        while (restClientCtxs.get(clientID) != null) {
          clientID = createClientId();
        }
        if (logger.isLoggable(BasicLevel.DEBUG))
          logger.log(BasicLevel.DEBUG, "Helper.getClientContext: create the new client Id " + clientID);
      }
      ret = new RestClientContext(clientID);
      restClientCtxs.put(clientID, ret);
    }
    return ret;
  }

  public void commit(String name) throws Exception {
    if (logger.isLoggable(BasicLevel.DEBUG))
      logger.log(BasicLevel.DEBUG, "Helper.commit " + name);
    SessionContext ctx = sessionCtxs.get(name);
    if (ctx == null)
      throw new Exception(name + " not found.");
    if (ctx.getJmsContext().getTransacted())
      ctx.getJmsContext().commit();
  }
  
  public void rollback(String name) throws Exception {
    if (logger.isLoggable(BasicLevel.DEBUG))
      logger.log(BasicLevel.DEBUG, "Helper.rollback " + name);
    SessionContext ctx = sessionCtxs.get(name);
    if (ctx == null)
      throw new Exception(name + " not found.");
    if (ctx.getJmsContext().getTransacted())
      ctx.getJmsContext().rollback();
  }
  
  public void acknowledgeAllMsg(String consName) throws Exception {
    if (logger.isLoggable(BasicLevel.DEBUG))
      logger.log(BasicLevel.DEBUG, "Helper.acknowledgeAllMsg " + consName);
    ConsumerContext consumerCtx = (ConsumerContext) sessionCtxs.get(consName);
    if (consumerCtx == null)
      throw new Exception(consName + " not found.");
    if (consumerCtx.getJmsContext().getSessionMode() == JMSContext.CLIENT_ACKNOWLEDGE) {
      consumerCtx.getJmsContext().acknowledge();
      consumerCtx.clear();
    }
  }

  public void acknowledgeMsg(String consName, long id) throws Exception {
    if (logger.isLoggable(BasicLevel.DEBUG))
      logger.log(BasicLevel.DEBUG, "Helper.acknowledgeMsg " + consName + ", " + id);
    ConsumerContext consumerCtx = (ConsumerContext) sessionCtxs.get(consName);
    if (consumerCtx == null)
      throw new Exception(consName + " not found.");
    if (consumerCtx.getJmsContext().getSessionMode() == JMSContext.CLIENT_ACKNOWLEDGE) {
      Message msg = consumerCtx.getMessage(id);
      msg.acknowledge();
      consumerCtx.removeMessage(id);
    }
  }
}
