added UnboundedSpScPipe

src/main/java/teetime/framework/pipe/SpScPipe.java

@@ -19,8 +19,6 @@ import java.util.Queue;
 
 import org.jctools.queues.QueueFactory;
 import org.jctools.queues.spec.ConcurrentQueueSpec;
-import org.jctools.queues.spec.Ordering;
-import org.jctools.queues.spec.Preference;
 
 import teetime.framework.AbstractInterThreadPipe;
 import teetime.framework.InputPort;
@@ -36,7 +34,7 @@ public final class SpScPipe extends AbstractInterThreadPipe {
 
 	<T> SpScPipe(final OutputPort<? extends T> sourcePort, final InputPort<T> targetPort, final int capacity) {
 		super(sourcePort, targetPort);
-		this.queue = QueueFactory.newQueue(new ConcurrentQueueSpec(1, 1, capacity, Ordering.FIFO, Preference.THROUGHPUT));
+		this.queue = QueueFactory.newQueue(ConcurrentQueueSpec.createBoundedSpsc(capacity));
 	}
 
 	@Deprecated

src/main/java/teetime/framework/pipe/UnboundedSpScPipe.java

+/**
+ * Copyright (C) 2015 TeeTime (http://teetime.sourceforge.net)
+ *
+ * Licensed 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 teetime.framework.pipe;
+
+import java.util.Queue;
+
+import org.jctools.queues.QueueFactory;
+import org.jctools.queues.spec.ConcurrentQueueSpec;
+import org.jctools.queues.spec.Ordering;
+import org.jctools.queues.spec.Preference;
+
+import teetime.framework.AbstractInterThreadPipe;
+import teetime.framework.InputPort;
+import teetime.framework.OutputPort;
+
+public final class UnboundedSpScPipe extends AbstractInterThreadPipe {
+
+	private final Queue<Object> queue;
+	// statistics
+	private int numWaits;
+
+	<T> UnboundedSpScPipe(final OutputPort<? extends T> sourcePort, final InputPort<T> targetPort) {
+		super(sourcePort, targetPort);
+		ConcurrentQueueSpec specification = new ConcurrentQueueSpec(1, 1, 0, Ordering.FIFO, Preference.THROUGHPUT);
+		this.queue = QueueFactory.newQueue(specification);
+	}
+
+	@Override
+	public boolean add(final Object element) {
+		// BETTER introduce a QueueIsFullStrategy
+		while (!this.queue.offer(element)) {
+			this.numWaits++;
+			Thread.yield();
+		}
+		// this.reportNewElement();
+		return true;
+	}
+
+	@Override
+	public Object removeLast() {
+		return this.queue.poll();
+	}
+
+	@Override
+	public boolean isEmpty() {
+		return this.queue.isEmpty();
+	}
+
+	@Override
+	public int size() {
+		return this.queue.size();
+	}
+
+	@Override
+	public Object readLast() {
+		return this.queue.peek();
+	}
+
+	// BETTER find a solution w/o any thread-safe code in this stage
+	public synchronized int getNumWaits() {
+		return this.numWaits;
+	}
+
+}

src/main/java/teetime/framework/pipe/UnboundedSpScPipeFactory.java

+/**
+ * Copyright (C) 2015 TeeTime (http://teetime.sourceforge.net)
+ *
+ * Licensed 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 teetime.framework.pipe;
+
+import teetime.framework.InputPort;
+import teetime.framework.OutputPort;
+import teetime.framework.pipe.PipeFactoryRegistry.PipeOrdering;
+import teetime.framework.pipe.PipeFactoryRegistry.ThreadCommunication;
+
+public class UnboundedSpScPipeFactory implements IPipeFactory {
+
+	@Override
+	public <T> IPipe create(final OutputPort<? extends T> sourcePort, final InputPort<T> targetPort) {
+		return this.create(sourcePort, targetPort, 0);
+	}
+
+	/**
+	 * {@inheritDoc}
+	 *
+	 * The capacity is ignored.
+	 */
+	@Override
+	public <T> IPipe create(final OutputPort<? extends T> sourcePort, final InputPort<T> targetPort, final int capacity) {
+		return new UnboundedSpScPipe(sourcePort, targetPort);
+	}
+
+	@Override
+	public ThreadCommunication getThreadCommunication() {
+		return ThreadCommunication.INTER;
+	}
+
+	@Override
+	public PipeOrdering getOrdering() {
+		return PipeOrdering.QUEUE_BASED;
+	}
+
+	@Override
+	public boolean isGrowable() {
+		return true;
+	}
+
+}

src/main/resources/pipe-factories.conf

 teetime.framework.pipe.SingleElementPipeFactory
 teetime.framework.pipe.OrderedGrowableArrayPipeFactory
 teetime.framework.pipe.UnorderedGrowablePipeFactory
-teetime.framework.pipe.SpScPipeFactory
\ No newline at end of file
+teetime.framework.pipe.SpScPipeFactory
+teetime.framework.pipe.UnboundedSpScPipeFactory