/*
 *  DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS HEADER.
 *
 *  Copyright 2000-2009 Sun Microsystems, Inc. All rights reserved. 
 *
 *  The contents of this file are subject to the terms of either the GNU
 *  General Public License Version 2 only ("GPL") or the Common Development
 *  and Distribution License ("CDDL") (collectively, the "License").  You may
 *  not use this file except in compliance with the License.  You can obtain
 *  a copy of the License at https://glassfish.dev.java.net/public/CDDL+GPL.html
 *  or mq/legal/LICENSE.txt.  See the License for the specific language
 *  governing permissions and limitations under the License.
 * 
 *  When distributing the software, include this License Header Notice in each
 *  file and include the License file at mq/legal/LICENSE.txt.  Sun designates
 *  this particular file as subject to the "Classpath" exception as provided by
 *  Sun in the GPL Version 2 section of the License file that accompanied this
 *  code.  If applicable, add the following below the License Header, with the
 *  fields enclosed by brackets [] replaced by your own identifying information:
 *  "Portions Copyrighted [year] [name of copyright owner]"
 * 
 *  Contributor(s):
 * 
 *  If you wish your version of this file to be governed by only the CDDL or
 *  only the GPL Version 2, indicate your decision by adding "[Contributor]
 *  elects to include this software in this distribution under the [CDDL or GPL
 *  Version 2] license."  If you don't indicate a single choice of license, a
 *  recipient has the option to distribute your version of this file under
 *  either the CDDL, the GPL Version 2 or  to extend the choice of license to
 *  its licensees as provided above.  However, if you add GPL Version 2 code
 *  and therefore, elected the GPL Version 2 license, then the option applies
 *  only if the new code is made subject to such option by the copyright holder. 
 *
 *  DirectProducer.java
 *
 *  @(#)DirectProducer.java	1.9 07/12/07
 */

package com.sun.messaging.jms.ra;

import javax.jms.*;

import java.io.Serializable;
import java.util.HashMap;
import java.util.Iterator;
import java.util.logging.Logger;

import com.sun.messaging.jmq.jmsservice.JMSPacket;
import com.sun.messaging.jmq.jmsservice.JMSService;
import com.sun.messaging.jmq.jmsservice.JMSServiceReply;
import com.sun.messaging.jmq.jmsservice.JMSServiceException;
import com.sun.messaging.jmq.jmsservice.JMSService.MessageDeliveryMode;
import com.sun.messaging.jmq.jmsservice.JMSService.MessagePriority;

/**
 *  DirectProducer encapsulates JMS MessageProducer behavior for MQ DIRECT mode
 *  operation
 */
public class DirectProducer
        implements MessageProducer, QueueSender, TopicPublisher
    {
    
    /**
     *  The JMSService for this DirectProducer
     */
    private JMSService jmsservice;

    /**
     *  The parent DirectSession that created this DirectProducer
     */
    private DirectSession ds;

    /**
     *  The connectionId of the parent DirectConnectio
     */
    private long connectionId;

    /**
     *  The sessionId of the parent DirectSession
     */
    private long sessionId;

    /**
     *  The producerId for this DirectProducer when the Destination is specified
     *  at Producer creation time.
     */
    private long producerId = 0L;

    /**
     *  The default deliveyMode for this DirectProducer
     */
    private MessageDeliveryMode deliveryMode;
    private int jmsDeliveryMode = javax.jms.Message.DEFAULT_DELIVERY_MODE;

    /**
     *  The default message priority for this DirectProducer
     */
    private MessagePriority priority;
    private int jmsPriority = javax.jms.Message.DEFAULT_PRIORITY;

    /**
     *  The default timeToLive for messages produced by this DirectProducer
     */
    private long jmsTimeToLive = javax.jms.Message.DEFAULT_TIME_TO_LIVE;

    /**
     *  The JMS Destination that is associated with this DirectProducer when the
     *  Destination is specified at Producer creation time.
     */
    private Destination destination = null;

    /**
     *  The JMSService Destination that is associated with this DirectProducer
     */
    private com.sun.messaging.jmq.jmsservice.Destination jmsservice_destination;

    /**
     *  Holds whether MessageID is disabled or not
     */
    private boolean disableMessageID = false;

    /**
     *  Holds whether MessageTimestamp is disabled or not
     */
    private boolean disableMessageTimestamp = false;

    /**
     *  Holds the closed state of this DirectProducer
     */
    private boolean isClosed;
    private boolean isClosing;

    /**
     *  Holds the map of producerId entries for a DirectProducer that is created
     *  with an unspecified Destination
     */
    private HashMap<String, Long> producerIds;
    
    /** For optimized logging while messaging */
    protected int _logLevel;
    protected boolean _logFINE = false;

    /**
     *  Logging
     */
    private static transient final String _className =
            "com.sun.messaging.jms.ra.DirectProducer";
    private static transient final String _lgrNameOutboundConnection =
            "javax.resourceadapter.mqjmsra.outbound.connection";
    private static transient final String _lgrNameJMSProducer =
            "javax.jms.MessageProducer.mqjmsra";
    private static transient final Logger _loggerOC =
            Logger.getLogger(_lgrNameOutboundConnection);
    private static transient final Logger _loggerJMP =
            Logger.getLogger(_lgrNameJMSProducer);
    private static transient final String _lgrMIDPrefix = "MQJMSRA_DMP";
    private static transient final String _lgrMID_EET = _lgrMIDPrefix + "1001: ";
    private static transient final String _lgrMID_INF = _lgrMIDPrefix + "1101: ";
    private static transient final String _lgrMID_WRN = _lgrMIDPrefix + "2001: ";
    private static transient final String _lgrMID_ERR = _lgrMIDPrefix + "3001: ";
    private static transient final String _lgrMID_EXC = _lgrMIDPrefix + "4001: ";

    /** Creates a new instance of DirectProducer with unspecified destination */
    public DirectProducer(DirectSession ds, JMSService jmsservice) {
        Object params[] = new Object[2];
        params[0] = ds;
        params[1] = jmsservice;
        _loggerOC.entering(_className, "constructor()", params);        
        this.ds = ds;
        this.jmsservice = jmsservice;
        this.connectionId = ds.getConnectionId();
        this.sessionId = ds.getSessionId();
        producerIds = new HashMap<String, Long>();
    }

    /** Creates a new instance of DirectProducer with a specified destination */
    public DirectProducer(DirectSession ds, JMSService jmsservice,
            long producerId, Destination destination,
            com.sun.messaging.jmq.jmsservice.Destination jmsservice_dest) {
        Object params[] = new Object[5];
        params[0] = ds;
        params[1] = jmsservice;
        params[2] = producerId;
        params[3] = destination;
        params[4] = jmsservice_dest;
        _loggerOC.entering(_className, "constructor()", params);        
        this.ds = ds;
        this.jmsservice = jmsservice;
        this.producerId = producerId;
        this.destination = destination;
        if (this.destination == null) {
            //Initialize producerIds table if DirectProducer created for
            //unspecified Destination (i.e. null)
            producerIds = new HashMap<String, Long>();
        }
        this.jmsservice_destination = jmsservice_dest;
        this.connectionId = ds.getConnectionId();
        this.sessionId = ds.getSessionId();
        java.util.logging.Level _level = _loggerJMP.getLevel();
        if (_level != null) {
            this._logLevel = _level.intValue();
            if (this._logLevel <= java.util.logging.Level.FINE.intValue()){
                this._logFINE = true;
            }
        }
    }

    /////////////////////////////////////////////////////////////////////////
    //  methods that implement javax.jms.MessageProducer
    /////////////////////////////////////////////////////////////////////////
    /**
     *  Close this MessageProducer
     */
    public synchronized void close()
    throws JMSException {
        _loggerJMP.fine(_lgrMID_INF+"producerId="+producerId+":"+"close():");
        //harmless if already closed
        if (isClosed){
            return;
        } else {
            ds.removeProducer(this);
            this._close();
        }
    }

    /**
     *  Return the default JMS Message DeliveryMode for this MessageProducer
     *
     *  @return The deliveryMode
     *
     *  @throws javax.jms.JMSException If the MessageProducer has been closed.
     *
     *  @see javax.jms.DeliveryMode javax.jms.DeliveryMode
     */
    public int getDeliveryMode()
    throws JMSException{
        _loggerJMP.fine(_lgrMID_INF+"producerId="+producerId+":"+
                "getDeliveryMode():"+this.deliveryMode);
        this._checkIfClosed("getDeliveryMode()");
        return this.jmsDeliveryMode;
        /*
        switch(this.deliveryMode) {
            case NON_PERSISTENT:
                return javax.jms.DeliveryMode.NON_PERSISTENT;
            case PERSISTENT:
            default:
                return javax.jms.DeliveryMode.PERSISTENT;
        }
        */
    }

    /**
     *  Return the Destination associated with this MessageProducer
     *
     *  @return The Destination
     *
     *  @throws javax.jms.JMSException If the MessageProducer has been closed.
     */
    public Destination getDestination()
    throws JMSException {
        String methodName = "getDestination()";
        _loggerJMP.fine(_lgrMID_INF+"producerId="+producerId+":"+
                methodName +
                ((this.jmsservice_destination == null) 
                ? null : this.jmsservice_destination.toString()));
        this._checkIfClosed("getDestination()");
        return this.destination;
    }

    /**
     *  Return whether MessageIDs are disabled by this MessageProducer.<p>
     *
     *  @return {@code true} If MessageIDs are diabled;<br> {@code false}
     *          If MessageIDs are not diabled.
     *
     *  @throws javax.jms.JMSException If the MessageProducer has been closed.
     */
    public boolean getDisableMessageID()
    throws JMSException {
        _loggerJMP.fine(_lgrMID_INF+"producerId="+producerId+":"+
                "getDisableMessageID():"+this.disableMessageID);
        this._checkIfClosed("getDisableMessageID()");
        return this.disableMessageID;
    }

    /**
     *  Return whether MessageIDs are disabled by this MessageProducer.<p>
     *
     *  @return {@code true} If MessageIDs are diabled;<br> {@code false}
     *          If MessageIDs are not diabled.
     *
     *  @throws javax.jms.JMSException If the MessageProducer has been closed.
     */
    public boolean getDisableMessageTimestamp()
    throws JMSException {
        _loggerJMP.fine(_lgrMID_INF+"producerId="+producerId+":"+
                "getDisableMessageTimestamp():"+this.disableMessageTimestamp);
        this._checkIfClosed("getDisableMessageTimestamp()");
        return this.disableMessageTimestamp;
    }

    /**
     *  Return the default JMS Message Priority for this MessageProducer
     *
     *  @return The priority
     *
     *  @throws javax.jms.JMSException If the MessageProducer has been closed.
     *
     *  @see javax.jms.Message#DEFAULT_PRIORITY javax.jms.Message.DEFAULT_PRIORITY
     *  @see com.sun.messaging.jmq.jmsservice.JMSService.MessagePriority
     */
    public int getPriority()
    throws JMSException{
        _loggerJMP.fine(_lgrMID_INF+"producerId="+producerId+":"+
                "getPriority():"+this.priority);
        this._checkIfClosed("getPriority()");
        return this.jmsPriority;
        /*
        switch(this.priority) {
            case PRIORITY_0:
                return 0;
            case PRIORITY_1:
                return 1;
            case PRIORITY_2:
                return 2;
            case PRIORITY_3:
                return 3;
            case PRIORITY_4:
                return 4;
            case PRIORITY_5:
                return 5;
            case PRIORITY_6:
                return 6;
            case PRIORITY_7:
                return 7;
            case PRIORITY_8:
                return 8;
            case PRIORITY_9:
                return 9;
            default:
                return javax.jms.Message.DEFAULT_PRIORITY;
        }
        */
    }

    /**
     *  Return the timeToLive associated with this MessageProducer
     *
     *  @return The timeToLive
     *
     *  @throws javax.jms.JMSException If the MessageProducer has been closed.
     */
    public long getTimeToLive()
    throws JMSException {
        _loggerJMP.fine(_lgrMID_INF+"producerId="+producerId+":"+
                "getTimeToLive():"+this.jmsTimeToLive);
        this._checkIfClosed("getTimeToLive()");
        return this.jmsTimeToLive;
    }

    /**
     *  Send a Message to the specified Destination using this MessageProducer
     *
     *  @param  message The JMS Message to send
     *  @param  destination The JMS Destination to use when sending the
     *          message.
     *
     *  @throws javax.jms.JMSException If the MessageProducer has been closed or
     *          it fails to send the message due to an internal error in the
     *          provider.
     *  @throws javax.jms.MessageFormatException If an invalid Message is used
     *  @throws javax.jms.InvalidDestinationException If an invalid Destination
     *          is used.
     *  @throws javax.jms.UnsupportedOperationException If this method is used
     *          with a MessageProducer hat specified a Destination at creation
     *          time.
     */
    public void send(Destination destination, Message message)
    throws JMSException {
        String methodName = "send(Destination, Message)";
        if (this._logFINE){
            _loggerJMP.fine(_lgrMID_INF+"producerId="+producerId+":"+ methodName);
        }
        this._checkIfClosed(methodName);
        this._checkDestinationParam(methodName, destination);
        this._createAndAddProducerId(destination);
        this._send(destination, message,
                this.jmsDeliveryMode, this.jmsPriority, this.jmsTimeToLive);
    }

    /**
     *  Send a Message to the specified Destination using this MessageProducer
     *
     *  @param  message The JMS Message to send
     *  @param  destination The JMS Destination to use when sending the
     *          message.
     *  @param  deliveryMode The JMS DeliveryMode to use when sending the
     *          message.
     *  @param  priority The JMS Priority to use when sending the message.
     *  @param  timeToLive The JMS Expiration to use when sending the message
     *          in milliseconds.
     *
     *  @throws javax.jms.JMSException If the MessageProducer has been closed or
     *          it fails to send the message due to an internal error in the
     *          provider.
     *  @throws javax.jms.MessageFormatException If an invalid Message is used
     *  @throws javax.jms.InvalidDestinationException If an invalid Destination
     *          is used.
     *  @throws javax.jms.UnsupportedOperationException If this method is used
     *          with a MessageProducer hat specified a Destination at creation
     *          time.
     */
    public void send(Destination destination, Message message, int deliveryMode,
            int priority, long timeToLive)
    throws JMSException {
        String methodName = "send(Destination, Message, deliveryMode, priority, timeToLive)";
        if (this._logFINE){
            _loggerJMP.fine(_lgrMID_INF+"producerId="+producerId+":"+ methodName);
        }
        this._checkIfClosed(methodName);
        this._checkDestinationParam(methodName, destination);
        this._createAndAddProducerId(destination);
        this._send(destination, message, deliveryMode, priority, timeToLive);
    }

    /**
     *  Send a Message to the Destination specified by this MessageProducer
     *  using the values of deliveryMode, priority, timeToLive that are
     *  configured in this MessageProducer.
     *
     *  @param  message The JMS Message to send
     *
     *  @throws javax.jms.JMSException If the MessageProducer has been closed or
     *          it fails to send the message due to an internal error in the
     *          provider.
     *  @throws javax.jms.MessageFormatException If an invalid Message is used
     *  @throws javax.jms.InvalidDestinationException If the Destination is
     *          invalid at the time of the send.
     *  @throws javax.jms.UnsupportedOperationException If this method is used
     *          with a MessageProducer that did not specify a Destination at
     *          creation time.
     */
    public void send(Message message)
    throws JMSException {
        String methodName = "send(Message)";
        if (this._logFINE){
            _loggerJMP.fine(_lgrMID_INF+"producerId="+producerId+":"+ methodName);
        }
        this._checkIfClosed(methodName);
        this._checkForUnsupportedOperation(methodName, null, true);
        this._send(this.destination, message,
                this.jmsDeliveryMode, this.jmsPriority, this.jmsTimeToLive);
    }

    /**
     *  Send a Message to the Destination specified by this MessageProducer
     *  using the specified deliveryMode, priority, and timeToLive.
     *
     *  @param  message The JMS Message to send
     *  @param  deliveryMode The JMS DeliveryMode to use when sending the
     *          message.
     *  @param  priority The JMS Priority to use when sending the message.
     *  @param  timeToLive The JMS Expiration to use when sending the message
     *          in milliseconds.
     *
     *  @throws javax.jms.JMSException If the MessageProducer has been closed or
     *          it fails to send the message due to an internal error in the
     *          provider.
     *  @throws javax.jms.MessageFormatException If an invalid Message is used
     *  @throws javax.jms.InvalidDestinationException If the Destination is
     *          invalid at the time of the send.
     *  @throws javax.jms.UnsupportedOperationException If this method is used
     *          with a MessageProducer that did not specify a Destination at
     *          creation time.
     */
    public void send(Message message, int deliveryMode, int priority,
            long timeToLive)
    throws JMSException {
        String methodName = "send(Message, deliveryMode, priority, timeToLive)";
        _loggerJMP.fine(_lgrMID_INF+"producerId="+producerId+":"+ methodName);
        this._checkIfClosed(methodName);
        this._checkForUnsupportedOperation(methodName, null, true);
        this._send(this.destination, message, deliveryMode, priority, timeToLive);
        
    }

    /**
     *  Set the default JMS Message DeliveryMode for this MessageProducer
     *
     *  @param  deliveryMode The deliveryMode to set for this MessageProducer
     *
     *  @throws javax.jms.JMSException If the MessageProducer has been closed.
     *
     *  @see javax.jms.DeliveryMode javax.jms.DeliveryMode
     */
    public void setDeliveryMode(int deliveryMode)
    throws JMSException{
        _loggerJMP.fine(_lgrMID_INF+"producerId="+producerId+":"+
                "setDeliveryMode():"+deliveryMode);
        this._checkIfClosed("getDeliveryMode()");
        switch(deliveryMode) {
            case javax.jms.DeliveryMode.PERSISTENT:
                this.deliveryMode = JMSService.MessageDeliveryMode.PERSISTENT;
                this.jmsDeliveryMode = javax.jms.DeliveryMode.PERSISTENT;
                break;
            case javax.jms.DeliveryMode.NON_PERSISTENT:
                this.deliveryMode = JMSService.MessageDeliveryMode.NON_PERSISTENT;
                this.jmsDeliveryMode = javax.jms.DeliveryMode.NON_PERSISTENT;
                break;
            default:
                String excMsg = _lgrMID_EXC + "setDeliveryMode():" +
                    "Invalid deliveryMode=" + deliveryMode;
            _loggerJMP.warning(excMsg);
            throw new JMSException(excMsg);
        }
    }

    /**
     *  Set whether MessageID is disabled by this MessageProducer.<p>
     *
     *  @param  disableMessageID {@code true} if MessageIDs are to be disabled<br>
     *          {@code false} if MessageIDs are to be enabled
     *
     *  @throws javax.jms.JMSException If the MessageProducer has been closed.
     */
    public void setDisableMessageID(boolean disableMessageID)
    throws JMSException {
        _loggerJMP.fine(_lgrMID_INF+"producerId="+producerId+":"+
                "setDisableMessageID():"+disableMessageID);
        this._checkIfClosed("setDisableMessageID()");
        this.disableMessageID = disableMessageID;
    }

    /**
     *  Set whether MessageTimestamp is disabled by this MessageProducer.<p>
     *
     *  @param  disableMessageTimestamp {@code true} If MessageTimestamp is
     *          diabled<br>
     *          {@code false} If MessageTimestamp is not diabled.
     *
     *  @throws javax.jms.JMSException If the MessageProducer has been closed.
     */
    public void setDisableMessageTimestamp(boolean disableMessageTimestamp)
    throws JMSException {
        _loggerJMP.fine(_lgrMID_INF+"producerId="+producerId+":"+
                "setDisableMessageTimestamp():"+disableMessageTimestamp);
        this._checkIfClosed("setDisableMessageTimestamp()");
        this.disableMessageTimestamp = disableMessageTimestamp;
    }

    /**
     *  Set the JMS Message Priority for this MessageProducer
     *
     *  @param  priority The priority to set for this MessageProducer
     *
     *  @throws javax.jms.JMSException If the MessageProducer has been closed.
     *
     *  @see javax.jms.Message#DEFAULT_PRIORITY javax.jms.Message.DEFAULT_PRIORITY
     *  @see com.sun.messaging.jmq.jmsservice.JMSService.MessagePriority
     */
    public void setPriority(int priority)
    throws JMSException{
        _loggerJMP.fine(_lgrMID_INF+"producerId="+producerId+":"+
                "setPriority():"+priority);
        this._checkIfClosed("setPriority()");
        switch(priority) {
            case 0:
                this.priority = JMSService.MessagePriority.PRIORITY_0;
                break;
            case 1:
                this.priority = JMSService.MessagePriority.PRIORITY_1;
                break;
            case 2:
                this.priority = JMSService.MessagePriority.PRIORITY_2;
                break;
            case 3:
                this.priority = JMSService.MessagePriority.PRIORITY_3;
                break;
            case 4:
                this.priority = JMSService.MessagePriority.PRIORITY_4;
                break;
            case 5:
                this.priority = JMSService.MessagePriority.PRIORITY_5;
                break;
            case 6:
                this.priority = JMSService.MessagePriority.PRIORITY_6;
                break;
            case 7:
                this.priority = JMSService.MessagePriority.PRIORITY_7;
                break;
            case 8:
                this.priority = JMSService.MessagePriority.PRIORITY_8;
                break;
            case 9:
                this.priority = JMSService.MessagePriority.PRIORITY_9;
                break;
            default:
                String excMsg = _lgrMID_EXC + "setPriority():" +
                    "Invalid to set priority outside (0-9):value=" + priority;
            _loggerJMP.warning(excMsg);
            throw new JMSException(excMsg);
        }
        this.jmsPriority = priority;
    }

    /**
     *  Set the default timeToLive associated with this MessageProducer
     *
     *  @param  timeToLive The default message expiration time in milliseconds
     *
     *  @throws javax.jms.JMSException If the MessageProducer has been closed.
     */
    public void setTimeToLive(long timeToLive)
    throws JMSException {
        _loggerJMP.fine(_lgrMID_INF+"producerId="+producerId+":"+
                "setTimeToLive():"+timeToLive);
        this._checkIfClosed("setTimeToLive()"+timeToLive);
        if (timeToLive < 0){
                String excMsg = _lgrMID_EXC + "setTimeToLive():" +
                    "Invalid to set timeToLive as negative:value=" + timeToLive;
            _loggerJMP.warning(excMsg);
            throw new JMSException(excMsg);
        } else {
            this.jmsTimeToLive = timeToLive;
        }
    }
    /////////////////////////////////////////////////////////////////////////
    //  end javax.jms.MessageProducer
    /////////////////////////////////////////////////////////////////////////
    /////////////////////////////////////////////////////////////////////////
    //  methods that implement javax.jms.QueueSender
    /////////////////////////////////////////////////////////////////////////
    /**
     *  Return the Queue associated with this QueueSender
     *
     *  @return The Queue
     *
     *  @throws javax.jms.JMSException If the QueueSender has been closed.
     */
    public Queue getQueue()
    throws JMSException {
        String methodName = "getQueue()";
        _loggerJMP.fine(_lgrMID_INF+"producerId="+producerId+":" + methodName);
        this._checkIfClosed(methodName);
        if (this.destination == null) {
            return (Queue)null;
        } else {
            if (this.destination instanceof Queue) {
                return (Queue)this.destination;
            }
        }
        _loggerJMP.fine(_lgrMID_INF+"producerId="+producerId+":"+
                methodName + " called on Destination:"+this.destination);
        return (Queue)null;
    }

    /**
     *  Send a Message to the specified Queue using this QueueSender
     *
     *  @param  message The JMS Message to send
     *  @param  queue The JMS Queue to use when sending the message.
     *
     *  @throws javax.jms.JMSException If the QueueSender has been closed or
     *          it fails to send the message due to an internal error in the
     *          provider.
     *  @throws javax.jms.MessageFormatException If an invalid Message is used
     *  @throws javax.jms.InvalidDestinationException If an invalid Destination
     *          is used.
     *  @throws javax.jms.UnsupportedOperationException If this method is used
     *          with a QueueSender hat specified a Destination at creation
     *          time.
     */
    public void send(Queue queue, Message message)
    throws JMSException {
        String methodName = "send(Queue, Message)";
        if (this._logFINE){
            _loggerJMP.fine(_lgrMID_INF+"producerId="+producerId+":"+ methodName);
        }
        this._checkIfClosed(methodName);
        this._checkDestinationParam(methodName, queue);
        this._createAndAddProducerId(queue);
        this._send(queue, message,
                this.jmsDeliveryMode, this.jmsPriority, this.jmsTimeToLive);
    }

    /**
     *  Send a Message to the specified Queue using this QueueSender
     *
     *  @param  message The JMS Message to send
     *  @param  queue The JMS Queue to use when sending the
     *          message.
     *  @param  deliveryMode The JMS DeliveryMode to use when sending the
     *          message.
     *  @param  priority The JMS Priority to use when sending the message.
     *  @param  timeToLive The JMS Expiration to use when sending the message
     *          in milliseconds.
     *
     *  @throws javax.jms.JMSException If the QueueSender has been closed or
     *          it fails to send the message due to an internal error in the
     *          provider.
     *  @throws javax.jms.MessageFormatException If an invalid Message is used
     *  @throws javax.jms.InvalidDestinationException If an invalid Queue
     *          is used.
     *  @throws javax.jms.UnsupportedOperationException If this method is used
     *          with a QueueSender that specified a Queue at creation
     *          time.
     */
    public void send(Queue queue, Message message, int deliveryMode,
            int priority, long timeToLive)
    throws JMSException {
        String methodName = "send(Queue, Message, deliveryMode, priority, timeToLive)";
        if (this._logFINE){
            _loggerJMP.fine(_lgrMID_INF+"producerId="+producerId+":"+ methodName);
        }
        this._checkIfClosed(methodName);
        this._checkDestinationParam(methodName, queue);
        this._createAndAddProducerId(queue);
        this._send(queue, message, deliveryMode, priority, timeToLive);
    }
    /////////////////////////////////////////////////////////////////////////
    //  end javax.jms.QueueSender
    /////////////////////////////////////////////////////////////////////////
    /////////////////////////////////////////////////////////////////////////
    //  methods that implement javax.jms.TopicPublisher
    /////////////////////////////////////////////////////////////////////////
    /**
     *  Return the Topic associated with this TopicPublisher
     *
     *  @return The Topic
     *
     *  @throws javax.jms.JMSException If the TopicPublisher has been closed.
     */
    public Topic getTopic()
    throws JMSException {
        String methodName = "getTopic()";
        _loggerJMP.fine(_lgrMID_INF+"producerId="+producerId+":"+methodName);
        this._checkIfClosed(methodName);
        if (this.destination == null){
            return (Topic)null;
        } else {
            if (this.destination instanceof Topic) {
                return (Topic)this.destination;
            }
        }
        _loggerJMP.fine(_lgrMID_INF+"producerId="+producerId+":"+
                methodName+" called on Destination:"+this.destination);
        return (Topic)null;
    }

    /**
     *  Publish a Message to the Topic specified by this TopicPublisher
     *  using the values of deliveryMode, priority, timeToLive that are
     *  configured in this TopicPublisher.
     *
     *  @param  message The JMS Message to send
     *
     *  @throws javax.jms.JMSException If the TopicPublisher has been closed or
     *          it fails to send the message due to an internal error in the
     *          provider.
     *  @throws javax.jms.MessageFormatException If an invalid Message is used
     *  @throws javax.jms.InvalidDestinationException If the Topic is
     *          invalid at the time of the send.
     *  @throws javax.jms.UnsupportedOperationException If this method is used
     *          with a TopicPublisher that did not specify a Topic at
     *          creation time.
     */
    public void publish(Message message)
    throws JMSException {
        String methodName = "publish(Message)";
        _loggerJMP.fine(_lgrMID_INF+"producerId="+producerId+":"+ methodName);
        this._checkIfClosed(methodName);
        this._checkForUnsupportedOperation(methodName, null, true);
        this._send(this.destination, message,
                this.jmsDeliveryMode, this.jmsPriority, this.jmsTimeToLive);
    }

    /**
     *  Send a Message to the Topic specified by this TopicPublisher
     *  using the specified deliveryMode, priority, and timeToLive.
     *
     *  @param  message The JMS Message to send
     *  @param  deliveryMode The JMS DeliveryMode to use when sending the
     *          message.
     *  @param  priority The JMS Priority to use when sending the message.
     *  @param  timeToLive The JMS Expiration to use when sending the message
     *          in milliseconds.
     *
     *  @throws javax.jms.JMSException If the TopicPublisher has been closed or
     *          it fails to send the message due to an internal error in the
     *          provider.
     *  @throws javax.jms.MessageFormatException If an invalid Message is used
     *  @throws javax.jms.InvalidDestinationException If the Topic is
     *          invalid at the time of the send.
     *  @throws javax.jms.UnsupportedOperationException If this method is used
     *          with a TopicPublisher that did not specify a Destination at
     *          creation time.
     */
    public void publish(Message message, int deliveryMode, int priority,
            long timeToLive)
    throws JMSException {
        String methodName = "publish(Message, deliveryMode, priority, timeToLive)";
        if (this._logFINE){
            _loggerJMP.fine(_lgrMID_INF+"producerId="+producerId+":"+ methodName);
        }
        this._checkIfClosed(methodName);
        this._checkForUnsupportedOperation(methodName, null, true);
        this._send(this.destination, message, deliveryMode, priority, timeToLive);
    }

    /**
     *  Publish a Message to the specified Topic using this TopicPublisher
     *
     *  @param  message The JMS Message to send
     *  @param  topic The JMS Topic to use when sending the message.
     *
     *  @throws javax.jms.JMSException If the TopicPublisher has been closed or
     *          it fails to send the message due to an internal error in the
     *          provider.
     *  @throws javax.jms.MessageFormatException If an invalid Message is used
     *  @throws javax.jms.InvalidDestinationException If an invalid Destination
     *          is used.
     *  @throws javax.jms.UnsupportedOperationException If this method is used
     *          with a TopicPublisher that specified a Topic at creation
     *          time.
     */
    public void publish(Topic topic, Message message)
    throws JMSException {
        String methodName = "publish(Topic, Message)";
        if (this._logFINE){
            _loggerJMP.fine(_lgrMID_INF+"producerId="+producerId+":"+ methodName);
        }
        this._checkIfClosed(methodName);
        this._checkDestinationParam(methodName, topic);
        this._createAndAddProducerId(topic);
        this._send(topic, message,
                this.jmsDeliveryMode, this.jmsPriority, this.jmsTimeToLive);
    }

    /**
     *  Publish a Message to the specified Topic using this TopicPublisher
     *
     *  @param  message The JMS Message to send
     *  @param  topic The JMS Topic to use when sending the
     *          message.
     *  @param  deliveryMode The JMS DeliveryMode to use when sending the
     *          message.
     *  @param  priority The JMS Priority to use when sending the message.
     *  @param  timeToLive The JMS Expiration to use when sending the message
     *          in milliseconds.
     *
     *  @throws javax.jms.JMSException If the TopicPublisher has been closed or
     *          it fails to send the message due to an internal error in the
     *          provider.
     *  @throws javax.jms.MessageFormatException If an invalid Message is used
     *  @throws javax.jms.InvalidDestinationException If an invalid Topic
     *          is used.
     *  @throws javax.jms.UnsupportedOperationException If this method is used
     *          with a TopicPublisher that specified a Topic at creation
     *          time.
     */
    public void publish(Topic topic, Message message, int deliveryMode,
            int priority, long timeToLive)
    throws JMSException {
        String methodName = "publish(Topic, Message, deliveryMode, priority, timeToLive)";
        if (this._logFINE){
            _loggerJMP.fine(_lgrMID_INF+"producerId="+producerId+":"+ methodName);
        }
        this._checkIfClosed(methodName);
        this._checkDestinationParam(methodName, topic);
        this._createAndAddProducerId(topic);
        this._send(topic, message, deliveryMode, priority, timeToLive);
    }
    /////////////////////////////////////////////////////////////////////////
    //  end javax.jms.TopicPublisher
    /////////////////////////////////////////////////////////////////////////
    /////////////////////////////////////////////////////////////////////////
    //  MQ methods
    /////////////////////////////////////////////////////////////////////////
    /**
     *  Return the connectionId for this DirectProducer
     *
     *  @return The connectionId
     */
    public long getConnectionId() {
        return this.connectionId;
    }

    /**
     *  Return the sessionId for this DirectProducer
     *
     *  @return The sessionId
     */
    public long getSessionId() {
        return this.sessionId;
    }

    /**
     *  Return the producerId for this DirectProducer
     *
     *  @return The producerId
     */
    public long getProducerId() {
        return this.producerId;
    }

    /**
     *  Return the closed state of this DirectProducer
     *
     *  @return {@code true} if this producer has been closed;
     *          {@code false} otherwise
     */
    public synchronized boolean isClosed() {
        return this.isClosed;
    }

    /**
     *  Check if the DirectProducer is closed prior to performing an
     *  operation and throw a JMSException if it is closed.
     *
     *  @param  methodname The name of the method from which this is called
     *
     *  @throws JMSException if it is closed
     */
    private void _checkIfClosed(String methodname)
    throws JMSException {
        if (isClosed()) {
            String closedmsg = _lgrMID_EXC + methodname +
                    "MessageProducer is closed:Id=" + producerId;
            _loggerJMP.warning(closedmsg);
            throw new javax.jms.IllegalStateException(closedmsg);
        }
    }

    /**
     *  Create and add a producerId to this DirectProducer
     */
    private void _createAndAddProducerId(Destination destination)
    throws JMSException {
        //Spcified destination case
        String _name = null, dName = null;
        com.sun.messaging.jmq.jmsservice.Destination.Type _type =
                com.sun.messaging.jmq.jmsservice.Destination.Type.QUEUE;
        com.sun.messaging.jmq.jmsservice.Destination.Life _life =
                com.sun.messaging.jmq.jmsservice.Destination.Life.STANDARD;

        if (destination instanceof javax.jms.Queue){
            _name = ((Queue)destination).getQueueName();
            _type = com.sun.messaging.jmq.jmsservice.Destination.Type.QUEUE;
            if (destination instanceof javax.jms.TemporaryQueue){
                _life =
                    com.sun.messaging.jmq.jmsservice.Destination.Life.TEMPORARY;
                dName = _name;
            } else {
                dName = _type + _name;
            }
            
        }
        if (destination instanceof javax.jms.Topic){
            _name = ((Topic)destination).getTopicName();
            _type = com.sun.messaging.jmq.jmsservice.Destination.Type.TOPIC;
            if (destination instanceof javax.jms.TemporaryTopic){
                _life =
                    com.sun.messaging.jmq.jmsservice.Destination.Life.TEMPORARY;
                dName = _name;
            } else {
                dName = _type + ":" + _name;
            }
        }
        com.sun.messaging.jmq.jmsservice.Destination _destination =
                new com.sun.messaging.jmq.jmsservice.Destination(_name,
                _type, _life);
        //XXX:tharakan:Need to Verify Destination first ???
        //dc._verifyDestination(_destination);
        //Destination may have been removed even if it was created before
        this.ds.getConnection()._createDestination(_destination);
        //Check producerIds for existing producerId for this destination (dName)
        if (!this.producerIds.containsKey(dName)) {
            long producerId = this.ds._createProducerId(_destination);
            this.producerIds.put(dName, producerId);
        }
    }
    /**
     *  Check if a null Destination was used when a non-null Destination is
     *  required.
     *
     *  @param  methodname The name of the method from which this is called
     *  @param  dest The Destination parameter used to the calling
     *          method
     *
     *  @throws InvalidDestinationException If a null Destination is used
     *  @throws UnsupportedOperationException If a non-null Destination
     *          parameter method is used to produce a message when the
     *          MessageProducer has been created with a specified Destination
     */
    private void _checkDestinationParam(String methodname, Destination dest)
    throws JMSException {
        if (dest == null){
            String nullDestMsg = _lgrMID_EXC + methodname +
                    ":Null Destination used:";
            _loggerJMP.warning(nullDestMsg);
            throw new InvalidDestinationException(nullDestMsg);
        } else {
            this._checkForUnsupportedOperation(methodname, dest, false);
        }
    }

    /**
     *  Check for UnsupportedOperation
     *
     *  @param  methodname The name of the method from which this is called
     *  @param  dest The Destination parameter used to the calling
     *          method
     *  @param  mpDest_Required Indicates whether the MessageProducer
     *          destination is required to be non-null or not.<br>
     *          {@code true} if it is required to be non-null.<br>
     *          {@code false} if it is required to be null
     */
    private void _checkForUnsupportedOperation(String methodname,
            Destination dest, boolean mpDest_Required) {
        if (mpDest_Required == false){
            //the MessageProducer destination is required to be null
            //If found throw UnsupportedOperationException
            if (this.destination != null){
                String nonNullDestMsg = null;
                String dest_name = null;
                try {
                    dest_name = ((dest instanceof javax.jms.Queue)
                        ? ((Queue) dest).getQueueName() 
                        : ((Topic) dest).getTopicName());
                } catch (JMSException ex) {
                    dest_name = "null";
                }
                nonNullDestMsg = _lgrMID_EXC + methodname +
                        ":Unsupported Operation:" + 
                        "Producing to destination:" + dest_name +
                        ":in MessageProducer with specified Destination";
                if (this.jmsservice_destination != null) {
                    nonNullDestMsg = nonNullDestMsg + "=" +
                        this.jmsservice_destination.getType() + ":" +
                        this.jmsservice_destination.getName();
                }
                _loggerJMP.warning(nonNullDestMsg);
                throw new UnsupportedOperationException(nonNullDestMsg);
            }
        } else {
            //the MessageProducer destination is required to be non-null
            //If not found throw UnsupportedOperationException
            if (this.destination == null){
                String nullDestMsg = _lgrMID_EXC + methodname + 
                        ":Unsupported Operation:"+
                        "Producing destination not specified (null) " +
                        "in MessageProducer with unspecified Destination";
                _loggerJMP.warning(nullDestMsg);
                throw new UnsupportedOperationException(nullDestMsg);
            }
        }
    }

    /**
     *  Send a Message
     */
    private void _send(Destination destination, Message msg, int deliveryMode,
            int priority, long timeToLive)
    throws JMSException{
        JMSPacket msgPkt = null;
        DirectPacket dPkt = null;
        boolean foreignMessageConverted = false;
        //System.out.println("_send:Message is instanceof-"+msg.getClass().getName());
        if (msg instanceof DirectPacket) {
            dPkt = (DirectPacket)msg;
        } else {
            dPkt = DirectPacket.constructFromForeignMessage(this.jmsservice,
                    this.ds, msg);
            foreignMessageConverted = true;
        }
        assert dPkt != null;
        dPkt.setJMSDestination(destination);
        dPkt.setJMSDeliveryMode(deliveryMode);
        dPkt.setJMSPriority((priority));
        dPkt.setJMSExpiration(timeToLive);
        dPkt.preparePacketForSend();
        msgPkt = (JMSPacket)dPkt;
        this.ds._sendMessage(msgPkt);
        if (foreignMessageConverted){
            DirectPacket.updateForeignMessageAfterSend(dPkt, msg);
        }
    }

    /**
     *  Close Producer for use when Session.close is used
     */
    protected synchronized void _close()
    throws JMSException{
        //harmless if already closed
        if (isClosed){
            return;
        } else {
            isClosing = true;
            //Anything?
        }
        try {
            if (destination !=null) {
                //System.out.println("DP:Destroying prducerId="+producerId+":connectionId="+connectionId);
                jmsservice.deleteProducer(connectionId, sessionId, producerId);
            } else {
                //must cycle through producerIds and delete them one by one
                long t_producerId;
                String dName = null;
                Iterator<String> k = producerIds.keySet().iterator();
                while (k.hasNext()) {
                    dName = k.next();
                    t_producerId = producerIds.get(dName);
                    try {
                        //System.out.println("DP:Destroying t_producerId="+t_producerId);
                        jmsservice.deleteProducer(connectionId, sessionId,
                                t_producerId);
                    } catch (JMSServiceException jmssel){
                        _loggerJMP.warning(_lgrMID_WRN+
                                "producerId="+t_producerId+":"+"close():"+
                                "JMSService.deleteProducer():"+
                                "Destination=:" + dName +
                                ":JMSServiceException="+
                                jmssel.getMessage());
                    }
                    k.remove();
                }
                producerIds.clear();
            }
        } catch (JMSServiceException jmsse){
            _loggerJMP.warning(_lgrMID_WRN+
                    "producerId="+producerId+":"+"close():"+
                    "JMSService.deleteProducer():"+
                    "Destination=:" + destination +
                    ":JMSServiceException="+
                    jmsse.getMessage());
        }
        this.isClosed = true;
        this.isClosing = false;
    }
}
