working on simple adapt pipeline/conveyor nodes

source/async.cpp

@@ -5,6 +5,32 @@
 namespace gin {
 namespace {
 thread_local EventLoop *local_loop = nullptr;
+
+EventLoop &currentEventLoop() {
+	EventLoop *loop = local_loop;
+	assert(loop);
+	return *loop;
+}
+} // namespace
+
+ConveyorNode::ConveyorNode() : child{nullptr}, parent{nullptr} {}
+
+ConveyorNode::ConveyorNode(Own<ConveyorNode> &&node)
+	: child{std::move(node)}, parent{nullptr} {}
+
+void ConveyorNode::setParent(ConveyorNode *p) { parent = p; }
+
+PropagateError::Helper::Helper(Error &&error) : error{std::move(error)} {}
+
+Error PropagateError::Helper::asError() { return std::move(error); }
+
+PropagateError::Helper PropagateError::operator()(const Error &error) const {
+	Error err{error};
+	return PropagateError::Helper{std::move(err)};
+}
+
+PropagateError::Helper PropagateError::operator()(Error &&error) {
+	return PropagateError::Helper{std::move(error)};
 }
 
 Event::Event(EventLoop &loop) : loop{loop} {}
@@ -146,4 +172,6 @@ void WaitScope::wait(const std::chrono::steady_clock::time_point &time_point) {
 }
 
 void WaitScope::poll() { loop.poll(); }
+
+class YieldConveyorNode final : public ConveyorNode {};
 } // namespace gin

source/async.h

@@ -1,31 +1,43 @@
 #pragma once
 
 #include <list>
+#include <queue>
 
 #include "common.h"
+#include "error.h"
 #include "timer.h"
 
 namespace gin {
 class ConveyorNode {
 private:
-	Own<ConveyorNode> node;
-	ConveyorNode* parent = nullptr;
+	Own<ConveyorNode> child;
+	ConveyorNode *parent = nullptr;
+
 public:
 	ConveyorNode();
-	ConveyorNode(Own<ConveyorNode>&& node);
+	ConveyorNode(Own<ConveyorNode> &&child);
 	virtual ~ConveyorNode() = default;
 
-	void setParent(ConveyorNode* p);
+	void setParent(ConveyorNode *p);
 };
 
 class ConveyorBase {
 private:
 	Own<ConveyorNode> node;
+
 public:
-	ConveyorBase(Own<ConveyorNode>&& node);
+	ConveyorBase(Own<ConveyorNode> &&node);
 	virtual ~ConveyorBase() = default;
 };
 
+class ConveyorStorage {
+public:
+	virtual ~ConveyorStorage() = default;
+
+	virtual size_t space() const = 0;
+	virtual size_t queued() const = 0;
+};
+
 template <typename T> class Conveyor;
 
 template <typename T> Conveyor<T> chainedConveyorType(T *);
@@ -40,44 +52,45 @@ using ConveyorResult = ChainedConveyors<ReturnType<Func, T>>;
 
 struct PropagateError {
 public:
-	struct PropagateErrorHelper {
+	struct Helper {
+	private:
 		Error error;
 
-		PropagateError asError();
+	public:
+		Helper(Error &&error);
+
+		Error asError();
 	};
 
-	PropagateErrorHelper operator(const Error& error) const;
-	PropagateErrorHelper operator(Error&& error);
+	PropagateError::Helper operator()(const Error &error) const;
+	PropagateError::Helper operator()(Error &&error);
 };
 
 template <typename T> class Conveyor : public ConveyorBase {
 private:
 public:
-	template <typename Func, typename ErrorFunc>
-	ConveyorResult<Func, T> then(Func &&func, ErrorFunc &&error_func);
+	template <typename Func, typename ErrorFunc = PropagateError>
+	ConveyorResult<Func, T> then(Func &&func,
+								 ErrorFunc &&error_func = PropagateError());
 };
 
-template<typename T>
-class ConveyorFeeder {
-private:
-	Conveyor<T>* entry = nullptr;
-
-	friend class Conveyor<T>;
-
-	void entryDestroyed();
+template <typename T> class ConveyorFeeder {
 public:
-	ConveyorFeeder(Conveyor<T>& conv);
-	void feed(T&& data);
+	virtual ~ConveyorFeeder() = default;
+
+	virtual void feed(T &&data) = 0;
+	virtual void fail(Error &&error) = 0;
+
+	virtual size_t space() const = 0;
+	virtual size_t queued() const = 0;
 };
 
-template<typename T>
-struct ConveyorAndFeeder {
+template <typename T> struct ConveyorAndFeeder {
 	Own<ConveyorFeeder<T>> feeder;
 	Conveyor<T> conveyor;
 };
 
-template<typename T>
-ConveyorFeeder<T> newConveyorAndFeeder();
+template <typename T> ConveyorFeeder<T> newConveyorAndFeeder();
 
 class EventLoop;
 class Event {
@@ -139,23 +152,94 @@ public:
 	void poll();
 };
 
+// Secret stuff
+// Aka private semi hidden classes
+
+template <typename T> class AdaptConveyorNode;
+
+template <typename T> class AdaptConveyorFeeder : public ConveyorFeeder<T> {
+private:
+	AdaptConveyorNode<T> *feedee = nullptr;
+
+	std::queue<T> storage;
+
+public:
+	~AdaptConveyorFeeder();
+
+	void setFeedee(AdaptConveyorNode<T> *feedee);
+
+	void feed(T &&value) override;
+	void fail(Error &&error) override;
+
+	size_t space() const override;
+	size_t queued() const override;
+};
+
+template <typename T>
+class AdaptConveyorNode : public ConveyorNode, public ConveyorStorage {
+private:
+	AdaptConveyorFeeder<T> *feeder = nullptr;
+
+public:
+	~AdaptConveyorNode();
+
+	void setFeeder(AdaptConveyorFeeder<T> *feeder);
+
+	size_t space() const override;
+	size_t queued() const override;
+};
+
 template <typename T> class ConvertConveyorNode : public ConveyorNode {};
 
+class QueueBufferConveyorNodeBase : public ConveyorNode,
+									public ConveyorStorage {
+public:
+	virtual ~QueueBufferConveyorNodeBase() = default;
+};
+
 template <typename T, size_t S>
-class QueueBufferConveyorNode : public ConveyorNode, public Event {
+class QueueBufferConveyorNode : public QueueBufferConveyorNodeBase,
+								public Event {
 private:
-	std::queue<T, std::array<T,S>> storage;
+	std::queue<T, std::array<T, S>> storage;
+
 public:
 	void fire() override;
 };
 
-template<typename T>
-class QueueBufferConveyorNode<T> : public ConveyorNode, public Event {
+template <typename T>
+class QueueBufferConveyorNode<T, 0> : public QueueBufferConveyorNodeBase,
+									  public Event {
 private:
 	std::queue<T> storage;
+
 public:
 	void fire() override;
 };
 } // namespace gin
 // Template inlining
-namespace ent {}
+namespace gin {
+template <typename T> AdaptConveyorFeeder<T>::~AdaptConveyorFeeder() {
+	if (feedee) {
+		feedee->setFeeder(nullptr);
+		feedee = nullptr;
+	}
+}
+
+template <typename T>
+void AdaptConveyorFeeder<T>::setFeedee(AdaptConveyorNode<T> *feedee_p) {
+	feedee = feedee_p;
+}
+
+template <typename T> AdaptConveyorNode<T>::~AdaptConveyorNode() {
+	if (feeder) {
+		feeder->setFeedee(nullptr);
+		feeder = nullptr;
+	}
+}
+
+template <typename T>
+void AdaptConveyorNode<T>::setFeeder(AdaptConveyorFeeder<T> *feeder_p) {
+	feeder = feeder_p;
+}
+} // namespace gin

source/log.cpp

@@ -1,5 +1,7 @@
 #include "log.h"
 
 namespace gin {
+LogIo::LogIo(EventLoop &loop) : loop{loop} {}
 
-}
+Log::Log(LogIo &central, EventLoop &loop) : central{central}, loop{loop} {}
+} // namespace gin

source/log.h

@@ -4,25 +4,23 @@
 
 namespace gin {
 class EventLoop;
-class LogIo {
-private:
-	EventLoop& loop;
-public:
-	LogIo(EventLoop& loop);
-};
-
+class LogIo;
 class Log {
 public:
-	enum class Type : uint8_t {
-		Info,
-		Warning,
-		Error,
-		Debug
-	};
+	enum class Type : uint8_t { Info, Warning, Error, Debug };
+
 private:
-	LogIo& central;
-	EventLoop& loop;
+	LogIo &central;
+	EventLoop &loop;
+
 public:
-	Log(LogIo& central, EventLoop& loop);
+	Log(LogIo &central, EventLoop &loop);
 };
-}
+class LogIo {
+private:
+	EventLoop &loop;
+
+public:
+	LogIo(EventLoop &loop);
+};
+} // namespace gin