diff --git a/generator.cc b/generator.cc
index 96167c36..1be79d64 100644
--- a/generator.cc
+++ b/generator.cc
@@ -46,8 +46,9 @@ generator::generator(json const &config)
     , _o_file_name(out_name(config)) {
 
     seed_seq_from<pcg32> main_seeder(_seed);
+    std::unordered_map<std::string, std::shared_ptr<std::unique_ptr<stream>>> map;
 
-    _stream_a = make_stream(config.at("stream"), main_seeder, std::size_t(config.at("tv_size")));
+    _stream_a = make_stream(config.at("stream"), main_seeder, map, std::size_t(config.at("tv_size")));
 }
 
 void generator::generate() {
diff --git a/stream.h b/stream.h
index 76177ae5..a47b6e6c 100644
--- a/stream.h
+++ b/stream.h
@@ -4,6 +4,7 @@
 #include <eacirc-core/json.h>
 #include <eacirc-core/logger.h>
 #include <eacirc-core/view.h>
+#include <unordered_map>
 #include <vector>
 
 using value_type = std::uint8_t;
diff --git a/streams.cc b/streams.cc
index 40054ea0..c5c986fa 100644
--- a/streams.cc
+++ b/streams.cc
@@ -18,11 +18,13 @@ vec_cview file_stream::next() {
     return make_cview(_data);
 }
 
-single_value_stream::single_value_stream(const json &config,
-                                         default_seed_source &seeder,
-                                         const std::size_t osize)
+single_value_stream::single_value_stream(
+    const json &config,
+    default_seed_source &seeder,
+    std::unordered_map<std::string, std::shared_ptr<std::unique_ptr<stream>>> &pipes,
+    const std::size_t osize)
     : stream(osize) {
-    auto stream = make_stream(config, seeder, osize);
+    auto stream = make_stream(config, seeder, pipes, osize);
     vec_cview single_vector = stream->next();
     std::copy(single_vector.begin(), single_vector.end(), _data.begin());
 }
@@ -31,22 +33,16 @@ vec_cview single_value_stream::next() {
     return make_cview(_data);
 }
 
-repeating_stream::repeating_stream(const json &config,
-                                   default_seed_source &seeder,
-                                   const std::size_t osize)
+repeating_stream::repeating_stream(
+    const json &config,
+    default_seed_source &seeder,
+    std::unordered_map<std::string, std::shared_ptr<std::unique_ptr<stream>>> &pipes,
+    const std::size_t osize)
     : stream(osize)
-    , _source(make_stream(config.at("source"), seeder, osize))
+    , _source(make_stream(config.at("source"), seeder, pipes, osize))
     , _period(unsigned(config.value("period", 0)))
     , _i(0) {}
 
-repeating_stream::repeating_stream(const std::size_t osize,
-                                   std::unique_ptr<stream> source,
-                                   const unsigned period)
-    : stream(osize)
-    , _source(std::move(source))
-    , _period(period)
-    , _i(0) {}
-
 vec_cview repeating_stream::next() {
     if (_i % _period == 0) {
         _data = _source->next().copy_to_vector();
@@ -79,9 +75,13 @@ random_start_counter::random_start_counter(default_seed_source &seeder, const st
 }
 
 template <typename Seeder>
-xor_stream::xor_stream(const nlohmann::json &config, Seeder &&seeder, const std::size_t osize)
+xor_stream::xor_stream(
+    const json &config,
+    Seeder &&seeder,
+    std::unordered_map<std::string, std::shared_ptr<std::unique_ptr<stream>>> &pipes,
+    const std::size_t osize)
     : stream(osize)
-    , _source(make_stream(config.at("source"), seeder, osize * 2)) {}
+    , _source(make_stream(config.at("source"), seeder, pipes, osize * 2)) {}
 
 vec_cview xor_stream::next() {
     vec_cview in = _source->next();
@@ -97,38 +97,6 @@ vec_cview xor_stream::next() {
     return make_cview(_data);
 }
 
-template <typename Seeder>
-rnd_plt_ctx_stream::rnd_plt_ctx_stream(const nlohmann::json &config,
-                                       Seeder &&seeder,
-                                       const std::size_t osize)
-    : stream(osize)
-    , _rng(std::make_unique<pcg32_stream>(seeder, osize / 2))
-    , _source(make_stream(config, seeder, osize / 2, core::optional<stream *>{_rng.get()})) {}
-
-vec_cview rnd_plt_ctx_stream::next() {
-    vec_cview ctx = _source->next();
-    vec_cview ptx = _rng->get_data();
-
-    std::copy_n(ptx.begin(), osize() / 2, _data.begin());
-    std::copy_n(ctx.begin(), osize() / 2, _data.begin() + osize() / 2);
-
-    return make_cview(_data);
-}
-
-template <typename Seeder>
-rho_stream::rho_stream(const nlohmann::json &config, Seeder &&seeder, const std::size_t osize)
-    : stream(osize)
-    , _ptx(std::make_unique<dummy_stream>(osize))
-    , _source(make_stream(config, seeder, osize, core::optional<stream *>{_ptx.get()})) {}
-
-vec_cview rho_stream::next() {
-    _ptx->set_data(make_cview(_data));
-    vec_cview ctx = _source->next();
-
-    std::copy(ctx.begin(), ctx.end(), _data.begin());
-    return make_cview(_data);
-}
-
 vec_cview hw_counter::next() {
     std::copy_n(_origin_data.begin(), osize(), _data.begin());
     for (const auto &pos : _cur_positions) {
@@ -152,14 +120,16 @@ vec_cview hw_counter::next() {
     return make_cview(_data);
 }
 
-column_stream::column_stream(const json &config,
-                             default_seed_source &seeder,
-                             const std::size_t osize)
+column_stream::column_stream(
+    const json &config,
+    default_seed_source &seeder,
+    std::unordered_map<std::string, std::shared_ptr<std::unique_ptr<stream>>> &pipes,
+    const std::size_t osize)
     : stream(osize)
     , _internal_bit_size(std::size_t(config.at("size")) * 8)
     , _buf(_internal_bit_size)
     , _position(0)
-    , _source(make_stream(config.at("source"), seeder, _internal_bit_size / 8)) {
+    , _source(make_stream(config.at("source"), seeder, pipes, _internal_bit_size / 8)) {
     for (auto &v : _buf)
         v.resize(osize);
 }
@@ -187,13 +157,15 @@ vec_cview column_stream::next() {
     return make_cview(_buf[_position++]); // return and increment
 }
 
-column_fixed_position_stream::column_fixed_position_stream(const json &config,
-                                                           default_seed_source &seeder,
-                                                           const std::size_t osize,
-                                                           const std::size_t position)
+column_fixed_position_stream::column_fixed_position_stream(
+    const json &config,
+    default_seed_source &seeder,
+    std::unordered_map<std::string, std::shared_ptr<std::unique_ptr<stream>>> &pipes,
+    const std::size_t osize,
+    const std::size_t position)
     : stream(osize)
     , _position(position)
-    , _source(make_stream(config.at("source"), seeder, std::size_t(config.at("size")))) {}
+    , _source(make_stream(config.at("source"), seeder, pipes, std::size_t(config.at("size")))) {}
 
 vec_cview column_fixed_position_stream::next() {
     for (auto &val : _data)
@@ -211,14 +183,52 @@ vec_cview column_fixed_position_stream::next() {
     return make_cview(_data); // return and increment
 }
 
-std::unique_ptr<stream> make_stream(const json &config,
-                                    default_seed_source &seeder,
-                                    const std::size_t osize,
-                                    core::optional<stream *> stream) {
+pipe_in_stream::pipe_in_stream(
+    const nlohmann::json &config,
+    default_seed_source &seeder,
+    std::unordered_map<std::string, std::shared_ptr<std::unique_ptr<stream>>> &pipes,
+    const std::size_t osize)
+    : stream(0) {
+    std::string pipe_id = config.at("id");
+
+    // substream has to be created in advance
+    // creating it separately in following if would cause inconsistances
+    std::unique_ptr<stream> new_stream = make_stream(config.at("source"), seeder, pipes, osize);
+
+    auto search = pipes.find(pipe_id);
+    if (search == pipes.end()) {
+        // pipe_id is not yet in hashtable, create new entry
+        _source = std::make_shared<std::unique_ptr<stream>>(std::move(new_stream));
+        pipes[pipe_id] = _source;
+    } else {
+        // pipe_id is in hashtable, update the entry for pipe_out
+        *pipes[pipe_id] = std::move(new_stream);
+        _source = pipes[pipe_id];
+    }
+}
+
+tuple_stream::tuple_stream(
+    const nlohmann::json &config,
+    default_seed_source &seeder,
+    std::unordered_map<std::string, std::shared_ptr<std::unique_ptr<stream>>> &pipes,
+    const std::size_t osize)
+    : stream(osize) {
+    for (const auto &stream_cfg : config.at("sources")) {
+        _sources.push_back(make_stream(
+            stream_cfg, seeder, pipes, std::size_t(stream_cfg.value("output_size", 0))));
+    }
+}
+
+std::unique_ptr<stream>
+make_stream(const json &config,
+            default_seed_source &seeder,
+            std::unordered_map<std::string, std::shared_ptr<std::unique_ptr<stream>>> &pipes,
+            const std::size_t osize) {
     const std::string type = config.at("type");
 
-    if (osize == 0 and type != "dummy_stream") // we allow dummy stream with 0 size
-        throw std::runtime_error("Stream " + type + " cannot have osize 0.");
+    if (osize == 0 and type != "dummy_stream") { // we allow dummy stream with 0 size
+        logger::warning() << "Stream " + type + " have osize 0." << std::endl;
+    }
 
     // trivial source only streams
     if (type == "dummy_stream")
@@ -234,11 +244,6 @@ std::unique_ptr<stream> make_stream(const json &config,
     else if (type == "pcg32_stream" or type == "random_stream")
         return std::make_unique<pcg32_stream>(seeder, osize);
 
-    // types of plaintext (also only sources)
-    else if (type == "rnd_plt_ctx_stream")
-        return std::make_unique<rnd_plt_ctx_stream>(config.at("source"), seeder, osize);
-    else if (type == "rho_stream")
-        return std::make_unique<rho_stream>(config.at("source"), seeder, osize);
     else if (type == "counter")
         return std::make_unique<counter>(osize);
     else if (type == "random_start_counter")
@@ -265,40 +270,48 @@ std::unique_ptr<stream> make_stream(const json &config,
     else if (type == "exponential_distribution")
         return std::make_unique<exponential_distribution_stream>(config, seeder, osize);
 
-    // modifiers -- streams that has other stream as an input (but these are used as source before
-    // cipher)
+    // modifiers -- streams that has other stream as an input (but are used as source before cipher)
     else if (type == "single_value_stream")
-        return std::make_unique<single_value_stream>(config.at("source"), seeder, osize);
+        return std::make_unique<single_value_stream>(config.at("source"), seeder, pipes, osize);
     else if (type == "repeating_stream")
-        return std::make_unique<repeating_stream>(config.at("source"), seeder, osize);
+        return std::make_unique<repeating_stream>(config.at("source"), seeder, pipes, osize);
+    else if (type == "tuple_stream")
+        return std::make_unique<tuple_stream>(config, seeder, pipes, osize);
+
+    // pipes
+    else if (type == "pipe_in_stream")
+        return std::make_unique<pipe_in_stream>(config, seeder, pipes, osize);
+    else if (type == "pipe_out_stream")
+        return std::make_unique<pipe_out_stream>(config, pipes);
 
     // postprocessing modifiers -- streams that has cipher stream as an input
     else if (type == "xor_stream")
-        return std::make_unique<xor_stream>(config, seeder, osize);
+        return std::make_unique<xor_stream>(config, seeder, pipes, osize);
     else if (type == "column")
-        return std::make_unique<column_stream>(config, seeder, osize);
+        return std::make_unique<column_stream>(config, seeder, pipes, osize);
     else if (type == "column_fixed_position") {
         const std::size_t pos = std::size_t(config.at("position"));
-        return std::make_unique<column_fixed_position_stream>(config, seeder, osize, pos);
+        return std::make_unique<column_fixed_position_stream>(config, seeder, pipes, osize, pos);
     }
 
-        // cryptoprimitives
+    // mock streams for testing
 #if (BUILD_testsuite && TEST_STREAM)
     else if (type == "test_stream")
         return std::make_unique<testsuite::test_stream>(config);
 #endif
 
+        // cryptoprimitives streams
 #ifdef BUILD_stream_ciphers
     else if (type == "stream_cipher" or type == "estream")
-        return std::make_unique<stream_ciphers::stream_stream>(config, seeder, osize, stream);
+        return std::make_unique<stream_ciphers::stream_stream>(config, seeder, pipes, osize);
 #endif
 #ifdef BUILD_hash
     else if (type == "hash" || type == "sha3")
-        return std::make_unique<hash::hash_stream>(config, seeder, osize, stream);
+        return std::make_unique<hash::hash_stream>(config, seeder, pipes, osize);
 #endif
 #ifdef BUILD_block
     else if (type == "block")
-        return std::make_unique<block::block_stream>(config, seeder, osize, stream);
+        return std::make_unique<block::block_stream>(config, seeder, pipes, osize);
 #endif
     throw std::runtime_error("requested stream named \"" + type + "\" does not exist");
 }
diff --git a/streams.h b/streams.h
index ef629561..0187c12e 100644
--- a/streams.h
+++ b/streams.h
@@ -80,10 +80,11 @@ struct file_stream : stream {
  * @brief Stream outputing a constant value for n iterations
  */
 struct repeating_stream : stream {
-    repeating_stream(const json &config, default_seed_source &seeder, const std::size_t osize);
-    repeating_stream(const std::size_t osize,
-                     std::unique_ptr<stream> source,
-                     const unsigned period);
+    repeating_stream(
+        const json &config,
+        default_seed_source &seeder,
+        std::unordered_map<std::string, std::shared_ptr<std::unique_ptr<stream>>> &pipes,
+        const std::size_t osize);
 
     vec_cview next() override;
 
@@ -97,7 +98,11 @@ struct repeating_stream : stream {
  * @brief Stream outputing a constant value
  */
 struct single_value_stream : stream {
-    single_value_stream(const json &config, default_seed_source &seeder, const std::size_t osize);
+    single_value_stream(
+        const json &config,
+        default_seed_source &seeder,
+        std::unordered_map<std::string, std::shared_ptr<std::unique_ptr<stream>>> &pipes,
+        const std::size_t osize);
 
     vec_cview next() override;
 };
@@ -123,7 +128,10 @@ struct random_start_counter : counter {
  */
 struct xor_stream : stream {
     template <typename Seeder>
-    xor_stream(const json &config, Seeder &&seeder, const std::size_t osize);
+    xor_stream(const json &config,
+               Seeder &&seeder,
+               std::unordered_map<std::string, std::shared_ptr<std::unique_ptr<stream>>> &pipes,
+               const std::size_t osize);
 
     vec_cview next() override;
 
@@ -131,38 +139,6 @@ struct xor_stream : stream {
     std::unique_ptr<stream> _source;
 };
 
-/**
- * @brief Stream XORing two parts of internal stream
- */
-struct rnd_plt_ctx_stream : stream {
-    template <typename Seeder>
-    rnd_plt_ctx_stream(const json &config, Seeder &&seeder, const std::size_t osize);
-
-    vec_cview next() override;
-
-private:
-    std::unique_ptr<stream> _rng;
-    std::unique_ptr<stream> _source;
-};
-
-/**
- * @brief Stream for testing (Pollard) rho-method
- *
- * Stateful generator
- * The last output is used as next input.
- * Initially starts with vector of zeros.
- */
-struct rho_stream : stream {
-    template <typename Seeder>
-    rho_stream(const json &config, Seeder &&seeder, const std::size_t osize);
-
-    vec_cview next() override;
-
-private:
-    std::unique_ptr<stream> _ptx;
-    std::unique_ptr<stream> _source;
-};
-
 /**
  * @brief Stream for testing strict avalanche criterion
  *
@@ -336,7 +312,10 @@ struct hw_counter : stream {
 };
 
 struct column_stream : stream {
-    column_stream(const json &config, default_seed_source &seeder, const std::size_t osize);
+    column_stream(const json &config,
+                  default_seed_source &seeder,
+                  std::unordered_map<std::string, std::shared_ptr<std::unique_ptr<stream>>> &pipes,
+                  const std::size_t osize);
 
     vec_cview next() override;
 
@@ -349,10 +328,12 @@ struct column_stream : stream {
 };
 
 struct column_fixed_position_stream : stream {
-    column_fixed_position_stream(const json &config,
-                                 default_seed_source &seeder,
-                                 const std::size_t osize,
-                                 const std::size_t position);
+    column_fixed_position_stream(
+        const json &config,
+        default_seed_source &seeder,
+        std::unordered_map<std::string, std::shared_ptr<std::unique_ptr<stream>>> &pipes,
+        const std::size_t osize,
+        const std::size_t position);
 
     vec_cview next() override;
 
@@ -480,6 +461,74 @@ struct exponential_distribution_stream : stream {
     std::exponential_distribution<double> _distribution;
 };
 
+/**
+ * @brief Pipe's sink - stores pointer on the intersnal stream to
+ * the pipes hashtable
+ */
+struct pipe_in_stream : stream {
+    pipe_in_stream(const json &config,
+                   default_seed_source &seeder,
+                   std::unordered_map<std::string, std::shared_ptr<std::unique_ptr<stream>>> &pipes,
+                   const std::size_t osize);
+
+    vec_cview next() override { return (*_source)->next(); }
+
+private:
+    std::shared_ptr<std::unique_ptr<stream>> _source;
+};
+
+/**
+ * @brief Pipe's source
+ */
+struct pipe_out_stream : stream {
+    pipe_out_stream(
+        const json &config,
+        std::unordered_map<std::string, std::shared_ptr<std::unique_ptr<stream>>> &pipes)
+        : stream(0) {
+        std::string pipe_id = config.at("id");
+
+        auto search = pipes.find(pipe_id);
+        if (search == pipes.end()) {
+            // pipe_id is not yet in hashtable, create new empty entry
+            _source =
+                std::make_shared<std::unique_ptr<stream>>(std::unique_ptr<stream>(nullptr));
+            pipes[pipe_id] = _source;
+        } else {
+            // pipe_id is in hashtable, use ptr from pipe_in
+            _source = pipes[pipe_id];
+        }
+    }
+
+    vec_cview next() override { return (*_source)->get_data(); }
+
+private:
+    std::shared_ptr<std::unique_ptr<stream>> _source;
+};
+
+/**
+ * @brief Ordered tuple of streams
+ */
+struct tuple_stream : stream {
+    tuple_stream(const json &config,
+                 default_seed_source &seeder,
+                 std::unordered_map<std::string, std::shared_ptr<std::unique_ptr<stream>>> &pipes,
+                 const std::size_t osize);
+
+    vec_cview next() override {
+        auto beg = _data.begin();
+        for (auto& source : _sources) {
+            vec_cview v = source->next();
+            copy(v.begin(), v.end(), beg);
+            beg += v.end() - v.begin();
+        }
+
+        return make_cview(_data);
+    }
+
+private:
+    std::vector<std::unique_ptr<stream>> _sources;
+};
+
 /**
  * \brief Stream of true bits
  */
@@ -500,8 +549,9 @@ using mt19937_stream = _impl::rng_stream<std::mt19937>;
  */
 using pcg32_stream = _impl::rng_stream<pcg32>;
 
-std::unique_ptr<stream> make_stream(const json &config,
-                                    default_seed_source &seeder,
-                                    std::size_t osize,
-                                    core::optional<stream *> stream = core::nullopt_t{});
+std::unique_ptr<stream>
+make_stream(const json &config,
+            default_seed_source &seeder,
+            std::unordered_map<std::string, std::shared_ptr<std::unique_ptr<stream>>> &pipes,
+            std::size_t osize);
 void stream_to_dataset(dataset &set, std::unique_ptr<stream> &source);
diff --git a/streams/block/block_stream.cc b/streams/block/block_stream.cc
index d45ee74e..350f4b31 100644
--- a/streams/block/block_stream.cc
+++ b/streams/block/block_stream.cc
@@ -25,25 +25,25 @@ static int64_t reinit_freq(const json &config) {
     }
 }
 
-block_stream::block_stream(const json &config,
-                           default_seed_source &seeder,
-                           const std::size_t osize,
-                           core::optional<stream *> plt_stream)
+block_stream::block_stream(
+    const json &config,
+    default_seed_source &seeder,
+    std::unordered_map<std::string, std::shared_ptr<std::unique_ptr<stream>>> &pipes,
+    const std::size_t osize)
     : stream(osize)
     , _round(config.at("round"))
     , _block_size(config.at("block_size"))
     , _reinit_freq(reinit_freq(config))
     , _i(0)
-    , _source(make_stream(config.at("plaintext"), seeder, osize))
-    , _iv(make_stream(config.at("iv"), seeder, _block_size))
-    , _key(make_stream(config.at("key"), seeder, unsigned(config.at("key_size"))))
+    , _source(make_stream(config.at("plaintext"), seeder, pipes, osize))
+    , _iv(make_stream(config.at("iv"), seeder, pipes, _block_size))
+    , _key(make_stream(config.at("key"), seeder, pipes, unsigned(config.at("key_size"))))
     , _run_encryption(config.value("encryption_mode", true))
     , _encryptor(make_block_cipher(config.at("algorithm"),
                                    unsigned(_round),
                                    unsigned(_block_size),
                                    unsigned(config.at("key_size")),
-                                   _run_encryption))
-    , _prepared_stream_source(!plt_stream ? nullptr : *plt_stream) {
+                                   _run_encryption)) {
     logger::info() << "stream source is block cipher: " << config.at("algorithm") << std::endl;
 
     if (int(config.at("round")) < 0)
@@ -77,7 +77,7 @@ vec_cview block_stream::next() {
 
     for (auto ctx_beg = _data.begin();
          ctx_beg != _data.end();) { // ctx_beg += _source->osize() from inside
-        vec_cview view = get_next_ptx();
+        vec_cview view = _source->next();
         for (auto ptx_beg = view.begin(); ptx_beg != view.end() and ctx_beg != _data.end();
              ptx_beg += _block_size, ctx_beg += _block_size) {
             _encryptor->crypt(&(*ptx_beg), &(*ctx_beg), _run_encryption);
@@ -87,12 +87,4 @@ vec_cview block_stream::next() {
     return make_view(_data.cbegin(), osize());
 }
 
-vec_cview block_stream::get_next_ptx() {
-    if (_prepared_stream_source) {
-        return _prepared_stream_source->next();
-    } else {
-        return _source->next();
-    }
-}
-
 } // namespace block
diff --git a/streams/block/block_stream.h b/streams/block/block_stream.h
index af870313..c500008d 100644
--- a/streams/block/block_stream.h
+++ b/streams/block/block_stream.h
@@ -15,15 +15,14 @@ struct block_stream : public stream {
 public:
     block_stream(const json &config,
                  default_seed_source &seeder,
-                 const std::size_t osize,
-                 core::optional<stream *> plt_stream);
+                 std::unordered_map<std::string, std::shared_ptr<std::unique_ptr<stream>>> &pipes,
+                 const std::size_t osize);
     block_stream(block_stream &&);
     ~block_stream() override;
 
     vec_cview next() override;
 
 private:
-    vec_cview get_next_ptx();
 
     const std::size_t _round;
     const std::size_t _block_size;
@@ -36,7 +35,6 @@ struct block_stream : public stream {
 
     const bool _run_encryption;
     std::unique_ptr<block_cipher> _encryptor;
-    stream *_prepared_stream_source;
 };
 
 } // namespace block
diff --git a/streams/hash/hash_stream.cc b/streams/hash/hash_stream.cc
index d03953d2..92097a84 100644
--- a/streams/hash/hash_stream.cc
+++ b/streams/hash/hash_stream.cc
@@ -26,22 +26,25 @@ void hash_data(hash_interface &hasher,
         throw std::runtime_error("cannot finalize the hash (code: " + to_string(status) + ")");
 }
 
-hash_stream::hash_stream(const json &config,
-                         default_seed_source &seeder,
-                         const std::size_t osize,
-                         core::optional<stream *> plt_stream)
+hash_stream::hash_stream(
+    const json &config,
+    default_seed_source &seeder,
+    std::unordered_map<std::string, std::shared_ptr<std::unique_ptr<stream>>> &pipes,
+    const std::size_t osize)
     : stream(osize) // round osize to multiple of _hash_input_size
     , _round(config.at("round"))
     , _hash_size(std::size_t(config.at("hash_size")))
     , _source(make_stream(
           config.at("source"),
           seeder,
+          pipes,
           config.value("input_size", _hash_size))) // if input size is not defined, use hash-size
-    , _prepared_stream_source(!plt_stream ? nullptr : *plt_stream)
     , _hasher(hash_factory::create(config.at("algorithm"), unsigned(_round))) {
-    if (osize % _hash_size != 0) // not necessary wrong, but we never needed this, we always did
-                                 // this by mistake. Change to warning if needed
+    if (osize % _hash_size != 0) {
+        // not necessary wrong, but we never needed this, we always did
+        // this by mistake. Change to warning if needed
         throw std::runtime_error("Output size is not multiple of hash size");
+    }
     logger::info() << "stream source is hash function: " << config.at("algorithm") << std::endl;
 }
 
@@ -51,7 +54,7 @@ hash_stream::~hash_stream() = default;
 vec_cview hash_stream::next() {
     auto hash = _data.data();
     for (std::size_t i = 0; i < _data.size(); i += _hash_size) {
-        vec_cview view = get_next_ptx();
+        vec_cview view = _source->next();
 
         hash_data(*_hasher, view, &hash[i], _hash_size);
     }
@@ -59,12 +62,4 @@ vec_cview hash_stream::next() {
     return make_view(_data.cbegin(), osize());
 }
 
-vec_cview hash_stream::get_next_ptx() {
-    if (_prepared_stream_source) {
-        return _prepared_stream_source->next();
-    } else {
-        return _source->next();
-    }
-}
-
 } // namespace hash
diff --git a/streams/hash/hash_stream.h b/streams/hash/hash_stream.h
index 012b1fd9..6d7f3bf8 100644
--- a/streams/hash/hash_stream.h
+++ b/streams/hash/hash_stream.h
@@ -19,16 +19,14 @@ void hash_data(hash_interface &hasher,
 struct hash_stream : stream {
     hash_stream(const json &config,
                 default_seed_source &seeder,
-                const std::size_t osize,
-                core::optional<stream *> plt_stream);
+                std::unordered_map<std::string, std::shared_ptr<std::unique_ptr<stream>>> &pipes,
+                const std::size_t osize);
     hash_stream(hash_stream &&);
     ~hash_stream() override;
 
     vec_cview next() override;
 
 private:
-    vec_cview get_next_ptx();
-
     const std::size_t _round;
     const std::size_t _hash_size;
 
diff --git a/streams/stream_ciphers/stream_stream.cc b/streams/stream_ciphers/stream_stream.cc
index 5523ebe2..71b55781 100644
--- a/streams/stream_ciphers/stream_stream.cc
+++ b/streams/stream_ciphers/stream_stream.cc
@@ -3,18 +3,18 @@
 
 namespace stream_ciphers {
 
-stream_stream::stream_stream(const json &config,
-                             default_seed_source &seeder,
-                             const std::size_t osize,
-                             core::optional<stream *> plt_stream)
+stream_stream::stream_stream(
+    const json &config,
+    default_seed_source &seeder,
+    std::unordered_map<std::string, std::shared_ptr<std::unique_ptr<stream>>> &pipes,
+    const std::size_t osize)
     : stream(osize)
     , _reinit(config.at("key").at("type") == "repeating_stream" or
               config.at("iv").at("type") == "repeating_stream")
     , _block_size(config.at("block_size"))
-    , _iv_stream(make_stream(config.at("iv"), seeder, config.value("iv_size", default_iv_size)))
-    , _key_stream(make_stream(config.at("key"), seeder, config.value("key_size", default_key_size)))
-    , _source(make_stream(config.at("plaintext"), seeder, _block_size))
-    , _prepared_stream_source(!plt_stream ? nullptr : *plt_stream)
+    , _iv_stream(make_stream(config.at("iv"), seeder, pipes, config.value("iv_size", default_iv_size)))
+    , _key_stream(make_stream(config.at("key"), seeder, pipes, config.value("key_size", default_key_size)))
+    , _source(make_stream(config.at("plaintext"), seeder, pipes, _block_size))
     , _plaintext(osize)
     , _algorithm(config.at("algorithm"),
                  unsigned(config.at("round")),
@@ -37,7 +37,7 @@ vec_cview stream_stream::next() {
         _algorithm.setup_key_iv(_key_stream, _iv_stream);
     }
     for (auto beg = _plaintext.begin(); beg != _plaintext.end(); beg += _block_size) {
-        vec_cview view = get_next_ptx();
+        vec_cview view = _source->next();
 
         std::move(view.begin(), view.end(), beg);
     }
@@ -47,12 +47,4 @@ vec_cview stream_stream::next() {
     return make_cview(_data);
 }
 
-vec_cview stream_stream::get_next_ptx() {
-    if (_prepared_stream_source) {
-        return _prepared_stream_source->next();
-    } else {
-        return _source->next();
-    }
-}
-
 } // namespace stream_ciphers
diff --git a/streams/stream_ciphers/stream_stream.h b/streams/stream_ciphers/stream_stream.h
index 23a58a61..09b74b3e 100644
--- a/streams/stream_ciphers/stream_stream.h
+++ b/streams/stream_ciphers/stream_stream.h
@@ -10,13 +10,12 @@ namespace stream_ciphers {
 struct stream_stream : stream {
     stream_stream(const json &config,
                   default_seed_source &seeder,
-                  const std::size_t osize,
-                  core::optional<stream *> plt_stream);
+                  std::unordered_map<std::string, std::shared_ptr<std::unique_ptr<stream>>> &pipes,
+                  const std::size_t osize);
 
     vec_cview next() override;
 
 private:
-    vec_cview get_next_ptx();
 
     const bool _reinit;
     const std::size_t _block_size;
@@ -27,7 +26,6 @@ struct stream_stream : stream {
     std::unique_ptr<stream> _key_stream;
 
     std::unique_ptr<stream> _source;
-    stream *_prepared_stream_source;
 
     std::vector<std::uint8_t> _plaintext;
 
diff --git a/testsuite/stream_tests.cc b/testsuite/stream_tests.cc
index 6574b970..bc71d5f7 100644
--- a/testsuite/stream_tests.cc
+++ b/testsuite/stream_tests.cc
@@ -197,8 +197,10 @@ TEST(column_streams, basic_test_with_counter) {
     json json_config = {{"size", 5}, {"source", {{"type", "counter"}}}};
 
     seed_seq_from<pcg32> seeder(testsuite::seed1);
+    std::unordered_map<std::string, std::shared_ptr<std::unique_ptr<stream>>> map;
 
-    std::unique_ptr<column_stream> stream = std::make_unique<column_stream>(json_config, seeder, 5);
+    std::unique_ptr<column_stream> stream =
+        std::make_unique<column_stream>(json_config, seeder, map, 5);
 
     // TODO: finish column test
     //    for(auto it : stream->next()) {
@@ -207,37 +209,59 @@ TEST(column_streams, basic_test_with_counter) {
 }
 
 TEST(rnd_plt_ctx_streams, aes_single_vector) {
-    const json json_config = {{"type", "rnd_plt_ctx_stream"},
-                              {"source",
-                               {{"type", "block"},
-                                {"init_frequency", "only_once"},
-                                {"algorithm", "AES"},
-                                {"round", 10},
-                                {"block_size", 16},
-                                {"plaintext", {{"type", "pcg32_stream"}}},
-                                {"key_size", 16},
-                                {"key", {{"type", "pcg32_stream"}}},
-                                {"iv", {{"type", "pcg32_stream"}}}}}};
+    const json json_config = R"({
+         "type": "tuple_stream",
+         "sources": [{
+                 "type": "block",
+                 "output_size": 16,
+                 "init_frequency": "only_once",
+                 "algorithm": "AES",
+                 "round": 10,
+                 "block_size": 16,
+                 "plaintext": {
+                     "type": "pipe_in_stream",
+                     "id": "ptx_stream",
+                     "source": {
+                         "type": "pcg32_stream"
+                     }
+                 },
+                 "key_size": 16,
+                 "key": {
+                     "type": "pcg32_stream"
+                 },
+                 "iv": {
+                     "type": "pcg32_stream"
+                 }
+             },
+             {
+                 "type": "pipe_out_stream",
+                 "id": "ptx_stream"
+             }
+         ]
+     }
+    )"_json;
 
     seed_seq_from<pcg32> seeder(testsuite::seed1);
-    std::unique_ptr<stream> stream = make_stream(json_config, seeder, 32);
+    std::unordered_map<std::string, std::shared_ptr<std::unique_ptr<stream>>> map;
+
+    std::unique_ptr<stream> stream = make_stream(json_config, seeder, map, 32);
 
     vec_cview view = stream->next();
 
     // fixed expected data AES 10 rounds, with PTX choosen by PCG32 with fixed seed testsuite::seed1
 #if defined(__clang__) // if compiler is clang, data are different from gcc :(
     std::vector<value_type> expected_data = {
+        0x99, 0x72, 0x5d, 0x0a, 0xac, 0xcb, 0x1a, 0x99,
+        0x50, 0xe6, 0x27, 0x53, 0x00, 0x87, 0x8b, 0x57, // ctx
         0x02, 0x6c, 0xc7, 0x6f, 0xaf, 0x91, 0x91, 0xb4,
-        0x82, 0x4c, 0x67, 0x8c, 0x22, 0x66, 0x39, 0x90, // ptx
-        0xd5, 0x36, 0xd3, 0x30, 0xf5, 0xe5, 0xde, 0x87,
-        0x75, 0x54, 0x82, 0x19, 0xf2, 0xb9, 0xa6, 0xcc // ctx
+        0x82, 0x4c, 0x67, 0x8c, 0x22, 0x66, 0x39, 0x90 // ptx
     };
 #elif defined(__GNUC__) || defined(__GNUG__)
     std::vector<value_type> expected_data = {
+        0x21, 0x66, 0xaf, 0xa3, 0xe2, 0xf9, 0x1d, 0xc3,
+        0xc1, 0xd8, 0xf6, 0xa6, 0xcf, 0xbb, 0xbb, 0x23, // ctx
         0xc5, 0x1e, 0xf0, 0xd7, 0x40, 0x1c, 0xa5, 0xea,
-        0x6f, 0x85, 0x11, 0x1e, 0x55, 0x74, 0x64, 0xcc, // ptx
-        0x54, 0x30, 0x76, 0x5f, 0xbf, 0xa7, 0x0d, 0xcb,
-        0xf0, 0x9b, 0xb3, 0xc0, 0xbe, 0x2b, 0xba, 0xaf // ctx
+        0x6f, 0x85, 0x11, 0x1e, 0x55, 0x74, 0x64, 0xcc // ptx
     };
 #elif defined(_MSC_VER) // MS data unknown
     std::vector<value_type> expected_data = {
@@ -249,17 +273,30 @@ TEST(rnd_plt_ctx_streams, aes_single_vector) {
 }
 
 TEST(rho_streams, aes_vector) {
-    const json json_config = {{"type", "rho_stream"},
-                              {"source",
-                               {{"type", "block"},
-                                {"init_frequency", "only_once"},
-                                {"algorithm", "AES"},
-                                {"round", 10},
-                                {"block_size", 16},
-                                {"plaintext", {{"type", "dummy_stream"}}},
-                                {"key_size", 16},
-                                {"key", {{"type", "false_stream"}}},
-                                {"iv", {{"type", "pcg32_stream"}}}}}};
+    const json json_config = R"({
+         "type": "pipe_in_stream",
+         "id": "rho_ptx",
+         "source": {
+                 "type": "block",
+                 "init_frequency": "only_once",
+                 "algorithm": "AES",
+                 "round": 10,
+                 "block_size": 16,
+                 "plaintext": {
+                     "type": "pipe_out_stream",
+                     "id": "rho_ptx"
+                 },
+                 "key_size": 16,
+                 "key": {
+                     "type": "false_stream"
+                 },
+                 "iv": {
+                     "type": "pcg32_stream"
+                 }
+             }
+
+     }
+    )"_json;
 
     json aes_json_config = {{"type", "block"},
                             {"init_frequency", "only_once"},
@@ -272,12 +309,14 @@ TEST(rho_streams, aes_vector) {
                             {"iv", {{"type", "pcg32_stream"}}}};
 
     seed_seq_from<pcg32> seeder(testsuite::seed1);
-    std::unique_ptr<stream> tested_stream = make_stream(json_config, seeder, 16);
+    std::unordered_map<std::string, std::shared_ptr<std::unique_ptr<stream>>> map;
+
+    std::unique_ptr<stream> tested_stream = make_stream(json_config, seeder, map, 16);
     std::vector<value_type> prev(16);
 
     for (unsigned i = 0; i < testing_size; ++i) {
         aes_json_config["plaintext"]["outputs"] = prev;
-        std::unique_ptr<stream> reference_stream = make_stream(aes_json_config, seeder, 16);
+        std::unique_ptr<stream> reference_stream = make_stream(aes_json_config, seeder, map, 16);
 
         vec_cview tested_view = tested_stream->next();
         vec_cview reference_view = reference_stream->next();
@@ -286,3 +325,66 @@ TEST(rho_streams, aes_vector) {
         std::copy(tested_view.begin(), tested_view.end(), prev.begin());
     }
 }
+
+TEST(pipe_streams, pipe_in_fist) {
+    const json json_in = R"({
+         "type": "pipe_in_stream",
+         "id": "id",
+         "source": {
+             "type": "counter"
+         }
+
+     }
+    )"_json;
+    const json json_out = R"({
+         "type": "pipe_out_stream",
+         "id": "id"
+     }
+    )"_json;
+
+    seed_seq_from<pcg32> seeder(testsuite::seed1);
+    std::unordered_map<std::string, std::shared_ptr<std::unique_ptr<stream>>> map;
+
+    std::unique_ptr<stream> pipe_in = make_stream(json_in, seeder, map, 16);
+    std::unique_ptr<stream> pipe_out = make_stream(json_out, seeder, map, 16);
+
+    for (unsigned i = 0; i < 4; ++i) {
+        vec_cview in_view = pipe_in->next();
+        vec_cview out_view = pipe_out->next();
+
+        ASSERT_EQ(in_view.copy_to_vector(), out_view.copy_to_vector());
+    }
+}
+
+TEST(pipe_streams, pipe_out_fist) {
+    const json json_in = R"({
+         "type": "pipe_in_stream",
+         "id": "id",
+         "source": {
+             "type": "counter"
+         }
+
+     }
+    )"_json;
+    const json json_out = R"({
+         "type": "pipe_out_stream",
+         "id": "id"
+     }
+    )"_json;
+
+    seed_seq_from<pcg32> seeder(testsuite::seed1);
+    std::unordered_map<std::string, std::shared_ptr<std::unique_ptr<stream>>> map;
+
+    std::unique_ptr<stream> pipe_out = make_stream(json_out, seeder, map, 16);
+    std::unique_ptr<stream> pipe_in = make_stream(json_in, seeder, map, 16);
+
+    for (unsigned i = 0; i < 4; ++i) {
+        vec_cview in_view = pipe_in->next();
+        vec_cview out_view = pipe_out->next();
+
+        ASSERT_EQ(in_view.copy_to_vector(), out_view.copy_to_vector());
+    }
+}
+
+// TODO: add test for creation of the pipes
+//       we should test both orders of creation
diff --git a/testsuite/test_utils/block_test_case.cc b/testsuite/test_utils/block_test_case.cc
index df7fb5bb..aa31709b 100644
--- a/testsuite/test_utils/block_test_case.cc
+++ b/testsuite/test_utils/block_test_case.cc
@@ -77,7 +77,9 @@ std::unique_ptr<stream> block_test_case::prepare_stream() {
     }
 
     seed_seq_from<pcg32> seeder(seed1);
-    return make_stream(_stream_config, seeder, block_size);
+    std::unordered_map<std::string, std::shared_ptr<std::unique_ptr<stream>>> map;
+
+    return make_stream(_stream_config, seeder, map, block_size);
 }
 
 std::istream &operator>>(std::istream &input, block_test_case &test_case) {
diff --git a/testsuite/test_utils/hash_test_case.cc b/testsuite/test_utils/hash_test_case.cc
index e78bdb66..4d6ddffe 100644
--- a/testsuite/test_utils/hash_test_case.cc
+++ b/testsuite/test_utils/hash_test_case.cc
@@ -65,7 +65,9 @@ std::unique_ptr<stream> hash_test_case::prepare_stream() {
     config_instance["hash_size"] = hash_size;
 
     seed_seq_from<pcg32> seeder(seed1);
-    return make_stream(config_instance, seeder, hash_size);
+    std::unordered_map<std::string, std::shared_ptr<std::unique_ptr<stream>>> map;
+
+    return make_stream(config_instance, seeder, map, hash_size);
 }
 
 std::istream &operator>>(std::istream &input, hash_test_case &test_case) {
diff --git a/testsuite/test_utils/stream_ciphers_test_case.cc b/testsuite/test_utils/stream_ciphers_test_case.cc
index d864bb6d..8a31fdb1 100644
--- a/testsuite/test_utils/stream_ciphers_test_case.cc
+++ b/testsuite/test_utils/stream_ciphers_test_case.cc
@@ -49,8 +49,9 @@ std::unique_ptr<stream> stream_cipher_test_case::prepare_stream() {
     _stream_config["block_size"] = block_size();
 
     seed_seq_from<pcg32> seeder(seed1);
+    std::unordered_map<std::string, std::shared_ptr<std::unique_ptr<stream>>> map;
 
-    return make_stream(_stream_config, seeder, block_size());
+    return make_stream(_stream_config, seeder, map, block_size());
 }
 
 void stream_cipher_test_case::test(