/* JFox, the OpenSource J2EE Application Server
 *
 * Distributable under GNU LGPL license by gun.org
 * more details please visit http://www.huihoo.org/jfox
 */

package org.jfox.jms;

import java.rmi.RemoteException;
import java.rmi.server.UnicastRemoteObject;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import javax.jms.Connection;
import javax.jms.ConnectionConsumer;
import javax.jms.ConnectionMetaData;
import javax.jms.Destination;
import javax.jms.ExceptionListener;
import javax.jms.InvalidClientIDException;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.Queue;
import javax.jms.QueueConnection;
import javax.jms.QueueSession;
import javax.jms.ServerSessionPool;
import javax.jms.Session;
import javax.jms.Topic;
import javax.jms.TopicConnection;
import javax.jms.TopicSession;
import javax.jms.XAConnection;
import javax.jms.XAQueueConnection;
import javax.jms.XAQueueSession;
import javax.jms.XASession;
import javax.jms.XATopicConnection;
import javax.jms.XATopicSession;

import org.jfox.framework.util.UUID;
import org.jfox.jms.connector.JMSConnectorService;

/**
 * @author <a href="mailto:young_yy@hotmail.com">Young Yang</a>
 */

public class JMSConnection implements Connection,
        QueueConnection,
        TopicConnection,
        XAConnection,
        XAQueueConnection,
        XATopicConnection,
        JMSConnectionRemote {

    protected JMSConnectorService connectorService = null;
    private boolean started = false;
    private boolean closed = false;

    private JMSConnectionRemote stub = null;
    private String clientId = null;

    private boolean isXA = false;

    /*
     * sessions created by this connection
     * sessionId => session
     */
    private Map sessions = new HashMap();

    public JMSConnection(JMSConnectorService broker, boolean isXA) {
        this.connectorService = broker;
        this.isXA = isXA;
    }

    /**
     * If a <CODE>Session</CODE> is transacted, the acknowledgement mode
     * is ignored.
     *
     * @param transacted
     * @param acknowledgeMode
     * @return
     * @throws JMSException
     */
    public Session createSession(boolean transacted, int acknowledgeMode) throws JMSException {
        checkClosed();
        if(transacted) {
            acknowledgeMode = Session.SESSION_TRANSACTED;
        }
        JMSSession session = new JMSSession(this, transacted, acknowledgeMode, false);
        synchronized(sessions) {
            sessions.put(session.getSessionId(), session);
        }
        return session;
    }

    public String getClientID() throws JMSException {
        if(clientId == null) { // not set ClientID, auto set it
            this.setClientID(UUID.randomUUID().toString());
        }
        return clientId;
    }

    public void setClientID(String clientId) throws JMSException {
        checkClosed();
        if(started) {
            throw new IllegalStateException("connection has already started, can not set client identifier.");
        }
        if(this.clientId != null) {
            throw new IllegalStateException("client identifier has already been setted by system.");
        }
        try {
            boolean registered = connectorService.isConnectionRegistered(clientId);
            if(registered) {
                throw new InvalidClientIDException("client identifier " + clientId + " has alreay registered.");
            }
        }
        catch(RemoteException e) {
            throw new JMSException("setClientID error: ", e.getMessage());
        }
        this.clientId = clientId;
    }

    public ConnectionMetaData getMetaData() throws JMSException {
        throw new JMSException("not support now!");
    }

    public ExceptionListener getExceptionListener() throws JMSException {
        throw new JMSException("not support now!");
    }

    public void setExceptionListener(ExceptionListener listener) throws JMSException {
        throw new JMSException("not support now!");
    }

    /**
     * start to receive messages
     * first register the connection to JMSConnectorService
     *
     * @throws JMSException
     */
    public synchronized void start() throws JMSException {
        if(!started) {
            try {
                if(clientId == null) { // not set ClientID, auto set it
                    this.setClientID(UUID.randomUUID().toString());
                }
                stub = (JMSConnectionRemote) UnicastRemoteObject.exportObject(this);
                connectorService.registerConnection(this.getClientID(), stub);
            }
            catch(Exception e) {
                e.printStackTrace();
            }
            this.started = true;
        }
    }

    public synchronized void stop() throws JMSException {
        if(started) {
            try {
                UnicastRemoteObject.unexportObject(this, false);
                connectorService.unregisterConnection(this.getClientID());
            }
            catch(Exception e) {
                e.printStackTrace();
            }
            started = false;
        }
    }

    public synchronized void close() throws JMSException {
        if(closed) return;
        this.stop();
        synchronized(sessions) {
            for(Iterator it = sessions.values().iterator(); it.hasNext();) {
                JMSSession session = (JMSSession) it.next();
                session.close();
            }
            sessions.clear();
        }
    }

    public ConnectionConsumer createConnectionConsumer(Destination destination,
                                                       String messageSelector,
                                                       ServerSessionPool sessionPool,
                                                       int maxMessages) throws JMSException {
        checkClosed();
        throw new JMSException("not support now!");
    }

    public ConnectionConsumer createDurableConnectionConsumer(Topic topic,
                                                              String subscriptionName,
                                                              String messageSelector,
                                                              ServerSessionPool sessionPool,
                                                              int maxMessages) throws JMSException {
        checkClosed();
        throw new JMSException("not support now!");
    }

    public QueueSession createQueueSession(boolean transacted, int acknowledgeMode) throws JMSException {
        return (QueueSession) createSession(transacted, acknowledgeMode);
    }

    public ConnectionConsumer createConnectionConsumer(Queue queue,
                                                       String messageSelector,
                                                       ServerSessionPool sessionPool,
                                                       int maxMessages) throws JMSException {
        throw new JMSException("not support now!");
    }

    public TopicSession createTopicSession(boolean transacted, int acknowledgeMode) throws JMSException {
        return (TopicSession) createSession(transacted, acknowledgeMode);
    }

    public ConnectionConsumer createConnectionConsumer(Topic topic,
                                                       String messageSelector,
                                                       ServerSessionPool sessionPool,
                                                       int maxMessages) throws JMSException {
        throw new JMSException("not support now!");
    }

    public XASession createXASession() throws JMSException {
        checkClosed();
        if(!isXA) {
            throw new JMSException("current connection " + this + " is not an xa connection");
        }

        JMSSession session = new JMSSession(this, true, Session.SESSION_TRANSACTED, true);
        synchronized(sessions) {
            sessions.put(session.getSessionId(), session);
        }
        return session;
    }

    public XAQueueSession createXAQueueSession() throws JMSException {
        return (XAQueueSession) createXASession();
    }

    public XATopicSession createXATopicSession() throws JMSException {
        return (XATopicSession) createXASession();
    }

    public void onMessage(Message msg) throws RemoteException {
        System.out.println("onMessage: " + msg);
    }

    public void acknowledge(JMSObjectId msgId) {
        System.out.println("acknowlege msg: " + msgId);
    }

    public JMSConnectorService getConnectorService() {
        return connectorService;
    }

    boolean isStarted() {
        return started;
    }

    private void checkClosed() throws javax.jms.IllegalStateException {
        if(closed) {
            throw new javax.jms.IllegalStateException("connection closed");
        }
    }

    /**
     * do send message
     * all connection's session use the method to send messages
     */
    void sendMessage(Destination dest, Message msg, int deliveryMode, int priority, long timeToLive) {

    }

    Message receiveMessage() {
        return null;
    }

    public static void main(String[] args) {

    }
}
