--- a/include/relpipe/reader/handlers/RelationalReaderBaseHandler.h	Mon Sep 03 23:40:12 2018 +0200
+++ b/include/relpipe/reader/handlers/RelationalReaderBaseHandler.h	Sat Sep 08 19:42:29 2018 +0200
@@ -12,10 +12,12 @@
 public:
 	virtual ~RelationalReaderBaseHadler() = default;
 
-	virtual void startRelation(string_t name, std::vector<std::pair<string_t, TypeId>> attributes) = 0;
+    virtual void startRelation(string_t name, std::vector<std::pair<string_t, TypeId>> attributes) = 0;
+
+    virtual void endOfPipe() = 0;
 
 };
 
 }
 }
-}
\ No newline at end of file
+}

--- a/src/Factory.cpp	Mon Sep 03 23:40:12 2018 +0200
+++ b/src/Factory.cpp	Sat Sep 08 19:42:29 2018 +0200
@@ -28,4 +28,4 @@
 static_assert(std::is_same<relpipe::reader::octet_t, typename relpipe::protocol::octet_t>::value, PROTOCOL_VIOLATION_TYPE_DEF_ERROR);
 
 }
-}
\ No newline at end of file
+}

--- a/src/StreamRelationalReader.h	Mon Sep 03 23:40:12 2018 +0200
+++ b/src/StreamRelationalReader.h	Sat Sep 08 19:42:29 2018 +0200
@@ -67,6 +67,11 @@
 		throw RelpipeReaderException(L"Unsupported data type: " + (int) typeId);
 	}
 
+	void endOfPipe() {
+		for (int i = 0; i < stringHandlers.size(); i++) stringHandlers[i]->endOfPipe();
+		for (int i = 0; i < valueHandlers.size(); i++) valueHandlers[i]->endOfPipe();
+	}
+
 public:
 
 	StreamRelationalReader(std::istream &input) :
@@ -100,11 +105,13 @@
 						// input was fully read
 						// we are finished
 						// TODO: printCachedData(output); ???
+						endOfPipe();
 						return;
 					} else if (dataPart == DATA_PART_START) {
 						// Empty relation might be weird but it is valid data.
 						// Actually, it is not so weird as it looks.
 						// fwprintf(stderr, L"Warning: The table has no rows. Weird… but OK.\n");
+						endOfPipe();
 						return;
 					} else {
 						// in current format, there is no other data part
@@ -182,4 +189,4 @@
 };
 
 }
-}
\ No newline at end of file
+}