Add documentation, basic unit test, createChannel(), close() and tryConsumeOne().

source/vibe/core/channel.d

@@ -1,27 +1,103 @@
 /** Implements a thread-safe, typed producer-consumer queue.
 
-	Copyright: © 2017 RejectedSoftware e.K.
+	Copyright: © 2017-2019 RejectedSoftware e.K.
 	Authors: Sönke Ludwig
 	License: Subject to the terms of the MIT license, as written in the included LICENSE.txt file.
 */
 module vibe.core.channel;
 
+import vibe.core.sync : TaskCondition;
+import vibe.internal.array : FixedRingBuffer;
+
+import std.algorithm.mutation : move, swap;
+import std.exception : enforce;
+import core.sync.mutex;
+
 // multiple producers allowed, multiple consumers allowed - Q: should this be restricted to allow higher performance? maybe configurable?
 // currently always buffered - TODO: implement blocking non-buffered mode
 // TODO: implement a multi-channel wait, e.g.
 // TaggedAlgebraic!(...) consumeAny(ch1, ch2, ch3); - requires a waitOnMultipleConditions function
-// TODO: implement close()
-// TODO: fully support non-copyable types using swap/move where appropriate
-// TODO: add unit tests
-// Q: Should this be exposed as a class, or as an RC struct?
 
-private final class Channel(T, size_t buffer_size = 100) {
+
+/** Creates a new channel suitable for cross-task and cross-thread communication.
+*/
+Channel!(T, buffer_size) createChannel(T, size_t buffer_size = 100)()
+{
+	Channel!(T, buffer_size) ret;
+	ret.m_impl = new shared ChannelImpl!(T, buffer_size);
+	return ret;
+}
+
+
+/** Thread-safe typed data channel implementation.
+
+	The implementation supports multiple-reader-multiple-writer operation across
+	multiple tasks in multiple threads.
+*/
+struct Channel(T, size_t buffer_size) {
+	enum bufferSize = buffer_size;
+
+	private shared ChannelImpl!(T, buffer_size) m_impl;
+
+	/** Determines whether there is more data to read.
+
+		This property is empty $(I iff) no more elements are in the internal
+		buffer and `close()` has been called. Once the channel is empty,
+		subsequent calls to `consumeOne` or `consumeAll` will throw an
+		exception.
+
+		Note that relying on the return value to determine whether another
+		element can be read is only safe in a single-reader scenario. Use
+		`tryConsumeOne` in a multiple-reader scenario instead.
+	*/
+	@property bool empty() { return m_impl.empty; }
+
+	/** Closes the channel.
+
+		A closed channel does not accept any new items enqueued using `put` and
+		causes `empty` to return `fals` as soon as all preceeding elements have
+		been consumed.
+	*/
+	void close() { m_impl.close(); }
+
+	/** Consumes a single element off the queue.
+
+		This function will block if no elements are available. If the `empty`
+		property is `true`, an exception will be thrown.
+	*/
+	T consumeOne() { return m_impl.consumeOne(); }
+
+	/** Attempts to consume a single element.
+
+		If no more elements are available and the channel has been closed,
+		a null value is returned.
+	*/
+	bool tryConsumeOne(ref T dst) { return m_impl.tryConsumeOne(dst); }
+
+	/** Attempts to consume all elements currently in the queue.
+
+		This function will block if no elements are available.  If the `empty`
+		property is `true`, an exception will be thrown.
+	*/
+	void consumeAll(ref FixedRingBuffer!(T, buffer_size) dst)
+	{ return m_impl.consumeAll(dst); }
+
+	/** Enqueues an element.
+	*/
+	void put(T item) { m_impl.put(item.move); }
+}
+
+
+private final class ChannelImpl(T, size_t buffer_size) {
 	import vibe.core.concurrency : isWeaklyIsolated;
 	static assert(isWeaklyIsolated!T, "Channel data type "~T.stringof~" is not safe to pass between threads.");
 
+	private {
 		Mutex m_mutex;
 		TaskCondition m_condition;
 		FixedRingBuffer!(T, buffer_size) m_items;
+		bool m_closed = false;
+	}
 
 	this()
 	shared {
@@ -29,50 +105,124 @@ private final class Channel(T, size_t buffer_size = 100) {
 		m_condition = cast(shared)new TaskCondition(cast(Mutex)m_mutex);
 	}
 
-	bool empty()
+	@property bool empty()
 	shared {
-		synchronized (m_mutex)
-			return (cast(Channel)this).m_items.empty;
+		synchronized (m_mutex) {
+			auto thisus = () @trusted { return cast(ChannelImpl)this; } ();
+			return thisus.m_closed && thisus.m_items.empty;
+		}
+	}
+
+	void close()
+	shared {
+		synchronized (m_mutex) {
+			auto thisus = () @trusted { return cast(ChannelImpl)this; } ();
+			thisus.m_closed = true;
+			thisus.m_condition.notifyAll();
+		}
+	}
+
+	bool tryConsumeOne(ref T dst)
+	shared {
+		auto thisus = () @trusted { return cast(ChannelImpl)this; } ();
+		bool was_full = false;
+
+		synchronized (m_mutex) {
+			while (thisus.m_items.empty) {
+				if (m_closed) return false;
+				thisus.m_condition.wait();
+			}
+			was_full = thisus.m_items.full;
+			move(thisus.m_items.front, dst);
+			thisus.m_items.popFront();
+		}
+
+		if (was_full) thisus.m_condition.notifyAll();
+
+		return true;
 	}
 
 	T consumeOne()
 	shared {
-		auto thisus = cast(Channel)this;
+		auto thisus = () @trusted { return cast(ChannelImpl)this; } ();
 		T ret;
 		bool was_full = false;
+
 		synchronized (m_mutex) {
-			while (thisus.m_items.empty)
+			while (thisus.m_items.empty) {
+				if (m_closed) throw new Exception("Attempt to consume from an empty channel.");
 				thisus.m_condition.wait();
-			was_full = thisus.m_items.full;
-			swap(thisus.m_items.front, ret);
 			}
+			was_full = thisus.m_items.full;
+			move(thisus.m_items.front, ret);
+			thisus.m_items.popFront();
+		}
+
 		if (was_full) thisus.m_condition.notifyAll();
+
 		return ret.move;
 	}
 
 	void consumeAll(ref FixedRingBuffer!(T, buffer_size) dst)
 	shared {
-		auto thisus = cast(Channel)this;
+		auto thisus = () @trusted { return cast(ChannelImpl)this; } ();
 		bool was_full = false;
+
 		synchronized (m_mutex) {
-			while (thisus.m_items.empty)
+			while (thisus.m_items.empty) {
+				if (m_closed) throw new Exception("Attempt to consume from an empty channel.");
 				thisus.m_condition.wait();
+			}
+
 			was_full = thisus.m_items.full;
 			swap(thisus.m_items, dst);
 		}
+
 		if (was_full) thisus.m_condition.notifyAll();
 	}
 
 	void put(T item)
 	shared {
-		auto thisus = cast(Channel)this;
+		auto thisus = () @trusted { return cast(ChannelImpl)this; } ();
 		bool need_notify = false;
+
 		synchronized (m_mutex) {
+			enforce(!m_closed, "Sending on closed channel.");
 			while (thisus.m_items.full)
 				thisus.m_condition.wait();
 			need_notify = thisus.m_items.empty;
 			thisus.m_items.put(item.move);
 		}
+
 		if (need_notify) thisus.m_condition.notifyAll();
 	}
 }
+
+unittest { // test basic operation and non-copyable struct compatiblity
+	static struct S {
+		int i;
+		@disable this(this);
+	}
+
+	auto ch = createChannel!S;
+	ch.put(S(1));
+	assert(ch.consumeOne().i == 1);
+	ch.put(S(4));
+	ch.put(S(5));
+	{
+		FixedRingBuffer!(S, 100) buf;
+		ch.consumeAll(buf);
+		assert(buf.length == 2);
+		assert(buf[0].i == 4);
+		assert(buf[1].i == 5);
+	}
+	ch.put(S(2));
+	assert(!ch.empty);
+	ch.close();
+	assert(!ch.empty);
+	S v;
+	assert(ch.tryConsumeOne(v));
+	assert(v.i == 2);
+	assert(ch.empty);
+	assert(!ch.tryConsumeOne(v));
+}