CAMEL-6588 Choose BlockingQueue implementation in Seda component

camel-core/src/main/java/org/apache/camel/component/seda/ArrayBlockingQueueFactory.java

@@ -0,0 +1,73 @@
+/**
+ * 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.camel.component.seda;
+
+
+import java.util.concurrent.ArrayBlockingQueue;
+
+/**
+ * Implementation of {@link BlockingQueueFactory} producing {@link java.util.concurrent.ArrayBlockingQueue}
+ */
+public class ArrayBlockingQueueFactory<E> implements BlockingQueueFactory<E> {
+	/**
+	 * Capacity used when none provided
+	 */
+	private int defaultCapacity=50;
+	/**
+	 * Lock fairness. null means default fairness
+	 */
+	private Boolean fair;
+	/**
+	 * @return Default array capacity
+	 */
+	public int getDefaultCapacity() {
+		return defaultCapacity;
+	}
+
+	/**
+	 * @param defaultCapacity Default array capacity
+	 */
+	public void setDefaultCapacity(int defaultCapacity) {
+		this.defaultCapacity = defaultCapacity;
+	}
+
+	/**
+	 * @return Lock fairness
+	 */
+	public boolean isFair() {
+		return fair;
+	}
+
+	/**
+	 * @param fair Lock fairness
+	 */
+	public void setFair(boolean fair) {
+		this.fair = fair;
+	}
+
+    @Override
+    public ArrayBlockingQueue<E> create() {
+        return create(defaultCapacity);
+    }
+
+    @Override
+    public ArrayBlockingQueue<E> create(int capacity) {
+        return fair == null ? 
+				new ArrayBlockingQueue<E>(defaultCapacity) :
+				new ArrayBlockingQueue<E>(defaultCapacity, fair) ;
+    }
+}

camel-core/src/main/java/org/apache/camel/component/seda/BlockingQueueFactory.java

@@ -0,0 +1,38 @@
+/**
+ * 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.camel.component.seda;
+
+
+import java.util.concurrent.BlockingQueue;
+import org.apache.camel.Exchange;
+
+/**
+ * Factory of {@link java.util.concurrent.BlockingQueue}
+ * @param <E> Element type, usually {@link Exchange}
+ */
+public interface BlockingQueueFactory<E> {
+    /**
+     * Create a new {@link java.util.concurrent.BlockingQueue} with default capacity
+     * @return New {@link java.util.concurrent.BlockingQueue}
+     */
+    BlockingQueue<E> create();
+    /**
+     * Create a new {@link java.util.concurrent.BlockingQueue} with given capacity
+     * @return New {@link java.util.concurrent.BlockingQueue}
+     */
+    BlockingQueue<E> create(int capacity);
+}

camel-core/src/main/java/org/apache/camel/component/seda/LinkedBlockingQueueFactory.java

@@ -0,0 +1,35 @@
+/**
+ * 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.camel.component.seda;
+
+
+import java.util.concurrent.LinkedBlockingQueue;
+
+/**
+ * Implementation of {@link BlockingQueueFactory} producing {@link java.util.concurrent.LinkedBlockingQueue}
+ */
+public class LinkedBlockingQueueFactory<E> implements BlockingQueueFactory<E> {
+    @Override
+    public LinkedBlockingQueue<E> create() {
+        return new LinkedBlockingQueue<E>();
+    }
+
+    @Override
+    public LinkedBlockingQueue<E> create(int capacity) {
+        return new LinkedBlockingQueue<E>(capacity);
+    }
+}

camel-core/src/main/java/org/apache/camel/component/seda/PriorityBlockingQueueFactory.java

@@ -0,0 +1,54 @@
+/**
+ * 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.camel.component.seda;
+
+import java.util.Comparator;
+
+import java.util.concurrent.PriorityBlockingQueue;
+
+/**
+ * Implementation of {@link BlockingQueueFactory} producing {@link java.util.concurrent.PriorityBlockingQueue}
+ */
+public class PriorityBlockingQueueFactory<E> implements BlockingQueueFactory<E> {
+	/**
+	 * Comparator used to sort exchanges
+	 */
+	private Comparator<E> comparator;
+
+	public Comparator<E> getComparator() {
+		return comparator;
+	}
+
+	public void setComparator(Comparator<E> comparator) {
+		this.comparator = comparator;
+	}
+
+    @Override
+    public PriorityBlockingQueue<E> create() {
+        return comparator==null ?
+				new PriorityBlockingQueue<E>() :
+				// PriorityQueue as a default capacity of 11
+				new PriorityBlockingQueue<E>(11, comparator);
+    }
+
+    @Override
+    public PriorityBlockingQueue<E> create(int capacity) {
+        return comparator==null?
+				new PriorityBlockingQueue<E>(capacity):
+				new PriorityBlockingQueue<E>(capacity, comparator);
+    }
+}

camel-core/src/main/java/org/apache/camel/component/seda/SedaComponent.java

@@ -19,7 +19,6 @@
 import java.util.HashMap;
 import java.util.Map;
 import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.LinkedBlockingQueue;
 
 import org.apache.camel.Endpoint;
 import org.apache.camel.Exchange;
@@ -39,7 +38,7 @@ public class SedaComponent extends UriEndpointComponent {
     protected int queueSize;
     protected int defaultConcurrentConsumers = 1;
     private final Map<String, QueueReference> queues = new HashMap<String, QueueReference>();
-
+    private BlockingQueueFactory<Exchange> defaultQueueFactory =new LinkedBlockingQueueFactory<Exchange>();
     public SedaComponent() {
         super(SedaEndpoint.class);
     }
@@ -60,15 +59,30 @@ public int getConcurrentConsumers() {
         return defaultConcurrentConsumers;
     }
 
+	public BlockingQueueFactory<Exchange> getDefaultQueueFactory() {
+		return defaultQueueFactory;
+	}
+
+	public void setDefaultQueueFactory(BlockingQueueFactory<Exchange> defaultQueueFactory) {
+		this.defaultQueueFactory = defaultQueueFactory;
+	}
+
     /**
-     * @deprecated use {@link #getOrCreateQueue(String, Integer, Boolean)}
+     * @deprecated use {@link #getOrCreateQueue(String, Integer, Boolean, BlockingQueueFactory)}
      */
     @Deprecated
     public synchronized QueueReference getOrCreateQueue(String uri, Integer size) {
         return getOrCreateQueue(uri, size, null);
     }
 
+    /**
+     * @deprecated use {@link #getOrCreateQueue(String, Integer, Boolean, BlockingQueueFactory)}
+     */
     public synchronized QueueReference getOrCreateQueue(String uri, Integer size, Boolean multipleConsumers) {
+        return getOrCreateQueue(uri, size, multipleConsumers, null);
+    }
+
+    public synchronized QueueReference getOrCreateQueue(String uri, Integer size, Boolean multipleConsumers, BlockingQueueFactory customQueueFactory) {
         String key = getQueueKey(uri);
 
         QueueReference ref = getQueues().get(key);
@@ -91,14 +105,15 @@ public synchronized QueueReference getOrCreateQueue(String uri, Integer size, Bo
 
         // create queue
         BlockingQueue<Exchange> queue;
+        BlockingQueueFactory<Exchange> queueFactory = customQueueFactory == null ? defaultQueueFactory : customQueueFactory;
         if (size != null && size > 0) {
-            queue = new LinkedBlockingQueue<Exchange>(size);
+            queue = queueFactory.create(size);
         } else {
             if (getQueueSize() > 0) {
                 size = getQueueSize();
-                queue = new LinkedBlockingQueue<Exchange>(getQueueSize());
+                queue = queueFactory.create(getQueueSize());
             } else {
-                queue = new LinkedBlockingQueue<Exchange>();
+                queue = queueFactory.create();
             }
         }
         log.debug("Created queue {} with size {}", key, size);
@@ -127,8 +142,17 @@ protected Endpoint createEndpoint(String uri, String remaining, Map<String, Obje
             throw new IllegalArgumentException("The limitConcurrentConsumers flag in set to true. ConcurrentConsumers cannot be set at a value greater than "
                     + maxConcurrentConsumers + " was " + consumers);
         }
-        // defer creating queue till endpoint is started, so we pass in null
-        SedaEndpoint answer = new SedaEndpoint(uri, this, null, consumers);
+		// Resolve queue reference
+		BlockingQueue<Exchange> queue=resolveAndRemoveReferenceParameter(parameters, "queue", BlockingQueue.class);
+		SedaEndpoint answer;
+		// Resolve queue factory when no queue specified
+		if (queue == null) {
+			BlockingQueueFactory<Exchange> queueFactory=resolveAndRemoveReferenceParameter(parameters, "queueFactory", BlockingQueueFactory.class);
+			// defer creating queue till endpoint is started, so we pass the queue factory
+			answer = new SedaEndpoint(uri, this, queueFactory, consumers);			
+		} else {
+			answer = new SedaEndpoint(uri, this, queue, consumers);
+		}
         answer.configureProperties(parameters);
         return answer;
     }

camel-core/src/main/java/org/apache/camel/component/seda/SedaEndpoint.java

@@ -23,7 +23,6 @@
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.CopyOnWriteArraySet;
 import java.util.concurrent.ExecutorService;
-import java.util.concurrent.LinkedBlockingQueue;
 
 import org.apache.camel.Component;
 import org.apache.camel.Consumer;
@@ -79,20 +78,32 @@ public class SedaEndpoint extends DefaultEndpoint implements BrowsableEndpoint,
     private int pollTimeout = 1000;
     @UriParam
     private boolean purgeWhenStopping;
+    private BlockingQueueFactory<Exchange> queueFactory;
 
     public SedaEndpoint() {
+		queueFactory = new LinkedBlockingQueueFactory<Exchange>();
     }
 
     public SedaEndpoint(String endpointUri, Component component, BlockingQueue<Exchange> queue) {
         this(endpointUri, component, queue, 1);
     }
 
     public SedaEndpoint(String endpointUri, Component component, BlockingQueue<Exchange> queue, int concurrentConsumers) {
-        super(endpointUri, component);
+		this(endpointUri, component, concurrentConsumers);
         this.queue = queue;
         if (queue != null) {
             this.size = queue.remainingCapacity();
         }
+		queueFactory = new LinkedBlockingQueueFactory<Exchange>();
+	}
+
+    public SedaEndpoint(String endpointUri, Component component, BlockingQueueFactory<Exchange> queueFactory, int concurrentConsumers) {
+		this(endpointUri, component, concurrentConsumers);
+		this.queueFactory = queueFactory;
+	}
+
+    private SedaEndpoint(String endpointUri, Component component, int concurrentConsumers) {
+        super(endpointUri, component);
         this.concurrentConsumers = concurrentConsumers;
     }
 
@@ -130,7 +141,7 @@ public synchronized BlockingQueue<Exchange> getQueue() {
             if (getComponent() != null) {
                 // use null to indicate default size (= use what the existing queue has been configured with)
                 Integer size = getSize() == Integer.MAX_VALUE ? null : getSize();
-                SedaComponent.QueueReference ref = getComponent().getOrCreateQueue(getEndpointUri(), size, isMultipleConsumers());
+                SedaComponent.QueueReference ref = getComponent().getOrCreateQueue(getEndpointUri(), size, isMultipleConsumers(), queueFactory);
                 queue = ref.getQueue();
                 String key = getComponent().getQueueKey(getEndpointUri());
                 LOG.info("Endpoint {} is using shared queue: {} with size: {}", new Object[]{this, key, ref.getSize() !=  null ? ref.getSize() : Integer.MAX_VALUE});
@@ -149,9 +160,9 @@ public synchronized BlockingQueue<Exchange> getQueue() {
 
     protected BlockingQueue<Exchange> createQueue() {
         if (size > 0) {
-            return new LinkedBlockingQueue<Exchange>(size);
+            return queueFactory.create(size);
         } else {
-            return new LinkedBlockingQueue<Exchange>();
+            return queueFactory.create();
         }
     }