1 files changed, 132 insertions, 0 deletions
diff --git a/brillo/streams/input_stream_set.h b/brillo/streams/input_stream_set.h
new file mode 100644
index 0000000..fda255f
--- /dev/null
+++ b/brillo/streams/input_stream_set.h
@@ -0,0 +1,132 @@
+// Copyright 2015 The Chromium OS Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#ifndef LIBCHROMEOS_BRILLO_STREAMS_INPUT_STREAM_SET_H_
+#define LIBCHROMEOS_BRILLO_STREAMS_INPUT_STREAM_SET_H_
+
+#include <vector>
+
+#include <base/macros.h>
+#include <brillo/brillo_export.h>
+#include <brillo/streams/stream.h>
+
+namespace brillo {
+
+// Multiplexer stream allows to bundle a bunch of secondary streams in one
+// logical stream and simulate a read operation across data concatenated from
+// all those source streams.
+//
+// When created on a set of source streams like stream1, stream2, stream3, etc.,
+// reading from the multiplexer stream will read all the data from stream1 until
+// end-of-stream is reached, then keep reading from stream2, stream3 and so on.
+//
+// InputStreamSet has an option of owning the underlying source streams
+// or just referencing them. Owned streams are passed to InputStreamSet
+// with exclusive ownership transfer (using StreamPtr) and those streams will
+// be closed/destroyed when InputStreamSet is closed/destroyed.
+// Referenced source streams' life time is maintained elsewhere and they must
+// be valid for the duration of InputStreamSet's life. Closing the
+// muliplexer stream does not close the referenced streams.
+class BRILLO_EXPORT InputStreamSet : public Stream {
+ public:
+  // == Construction ==========================================================
+
+  // Generic method that constructs a multiplexer stream on a list of source
+  // streams. |source_streams| is the list of all source stream references
+  // in the order they need to be read from. |owned_source_streams| is a list
+  // of source stream instances that the multiplexer stream will own.
+  // Note that the streams from |owned_source_streams| should still be
+  // referenced in |source_streams| if you need their data to be read from.
+  // |owned_source_streams| could be empty (in which case none of the source
+  // streams are not owned), or contain fewer items than in |source_streams|.
+  static StreamPtr Create(std::vector<Stream*> source_streams,
+                          std::vector<StreamPtr> owned_source_streams,
+                          ErrorPtr* error);
+
+  // Simple helper method to create a multiplexer stream with a list of
+  // referenced streams. None of the streams will be owned.
+  // Effectively calls Create(source_streams, {}, error);
+  static StreamPtr Create(std::vector<Stream*> source_streams, ErrorPtr* error);
+
+  // Simple helper method to create a multiplexer stream with a list of
+  // referenced streams. None of the streams will be owned.
+  // Effectively calls Create(source_streams, owned_source_streams, error)
+  // with |source_streams| containing pointers to the streams from
+  // |owned_source_streams| list.
+  static StreamPtr Create(std::vector<StreamPtr> owned_source_streams,
+                          ErrorPtr* error);
+
+  // == Stream capabilities ===================================================
+  bool IsOpen() const override;
+  bool CanRead() const override { return true; }
+  bool CanWrite() const override { return false; }
+  bool CanSeek() const override { return false; }
+  bool CanGetSize() const override;
+
+  // == Stream size operations ================================================
+  uint64_t GetSize() const override;
+  bool SetSizeBlocking(uint64_t size, ErrorPtr* error) override;
+  uint64_t GetRemainingSize() const override;
+
+  // == Seek operations =======================================================
+  uint64_t GetPosition() const override { return 0; }
+  bool Seek(int64_t offset,
+            Whence whence,
+            uint64_t* new_position,
+            ErrorPtr* error) override;
+
+  // == Read operations =======================================================
+  bool ReadNonBlocking(void* buffer,
+                       size_t size_to_read,
+                       size_t* size_read,
+                       bool* end_of_stream,
+                       ErrorPtr* error) override;
+
+  // == Write operations ======================================================
+  bool WriteNonBlocking(const void* buffer,
+                        size_t size_to_write,
+                        size_t* size_written,
+                        ErrorPtr* error) override;
+
+  // == Finalizing/closing streams  ===========================================
+  bool FlushBlocking(ErrorPtr* error) override { return true; }
+  bool CloseBlocking(ErrorPtr* error) override;
+
+  // == Data availability monitoring ==========================================
+  bool WaitForData(AccessMode mode,
+                   const base::Callback<void(AccessMode)>& callback,
+                   ErrorPtr* error) override;
+
+  bool WaitForDataBlocking(AccessMode in_mode,
+                           base::TimeDelta timeout,
+                           AccessMode* out_mode,
+                           ErrorPtr* error) override;
+
+  void CancelPendingAsyncOperations() override;
+
+ private:
+  friend class InputStreamSetTest;
+
+  // Internal constructor used by the Create() factory methods.
+  InputStreamSet(std::vector<Stream*> source_streams,
+                 std::vector<StreamPtr> owned_source_streams,
+                 uint64_t initial_stream_size);
+
+  // List of streams to read data from.
+  std::vector<Stream*> source_streams_;
+
+  // List of source streams this stream owns. Owned source streams will be
+  // closed when InputStreamSet::CloseBlocking() is called and will be
+  // destroyed when this stream is destroyed.
+  std::vector<StreamPtr> owned_source_streams_;
+
+  uint64_t initial_stream_size_{0};
+  bool closed_{false};
+
+  DISALLOW_COPY_AND_ASSIGN(InputStreamSet);
+};
+
+}  // namespace brillo
+
+#endif  // LIBCHROMEOS_BRILLO_STREAMS_INPUT_STREAM_SET_H_