Added in/out port and respective bidirectional connection.

conan/Flow/conanfile.py

@@ -5,7 +5,7 @@
 
 class FlowCore(ConanFile):
 	name = "Flow"
-	version = "1.3"
+	version = "1.4"
 	description = """Flow is a pipes and filters implementation tailored for microcontrollers. 
 		It provides 3 base concepts: component, port and connection."""
 	url = "https://github.com/CynaraKrewe/Flow"
@@ -14,10 +14,10 @@ class FlowCore(ConanFile):
 	build_policy = "missing"
 
 	def source(self):
-		download("https://github.com/CynaraKrewe/Flow/archive/v1.3.zip", "Flow-1.3.zip")
-		unzip("Flow-1.3.zip")
-		shutil.move("Flow-1.3", "Flow")
-		os.unlink("Flow-1.3.zip")
+		download("https://github.com/CynaraKrewe/Flow/archive/v1.4.zip", "Flow-1.4.zip")
+		unzip("Flow-1.4.zip")
+		shutil.move("Flow-1.4", "Flow")
+		os.unlink("Flow-1.4.zip")
 
 	def build(self):
 		self.output.info("Nothing to build, this package provides sources.")

conanfile.py

@@ -1,7 +1,7 @@
 from conans import ConanFile
 
 class ExampleUsingFlow(ConanFile):
-   requires = "Flow/1.3@spiessensm/stable"
+   requires = "Flow/1.4@spiessensm/stable"
 
    def imports(self):
       self.copy("*.h")

include/flow/flow.h

@@ -36,6 +36,9 @@ class InPort;
 template<typename Type>
 class OutPort;
 
+template<typename Type>
+class InOutPort;
+
 class Connection
 {
 public:
@@ -88,6 +91,20 @@ class ConnectionOfType: public Connection,
 	OutPort<Type>& sender;
 };
 
+template<typename Type>
+class BiDirectionalConnectionOfType: public Connection
+{
+public:
+	BiDirectionalConnectionOfType(InOutPort<Type>& portA, InOutPort<Type>& portB,
+			unsigned int size) :
+			connectionA(ConnectionOfType<Type>(portA, portB, size)),
+			connectionB(ConnectionOfType<Type>(portB, portA, size))
+	{}
+
+private:
+	ConnectionOfType<Type> connectionA, connectionB;
+};
+
 template<typename Type>
 class InPort
 {
@@ -168,13 +185,34 @@ class OutPort
 	friend class ConnectionOfType<Type> ;
 };
 
+
+
+template<typename Type>
+class InOutPort
+:	public InPort<Type>,
+	public OutPort<Type>
+{
+public:
+	InOutPort<Type>()
+	:	InPort<Type>(),
+		OutPort<Type>()
+	{}
+};
+
 template<typename Type>
 Connection* connect(OutPort<Type>& sender, InPort<Type>& receiver,
 		unsigned int size = 1)
 {
 	return new ConnectionOfType<Type>(sender, receiver, size);
 }
 
+template<typename Type>
+Connection* connect(InOutPort<Type>& portA, InOutPort<Type>& portB,
+		unsigned int size = 1)
+{
+	return new BiDirectionalConnectionOfType<Type>(portA, portB, size);
+}
+
 void disconnect(Connection* connection);
 
 class Component

source/flow_test/inoutport_tests.cpp

@@ -0,0 +1,138 @@
+/*
+ The MIT License (MIT)
+
+ Copyright (c) 2017 Cynara Krewe
+
+ Permission is hereby granted, free of charge, to any person obtaining a copy
+ of this software, hardware and associated documentation files (the "Solution"), to deal
+ in the Solution without restriction, including without limitation the rights
+ to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ copies of the Solution, and to permit persons to whom the Solution is
+ furnished to do so, subject to the following conditions:
+
+ The above copyright notice and this permission notice shall be included in all
+ copies or substantial portions of the Solution.
+
+ THE SOLUTION IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ OUT OF OR IN CONNECTION WITH THE SOLUTION OR THE USE OR OTHER DEALINGS IN THE
+ SOLUTION.
+ */
+
+#include <stdint.h>
+#include <thread>
+
+#include "CppUTest/TestHarness.h"
+
+#include "flow/flow.h"
+
+#include "flow_test/data.h"
+
+using Flow::Connection;
+using Flow::InOutPort;
+using Flow::connect;
+
+const static unsigned int CONNECTION_FIFO_SIZE = 10;
+
+TEST_GROUP(InOutPort_TestBench)
+{
+	Connection* connection;
+	InOutPort<Data> unitUnderTestA;
+	InOutPort<Data> unitUnderTestB;
+
+	void setup()
+	{
+		connection = connect(unitUnderTestA, unitUnderTestB,
+				CONNECTION_FIFO_SIZE);
+	}
+
+	void teardown()
+	{
+		disconnect(connection);
+	}
+};
+
+TEST(InOutPort_TestBench, IsEmptyAfterCreation)
+{
+	CHECK(!unitUnderTestA.peek());
+	Data response;
+	CHECK(!unitUnderTestA.receive(response));
+
+	CHECK(!unitUnderTestB.peek());
+	CHECK(!unitUnderTestB.receive(response));
+}
+
+TEST(InOutPort_TestBench, SendReceiveItem)
+{
+	CHECK(!unitUnderTestA.peek());
+	Data stimulus = Data(123, true);
+	CHECK(unitUnderTestB.send(stimulus));
+	CHECK(unitUnderTestA.peek());
+	Data response;
+	CHECK(unitUnderTestA.receive(response));
+	CHECK(stimulus == response);
+	CHECK(!unitUnderTestA.peek());
+	CHECK(!unitUnderTestA.receive(response));
+
+	CHECK(!unitUnderTestB.peek());
+	CHECK(unitUnderTestA.send(stimulus));
+	CHECK(unitUnderTestB.peek());
+	CHECK(unitUnderTestB.receive(response));
+	CHECK(stimulus == response);
+	CHECK(!unitUnderTestB.peek());
+	CHECK(!unitUnderTestB.receive(response));
+}
+
+static void producer(InOutPort<Data>* _unitUnderTest,
+		const unsigned long long count)
+{
+	for (unsigned long long c = 0; c <= count; c++)
+	{
+		while (!_unitUnderTest->send(Data(c, ((c % 2) == 0))))
+			;
+	}
+}
+
+static void consumer(InOutPort<Data>* _unitUnderTest,
+		const unsigned long long count, bool* success)
+{
+	unsigned long long c = 0;
+
+	while (c <= count)
+	{
+		Data response;
+		if (_unitUnderTest->receive(response))
+		{
+			Data expected = Data(c, ((c % 2) == 0));
+			*success = *success && (response == expected);
+			c++;
+		}
+	}
+}
+
+TEST(InOutPort_TestBench, Threadsafe)
+{
+	const unsigned long long count = 1000000;
+	bool success = true;
+
+	std::thread producerAThread(producer, &unitUnderTestA, count);
+	std::thread consumerBThread(consumer, &unitUnderTestB, count, &success);
+
+	producerAThread.join();
+	consumerBThread.join();
+
+	CHECK(success);
+
+	success = true;
+
+	std::thread producerBThread(producer, &unitUnderTestB, count);
+	std::thread consumerAThread(consumer, &unitUnderTestA, count, &success);
+
+	producerBThread.join();
+	consumerAThread.join();
+
+	CHECK(success);
+}

source/flow_test/port_tests.cpp

@@ -165,8 +165,8 @@ TEST(Port_TestBench, Threadsafe)
 	std::thread producerThread(producer, &outUnitUnderTest, count);
 	std::thread consumerThread(consumer, &inUnitUnderTest, count, &success);
 
-	CHECK(success);
-
 	producerThread.join();
 	consumerThread.join();
+
+	CHECK(success);
 }