intermediate changes

ref:cde9a383711a11544ce7e107a78147fb96cc4029

1 files changed, 51 insertions, 0 deletions

diff --git a/library/cpp/yson_pull/detail/output/buffered.h b/library/cpp/yson_pull/detail/output/buffered.h
new file mode 100644
index 0000000000..475cf34785
--- /dev/null
+++ b/library/cpp/yson_pull/detail/output/buffered.h
@@ -0,0 +1,51 @@
+#pragma once
+
+#include <library/cpp/yson_pull/detail/macros.h>
+
+#include <library/cpp/yson_pull/output.h>
+
+#include <util/generic/strbuf.h>
+
+namespace NYsonPull {
+    namespace NDetail {
+        namespace NOutput {
+            template <typename T>
+            class TBuffered: public NYsonPull::NOutput::IStream {
+                TArrayHolder<ui8> buffer_;
+                size_t size_;
+
+            public:
+                TBuffered(size_t buffer_size)
+                    : buffer_{new ui8[buffer_size]}
+                    , size_{buffer_size} {
+                    reset_buffer();
+                }
+
+            protected:
+                void do_flush_buffer(TStringBuf extra) override {
+                    auto& buf = buffer();
+                    if (!buf.is_empty()) {
+                        do_write({reinterpret_cast<const char*>(buf.begin()), buf.used()});
+                        reset_buffer();
+                    }
+                    if (extra.size() >= buf.available()) {
+                        do_write(extra);
+                    } else if (extra.size() > 0) {
+                        ::memcpy(buf.pos(), extra.data(), extra.size());
+                        buf.advance(extra.size());
+                    }
+                }
+
+            private:
+                void do_write(TStringBuf data) {
+                    // CRTP dispatch
+                    static_cast<T*>(this)->write(data);
+                }
+
+                void reset_buffer() {
+                    buffer().reset(buffer_.Get(), buffer_.Get() + size_);
+                }
+            };
+        }
+    }     // namespace NDetail
+}