Add an option to share connection and channel between Samplers.

src/main/com/zeroclue/jmeter/protocol/amqp/AMQPConsumer.java

@@ -291,8 +291,15 @@ public void testStarted(String arg0) {
     public void cleanup() {
 
         try {
+        	// TODO: Refactor closing mechanism (cleanup AMQPConsumers before closing connection)
             if (consumerTag != null) {
-               channel.basicCancel(consumerTag);
+            	// If "shareChannel" is enabled, channel may have been already 
+            	// closed by cleanup call from another Sampler
+            	if (channel.isOpen() ) {
+            		channel.basicCancel(consumerTag);
+            	} else {
+            		log.warn("channel.basicCancel not called because channel is already close");
+            	}
             }
         } catch(IOException e) {
             log.error("Couldn't safely cancel the sample " + consumerTag, e);

src/main/com/zeroclue/jmeter/protocol/amqp/AMQPSampler.java

@@ -11,6 +11,9 @@
 import org.apache.log.Logger;
 
 import com.rabbitmq.client.AMQP.BasicProperties;
+
+import com.zeroclue.jmeter.protocol.amqp.ChannelCache;
+
 import org.apache.commons.lang3.StringUtils;
 
 public abstract class AMQPSampler extends AbstractSampler implements ThreadListener {
@@ -42,6 +45,7 @@ public abstract class AMQPSampler extends AbstractSampler implements ThreadListe
     protected static final String HOST = "AMQPSampler.Host";
     protected static final String PORT = "AMQPSampler.Port";
     protected static final String SSL = "AMQPSampler.SSL";
+    protected static final String SHARE_CHANNEL = "AMQPSampler.ShareChannel";
     protected static final String USERNAME = "AMQPSampler.Username";
     protected static final String PASSWORD = "AMQPSampler.Password";
     private static final String TIMEOUT = "AMQPSampler.Timeout";
@@ -56,6 +60,7 @@ public abstract class AMQPSampler extends AbstractSampler implements ThreadListe
 
     private transient ConnectionFactory factory;
     private transient Connection connection;
+    private transient static final ChannelCache channelCache = new ChannelCache();
 
     protected AMQPSampler(){
         factory = new ConnectionFactory();
@@ -72,8 +77,27 @@ protected boolean initChannel() throws IOException, NoSuchAlgorithmException, Ke
         }
 
         if(channel == null) {
-            channel = createChannel();
-            setChannel(channel);
+
+        	// if channel sharing is enabled, look for channel in channelCache
+        	if(shareChannel()) {
+        		String cnxkey = ChannelCache.genKey(getVirtualHost(), getHost(), getPort(), getUsername(), getPassword(), getTimeout(), connectionSSL());
+        		channel = channelCache.get(cnxkey);
+
+        		// channel isn't in cache (occurs for first AMQPSampler with this key)
+        		if(channel == null) {
+            		channel = createChannel();
+            		log.info("Caching new channel for connection " + cnxkey);
+            		channelCache.set(cnxkey, channel);
+            		setChannel(channel);
+
+        		} else {
+        			log.info("Recycling channel for connection " + cnxkey);
+        			setChannel(channel);
+        		}
+        	} else { // create a new dedicated connection and channel for current AMQPSampler instance
+        		channel = createChannel();
+        		setChannel(channel);        		
+        	}
 
             //TODO: Break out queue binding
             boolean queueConfigured = (getQueue() != null && !getQueue().isEmpty());
@@ -289,6 +313,17 @@ public boolean connectionSSL() {
         return getPropertyAsBoolean(SSL);
     }
 
+    public void setShareChannel(String content) {
+        setProperty(SHARE_CHANNEL, content);
+    }
+
+    public void setShareChannel(Boolean value) {
+        setProperty(SHARE_CHANNEL, value.toString());
+    }
+
+    public boolean shareChannel() {
+        return getPropertyAsBoolean(SHARE_CHANNEL);
+    }
 
     public String getUsername() {
         return getPropertyAsString(USERNAME);
@@ -374,10 +409,16 @@ public void setQueueRedeclare(Boolean content) {
     }
 
     protected void cleanup() {
+
         try {
             //getChannel().close();   // closing the connection will close the channel if it's still open
-            if(connection != null && connection.isOpen())
-                connection.close();
+        	// When shareChannel is enabled, this will close connection
+        	// used by other samplers (their cleanup code won't complete normally)
+        	synchronized(getChannel()) {
+	       		log.info("closing connection: "+connection);
+	            if(connection != null && connection.isOpen())
+	                connection.close();
+        	}
         } catch (IOException e) {
             log.error("Failed to close connection", e);
         }

src/main/com/zeroclue/jmeter/protocol/amqp/ChannelCache.java

@@ -0,0 +1,62 @@
+package com.zeroclue.jmeter.protocol.amqp;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.jorphan.logging.LoggingManager;
+import org.apache.log.Logger;
+
+import com.rabbitmq.client.Channel;
+
+
+/**
+ * @author rbnxx
+ * ChannelCache caches Channel objects for a given thread (user) with the 
+ * same connection parameters (ie. if several AMQPSamplers are configured
+ * to use the same broker, they will share the same connection and channel). 
+ *
+ */
+class ChannelCache {
+
+	private static final Logger log = LoggingManager.getLoggerForClass();
+
+	private final ThreadLocal<Map<String,Channel>> cnxChannelMap = new ThreadLocal<Map<String,Channel>>(){
+		{
+			log.debug("initializing ChannelCache (global)");
+		}
+
+		@Override protected Map<String,Channel> initialValue() {
+			log.debug("initializing ChannelCache HashMap for thread");
+			return new HashMap<String,Channel>();
+		}
+
+	};
+
+	public static String genKey(String vhost, String host, String port, String user, String pass, String timeout, Boolean ssl) {
+		// generated as amqp uri (cf. https://www.rabbitmq.com/uri-query-parameters.html )
+		return new StringBuilder()
+				.append(ssl?"amqps://":"amqp://")
+				.append(user)
+				.append(":")
+				.append(pass)
+				.append("@host:")
+				.append(host)
+				.append(":")
+				.append(port)
+				.append("/")
+				.append(vhost)
+				.append("?connection_timeout=")
+				.append(timeout)
+				.toString();
+	}
+
+	public void set(String cnxString, Channel Channel) {
+		cnxChannelMap.get().put(cnxString, Channel);
+	}
+
+	public Channel get(String cnxString) {
+		return cnxChannelMap.get().get(cnxString);
+	}
+
+};
+

src/main/com/zeroclue/jmeter/protocol/amqp/gui/AMQPSamplerGui.java

@@ -43,6 +43,7 @@ public abstract class AMQPSamplerGui extends AbstractSamplerGui {
     protected JLabeledTextField username = new JLabeledTextField("Username");
     protected JLabeledTextField password = new JLabeledTextField("Password");
     private final JCheckBox SSL = new JCheckBox("SSL?", false);
+    private final JCheckBox shareChannel = new JCheckBox("Share?", false);
 
     private final JLabeledTextField iterations = new JLabeledTextField("Number of samples to Aggregate");
 
@@ -81,6 +82,7 @@ public void configure(TestElement element) {
         username.setText(sampler.getUsername());
         password.setText(sampler.getPassword());
         SSL.setSelected(sampler.connectionSSL());
+        shareChannel.setSelected(sampler.shareChannel());
         log.info("AMQPSamplerGui.configure() called");
     }
 
@@ -112,6 +114,7 @@ public void clearGui() {
         username.setText("guest");
         password.setText("guest");
         SSL.setSelected(false);
+        shareChannel.setSelected(false);
     }
 
     /**
@@ -145,6 +148,7 @@ public void modifyTestElement(TestElement element) {
         sampler.setUsername(username.getText());
         sampler.setPassword(password.getText());
         sampler.setConnectionSSL(SSL.isSelected());
+        sampler.setShareChannel(shareChannel.isSelected());
         log.info("AMQPSamplerGui.modifyTestElement() called, set user/pass to " + username.getText() + "/" + password.getText() + " on sampler " + sampler);
     }
 
@@ -264,6 +268,10 @@ private Component makeCommonPanel() {
         gridBagConstraints.gridy = 2;
         serverSettings.add(SSL, gridBagConstraints);
 
+        gridBagConstraints.gridx = 1;
+        gridBagConstraints.gridy = 3;
+        serverSettings.add(shareChannel, gridBagConstraints);
+
         gridBagConstraints.gridx = 0;
         gridBagConstraints.gridy = 3;
         serverSettings.add(username, gridBagConstraints);