diff options
author | Anna Henningsen <anna@addaleax.net> | 2018-02-13 03:23:50 +0300 |
---|---|---|
committer | Anna Henningsen <anna@addaleax.net> | 2018-03-15 14:53:13 +0300 |
commit | 67f1d76956a8a5da9875b113371c8786ad579086 (patch) | |
tree | 220df0ab56ebc499c1edec20fe7b90449189d44b /src/stream_pipe.h | |
parent | f7f1437d44f3e4b745e36540a752065cc58d993b (diff) |
src: introduce native-layer stream piping
Provide a way to create pipes between native `StreamBase` instances
that acts more directly than a `.pipe()` call would.
PR-URL: https://github.com/nodejs/node/pull/18936
Reviewed-By: James M Snell <jasnell@gmail.com>
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
Diffstat (limited to 'src/stream_pipe.h')
-rw-r--r-- | src/stream_pipe.h | 68 |
1 files changed, 68 insertions, 0 deletions
diff --git a/src/stream_pipe.h b/src/stream_pipe.h new file mode 100644 index 00000000000..98d6dae11be --- /dev/null +++ b/src/stream_pipe.h @@ -0,0 +1,68 @@ +#ifndef SRC_STREAM_PIPE_H_ +#define SRC_STREAM_PIPE_H_ + +#if defined(NODE_WANT_INTERNALS) && NODE_WANT_INTERNALS + +#include "stream_base.h" + +namespace node { + +class StreamPipe : public AsyncWrap { + public: + StreamPipe(StreamBase* source, StreamBase* sink, v8::Local<v8::Object> obj); + ~StreamPipe(); + + void Unpipe(); + + static void New(const v8::FunctionCallbackInfo<v8::Value>& args); + static void Start(const v8::FunctionCallbackInfo<v8::Value>& args); + static void Unpipe(const v8::FunctionCallbackInfo<v8::Value>& args); + + size_t self_size() const override { return sizeof(*this); } + + private: + StreamBase* source(); + StreamBase* sink(); + + void ShutdownWritable(); + void FlushToWritable(); + + bool is_reading_ = false; + bool is_writing_ = false; + bool is_eof_ = false; + bool is_closed_ = true; + + // Set a default value so that when we’re coming from Start(), we know + // that we don’t want to read just yet. + // This will likely need to be changed when supporting streams without + // `OnStreamWantsWrite()` support. + size_t wanted_data_ = 0; + + void ProcessData(size_t nread, const uv_buf_t& buf); + + class ReadableListener : public StreamListener { + public: + uv_buf_t OnStreamAlloc(size_t suggested_size) override; + void OnStreamRead(ssize_t nread, const uv_buf_t& buf) override; + void OnStreamDestroy() override; + }; + + class WritableListener : public StreamListener { + public: + uv_buf_t OnStreamAlloc(size_t suggested_size) override; + void OnStreamRead(ssize_t nread, const uv_buf_t& buf) override; + void OnStreamAfterWrite(WriteWrap* w, int status) override; + void OnStreamAfterShutdown(ShutdownWrap* w, int status) override; + void OnStreamWantsWrite(size_t suggested_size) override; + void OnStreamDestroy() override; + }; + + ReadableListener readable_listener_; + WritableListener writable_listener_; +}; + +} // namespace node + +#endif + +#endif // SRC_STREAM_PIPE_H_ |