/*
 * The contents of this file are subject to the terms
 * of the Common Development and Distribution License
 * (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/CDDLv1.0.html or
 * glassfish/bootstrap/legal/CDDLv1.0.txt.
 * See the License for the specific language governing
 * permissions and limitations under the License.
 *
 * When distributing Covered Code, include this CDDL
 * Header Notice in each file and include the License file
 * at glassfish/bootstrap/legal/CDDLv1.0.txt.
 * If applicable, add the following below the CDDL Header,
 * with the fields enclosed by brackets [] replaced by
 * you own identifying information:
 * "Portions Copyrighted [year] [name of copyright owner]"
 *
 * Copyright 2006 Sun Microsystems, Inc. All rights reserved.
 */
package com.sun.enterprise.jbi.serviceengine.comm;

import com.sun.enterprise.jbi.serviceengine.core.JavaEEServiceEngineContext;
import java.util.HashMap;
import java.util.logging.Level;

import javax.jbi.messaging.MessageExchange;
import javax.jbi.messaging.MessagingException;
import com.sun.enterprise.jbi.serviceengine.work.OneWork;

/**
 * Acceptor object that continuously receives messages from
 * NMR. There is one MessageAcceptor object per WorkManager.
 *
 * MessageAcceptor also keeps a house keeping datastructure
 * that holds information about all threads that are waiting
 * for reply from on NMR for their two-way message exchange.
 *
 * @authod Binod PG
 */
public class MessageAcceptor extends OneWork {
    
    private HashMap<String,MessageReceiver> receivers = new HashMap();
    private boolean released = false;
    
    /**
     * Start the acceptor thread. Note that execute() inturn call
     * doWork method, where bulk of logic is present.
     */
    public void startAccepting() {
        execute();
    }
    
    /**
     * Add a MessageReceiver object that waits for a reply from
     * from NMR on their 2-way message exchange.
     *
     * @param receiver MessageReceiver instance.
     */
    public void register(MessageReceiver receiver) {
        String id = receiver.getMessageExchange().getExchangeId();
        logger.log(Level.FINER, "Adding recever for " + id);
        synchronized (receivers) {
            receivers.put(id, receiver);
        }
    }
    
    /**
     * Release the thread from accepting. This method doesnt interrupt
     * the thread. It is just a soft release applicable only from the
     * next iteration of acceptor thread.
     */
    public void release() {
        released = true;
    }
    
    /**
     * Actual work happens in this method. DeliveryChannel available
     * from the super class. If there is any MessageReceiver waiting
     * for this message, then the MessageExchange is made avalable to that
     * MessageReceiver. If no MessageReceiver is interested in this MEP,
     * then a new MessageProcessor will process this message.
     * In the latter case, the message is for a 109 webservice deployed
     * in appserver.
     */
    public void doWork() {
        while (true) {
            try {
                
                MessageExchange me = getDeliveryChannel().accept();
                if (released) {
                    break;
                }
                
                if(me != null) {
                    String id = me.getExchangeId();
                    
                    // The full block is not synchronized since,
                    // 1. Id will always be unique
                    // 2. Service engine will register the receiver
                    //    before sending and hence we wont miss any.
                    if (receivers.containsKey(id)) {
                        synchronized(receivers) {
                            MessageReceiver receiver = receivers.remove(id);
                            receiver.setMessageExchange(me);
                            if (logger.isLoggable(Level.FINE)) {
                                logger.log(Level.FINE,
                                        "Releasing MessageReceiver:" + id + ",MEP :" + me);
                            }
                            receiver.release();
                        }
                    } else {
                        MessageProcessor processor =
                                JavaEEServiceEngineContext.getInstance().
                                getBridge().getMessageProcessor(me);
                        processor.setUseCurrentThread(false);
                        processor.setMessageExchange(me);
                        if (logger.isLoggable(Level.FINE)) {
                            logger.log(Level.FINE,
                                    "Spawning MessageProcessorfor MEP :" + me);
                        }
                        processor.process();
                    }
                }
            } catch (MessagingException ie) {
                //Someone has interrupted the acceptor. Gracefully comeout.
                logger.log(Level.INFO, "Stopping the acceptor thread");
                break;
            }
        }
    }
}

