/*
 * 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.client;

import java.util.concurrent.ConcurrentLinkedQueue;

import javax.jms.Connection;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.naming.Context;
import javax.naming.NamingException;

import org.apache.jmeter.protocol.jms.Utils;
import org.apache.jorphan.logging.LoggingManager;
import org.apache.log.Logger;

/**
 * Receives messages in a separate thread until told to stop.
 * Run loop permanently receives messages; the sampler calls reset()
 * when it has taken enough messages.
 *
 */
/*
 * TODO Needs rework - there is a window between receiving a message and calling reset()
 * which means that a message can be lost. It's not clear why a separate thread is needed,
 * given that the sampler loops until enough samples have been received.
 * Also, messages are received in wait mode, so the RUN flag won't be checked until
 * at least one more message has been received.
*/
public class ReceiveSubscriber implements Runnable {

    private static final Logger log = LoggingManager.getLoggerForClass();

    private final Connection CONN;

    private final Session SESSION;

    private final MessageConsumer SUBSCRIBER;

    //@GuardedBy("this")
    private int counter;

    @SuppressWarnings("unused")
    private int loop = 1; // TODO never read
    
    private final ConcurrentLinkedQueue<TextMessage> queue = new ConcurrentLinkedQueue<TextMessage>();

    //@GuardedBy("this")
    private volatile boolean RUN = true;
    // Needs to be volatile to ensure value is picked up

    //@GuardedBy("this")
    private Thread CLIENTTHREAD;

    public ReceiveSubscriber(boolean useProps, String jndi, String url, String connfactory, String topic,
            boolean useAuth, String user, String pwd) throws NamingException, JMSException {
        Context ctx = InitialContextFactory.getContext(useProps, jndi, url, useAuth, user, pwd);
        CONN = Utils.getConnection(ctx, connfactory);
        Destination _topic = Utils.lookupDestination(ctx, topic);
        SESSION = CONN.createSession(false, Session.AUTO_ACKNOWLEDGE);
        SUBSCRIBER = SESSION.createConsumer(_topic);
    }

    /**
     * Set the number of iterations for each call to sample()
     *
     * @param loop
     */
    public void setLoop(int loop) {
        this.loop = loop;
    }

    /**
     * Resume will call Connection.start() and begin receiving messages from the
     * JMS provider.
     */
    public void resume() {
        if (this.CONN == null) {
            log.error("Connection not set up");
            return;
        }
        try {
            this.CONN.start();
        } catch (JMSException e) {
            log.error("failed to start recieving");
        }
    }

    /**
     * Get the message
     * @return the next message from the queue or null if none
     */
    public synchronized TextMessage getMessage() {
        TextMessage msg = queue.poll();
        this.counter--;
        return msg;
    }

    /**
     * close() will stop the connection first. Then it closes the subscriber,
     * session and connection.
     */
    public synchronized void close() { // called from testEnded() thread
        this.RUN = false;
        try {
            this.CONN.stop();
            Utils.close(SUBSCRIBER, log);
            Utils.close(SESSION, log);
            Utils.close(CONN, log);
            this.CLIENTTHREAD.interrupt();
            this.CLIENTTHREAD = null;
            queue.clear();
        } catch (JMSException e) {
            log.error(e.getMessage());
        } catch (Exception e) {
            log.error(e.getMessage());
        }
    }

    /**
     * Increment the count and return the new value
     *
     * @param increment
     */
    public synchronized int count(int increment) {
        counter += increment;
        return counter;
    }

    /**
     * start will create a new thread and pass this class. once the thread is
     * created, it calls Thread.start().
     */
    public void start() {
        // No point starting thread unless subscriber exists
        if (SUBSCRIBER == null) {
            log.error("Subscriber has not been set up");
            return;
        }
        this.CLIENTTHREAD = new Thread(this, "Subscriber2");
        this.CLIENTTHREAD.start();
    }

    /**
     * run calls listen to begin listening for inbound messages from the
     * provider.
     * 
     * Updates the count field so the caller can check how many messages have been receieved.
     * 
     */
    public void run() {
        if (SUBSCRIBER == null) { // just in case
            log.error("Subscriber has not been set up");
            return;
        }
        while (RUN) {
            try {
                Message message = this.SUBSCRIBER.receive();
                if (message != null && message instanceof TextMessage) {
                    queue.add((TextMessage)message);
                    count(1);
                }
            } catch (JMSException e) {
                log.error("Communication error: " + e.getMessage());
            }
        }
    }
}

@@ -2170,90 +2170,8 @@
 r;%0A%0A
-    @SuppressWarnings(%22unused%22)%0A    private int loop = 1; // TODO never read%0A    %0A
     
@@ -2274,33 +2274,8 @@
 );%0A%0A
-    //@GuardedBy(%22this%22)%0A
     
@@ -2527,21 +2527,31 @@
  String 
-topic
+destinationName
 ,%0A      
@@ -2792,74 +2792,8 @@
 y);%0A
-        Destination _topic = Utils.lookupDestination(ctx, topic);%0A
     
@@ -2871,224 +2871,123 @@
     
-SUBSCRIBER = SESSION.createConsumer(_topic);%0A    %7D%0A%0A    /**%0A     * Set the number of iterations for each call to sample()
+Destination dest = Utils.lookupDestination(ctx, destinationName);
 %0A    
- *%0A
     
- * @param loop%0A     */%0A    public void setLoop(int loop) %7B%0A        this.loop = loop
+SUBSCRIBER = SESSION.createConsumer(dest)
 ;%0A  
@@ -4671,190 +4671,8 @@
 ) %7B%0A
-        // No point starting thread unless subscriber exists%0A        if (SUBSCRIBER == null) %7B%0A            log.error(%22Subscriber has not been set up%22);%0A            return;%0A        %7D%0A
     
@@ -5014,145 +5014,8 @@
 ) %7B%0A
-        if (SUBSCRIBER == null) %7B // just in case%0A            log.error(%22Subscriber has not been set up%22);%0A            return;%0A        %7D%0A
     
@@ -5143,27 +5143,8 @@
 age 
-!= null && message 
 inst
@@ -5248,16 +5248,129 @@
 unt(1);%0A
+                %7D else if (message != null)%7B%0A                %09log.warn(%22Discarded non TextMessage %22 +  message);%0A
         

Tidy up; drop unused loop count