/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 */

package org.apache.jmeter.protocol.jms.sampler;

import org.apache.jorphan.logging.LoggingManager;
import org.apache.log.Logger;

import javax.jms.*;

/**
 * Receiver of pseudo-synchronous reply messages.
 *
 */
public class Receiver implements Runnable {
    private static final Logger log = LoggingManager.getLoggerForClass();

    private volatile boolean active;

    private final QueueSession session;

    private final QueueReceiver consumer;

    private final QueueConnection conn;

    private final boolean useReqMsgIdAsCorrelId;

    private Receiver(QueueConnectionFactory factory, Queue receiveQueue, String principal, String credentials, boolean useReqMsgIdAsCorrelId) throws JMSException {
        if (null != principal && null != credentials) {
            log.info("creating receiver WITH authorisation credentials");
            conn = factory.createQueueConnection(principal, credentials);
        }else{
            log.info("creating receiver without authorisation credentials");
            conn = factory.createQueueConnection(); 
        }
        session = conn.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
        consumer = session.createReceiver(receiveQueue);
        this.useReqMsgIdAsCorrelId = useReqMsgIdAsCorrelId;
        log.debug("Receiver - ctor. Starting connection now");
        conn.start();
        log.info("Receiver - ctor. Connection to messaging system established");
    }

    public static Receiver createReceiver(QueueConnectionFactory factory, Queue receiveQueue,
            String principal, String credentials, boolean useReqMsgIdAsCorrelId)
            throws JMSException {
        Receiver receiver = new Receiver(factory, receiveQueue, principal, credentials, useReqMsgIdAsCorrelId);
        Thread thread = new Thread(receiver);
        thread.start();
        return receiver;
    }

    public void run() {
        active = true;
        Message reply;

        while (active) {
            reply = null;
            try {
                reply = consumer.receive(5000);
                if (reply != null) {
                    String messageKey;
                    final MessageAdmin admin = MessageAdmin.getAdmin();
                    if (useReqMsgIdAsCorrelId){
                        messageKey = reply.getJMSMessageID();
                        synchronized (admin) {// synchronize with FixedQueueExecutor
                            admin.putReply(messageKey, reply);                            
                        }
                    } else {
                        messageKey = reply.getJMSCorrelationID();
                        if (messageKey == null) {// JMSMessageID cannot be null
                            log.warn("Received message with correlation id null. Discarding message ...");
                        } else {
                            admin.putReply(messageKey, reply);
                        }
                    }
                }

            } catch (JMSException e1) {
                log.error("Error handling receive",e1);
            }
        }
        // not active anymore
        if (consumer != null) {
            try {
                consumer.close();
            } catch (JMSException e) {
                log.error("Error closing connection",e);
            }
        }

        // session and conn cannot be null (or ctor would have caused NPE)
        try {
            session.close();
        } catch (JMSException e) {
            log.error("Error closing session",e);
        }
        try {
            conn.close();
        } catch (JMSException e) {
            log.error("Error closing connection",e);
        }
    }

    public void deactivate() {
        active = false;
    }

}

@@ -1314,33 +1314,33 @@
 al boolean useRe
-q
+s
 MsgIdAsCorrelId;
@@ -1457,33 +1457,33 @@
 s, boolean useRe
-q
+s
 MsgIdAsCorrelId)
@@ -1623,33 +1623,69 @@
 tion credentials
-%22
+. UseResMsgId=%22+useResMsgIdAsCorrelId
 );%0A            c
@@ -1833,17 +1833,53 @@
 dentials
-%22
+. UseResMsgId=%22+useResMsgIdAsCorrelId
 );%0A     
@@ -2083,25 +2083,25 @@
   this.useRe
-q
+s
 MsgIdAsCorre
@@ -2107,25 +2107,25 @@
 elId = useRe
-q
+s
 MsgIdAsCorre
@@ -2294,32 +2294,397 @@
 ished%22);%0A    %7D%0A%0A
+    /**%0A     * Create a receiver to process responses.%0A     * %0A     * @param factory%0A     * @param receiveQueue%0A     * @param principal%0A     * @param credentials%0A     * @param useResMsgIdAsCorrelId true if should use JMSMessageId, false if should use JMSCorrelationId%0A     * @return the Receiver which will process the responses%0A     * @throws JMSException%0A     */%0A
     public stati
@@ -2820,25 +2820,25 @@
 oolean useRe
-q
+s
 MsgIdAsCorre
@@ -2965,25 +2965,25 @@
 tials, useRe
-q
+s
 MsgIdAsCorre
@@ -3455,17 +3455,17 @@
 f (useRe
-q
+s
 MsgIdAsC

Rename useReq to useRes as it is the response message fields that are used