2018-12-31 20:04:05 +01:00
|
|
|
//
|
2020-01-01 02:23:48 +01:00
|
|
|
// Copyright Aliaksei Levin (levlam@telegram.org), Arseny Smirnov (arseny30@gmail.com) 2014-2020
|
2018-12-31 20:04:05 +01:00
|
|
|
//
|
|
|
|
// Distributed under the Boost Software License, Version 1.0. (See accompanying
|
|
|
|
// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
|
|
|
|
//
|
|
|
|
#pragma once
|
|
|
|
|
|
|
|
#include "td/utils/buffer.h"
|
|
|
|
#include "td/utils/common.h"
|
|
|
|
#include "td/utils/Status.h"
|
2020-07-22 20:52:00 +02:00
|
|
|
#include <limits>
|
2018-12-31 20:04:05 +01:00
|
|
|
|
|
|
|
namespace td {
|
|
|
|
|
|
|
|
class ByteFlowInterface {
|
|
|
|
public:
|
|
|
|
virtual void close_input(Status status) = 0;
|
|
|
|
virtual void wakeup() = 0;
|
|
|
|
virtual void set_parent(ByteFlowInterface &other) = 0;
|
|
|
|
virtual void set_input(ChainBufferReader *input) = 0;
|
|
|
|
virtual size_t get_need_size() = 0;
|
2020-07-22 20:52:00 +02:00
|
|
|
virtual size_t get_read_size() = 0;
|
|
|
|
virtual size_t get_write_size() = 0;
|
2018-12-31 20:04:05 +01:00
|
|
|
ByteFlowInterface() = default;
|
|
|
|
ByteFlowInterface(const ByteFlowInterface &) = delete;
|
|
|
|
ByteFlowInterface &operator=(const ByteFlowInterface &) = delete;
|
|
|
|
ByteFlowInterface(ByteFlowInterface &&) = default;
|
|
|
|
ByteFlowInterface &operator=(ByteFlowInterface &&) = default;
|
|
|
|
virtual ~ByteFlowInterface() = default;
|
|
|
|
};
|
|
|
|
|
|
|
|
class ByteFlowBaseCommon : public ByteFlowInterface {
|
|
|
|
public:
|
|
|
|
ByteFlowBaseCommon() = default;
|
|
|
|
|
|
|
|
void close_input(Status status) final {
|
|
|
|
if (status.is_error()) {
|
|
|
|
finish(std::move(status));
|
|
|
|
} else {
|
|
|
|
is_input_active_ = false;
|
|
|
|
wakeup();
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
void wakeup() final {
|
|
|
|
if (stop_flag_) {
|
|
|
|
return;
|
|
|
|
}
|
|
|
|
input_->sync_with_writer();
|
2020-07-22 20:52:00 +02:00
|
|
|
|
2018-12-31 20:04:05 +01:00
|
|
|
if (waiting_flag_) {
|
|
|
|
if (!is_input_active_) {
|
|
|
|
finish(Status::OK());
|
|
|
|
}
|
|
|
|
return;
|
|
|
|
}
|
2020-07-22 20:52:00 +02:00
|
|
|
while (true) {
|
|
|
|
if (stop_flag_) {
|
|
|
|
break;
|
|
|
|
}
|
|
|
|
|
|
|
|
// update can_read
|
|
|
|
if (is_input_active_) {
|
|
|
|
auto read_size = get_read_size();
|
|
|
|
if (read_size < min(need_size_, options_.read_watermark.low)) {
|
|
|
|
can_read = false;
|
|
|
|
}
|
|
|
|
if (read_size >= max(need_size_, options_.read_watermark.hight)) {
|
|
|
|
can_read = true;
|
|
|
|
}
|
|
|
|
|
|
|
|
} else {
|
|
|
|
//Alway can read when input is closed
|
|
|
|
can_read = true;
|
|
|
|
}
|
|
|
|
|
|
|
|
// update can_write
|
|
|
|
{
|
|
|
|
auto write_size = get_write_size();
|
|
|
|
if (write_size > options_.write_watermark.hight) {
|
|
|
|
can_write = false;
|
|
|
|
}
|
|
|
|
if (write_size <= options_.write_watermark.low) {
|
|
|
|
can_write = true;
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
if (!can_read || !can_write) {
|
|
|
|
break;
|
|
|
|
}
|
|
|
|
need_size_ = 0;
|
|
|
|
|
|
|
|
if (!loop()) {
|
|
|
|
if (need_size_ <= get_read_size()) {
|
|
|
|
need_size_ = get_read_size() + 1;
|
|
|
|
}
|
2018-12-31 20:04:05 +01:00
|
|
|
}
|
|
|
|
}
|
2020-07-22 20:52:00 +02:00
|
|
|
on_output_updated();
|
2018-12-31 20:04:05 +01:00
|
|
|
}
|
|
|
|
|
|
|
|
size_t get_need_size() final {
|
|
|
|
return need_size_;
|
|
|
|
}
|
2020-07-22 20:52:00 +02:00
|
|
|
size_t get_read_size() override {
|
|
|
|
input_->sync_with_writer();
|
|
|
|
return input_->size();
|
|
|
|
}
|
|
|
|
size_t get_write_size() override {
|
|
|
|
CHECK(parent_);
|
|
|
|
return parent_->get_read_size();
|
|
|
|
}
|
|
|
|
|
|
|
|
struct Watermark {
|
|
|
|
size_t low{std::numeric_limits<size_t>::max()};
|
|
|
|
size_t hight{0};
|
|
|
|
};
|
|
|
|
struct Options {
|
|
|
|
Watermark write_watermark;
|
|
|
|
Watermark read_watermark;
|
|
|
|
};
|
|
|
|
void set_options(Options options) {
|
|
|
|
options_ = options;
|
|
|
|
}
|
2018-12-31 20:04:05 +01:00
|
|
|
|
2020-07-22 20:52:00 +02:00
|
|
|
virtual bool loop() = 0;
|
2018-12-31 20:04:05 +01:00
|
|
|
|
|
|
|
protected:
|
|
|
|
bool waiting_flag_ = false;
|
2018-10-26 16:11:20 +02:00
|
|
|
ChainBufferReader *input_ = nullptr;
|
2018-12-31 20:04:05 +01:00
|
|
|
bool is_input_active_ = true;
|
|
|
|
size_t need_size_ = 0;
|
2020-07-22 20:52:00 +02:00
|
|
|
bool can_read{true};
|
|
|
|
bool can_write{true};
|
|
|
|
Options options_;
|
2018-12-31 20:04:05 +01:00
|
|
|
void finish(Status status) {
|
|
|
|
stop_flag_ = true;
|
|
|
|
need_size_ = 0;
|
|
|
|
if (parent_) {
|
|
|
|
parent_->close_input(std::move(status));
|
|
|
|
parent_ = nullptr;
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
void set_need_size(size_t need_size) {
|
|
|
|
need_size_ = need_size;
|
|
|
|
}
|
|
|
|
|
|
|
|
void on_output_updated() {
|
|
|
|
if (parent_) {
|
|
|
|
parent_->wakeup();
|
|
|
|
}
|
|
|
|
}
|
|
|
|
void consume_input() {
|
|
|
|
waiting_flag_ = true;
|
|
|
|
if (!is_input_active_) {
|
|
|
|
finish(Status::OK());
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
private:
|
|
|
|
ByteFlowInterface *parent_ = nullptr;
|
|
|
|
bool stop_flag_ = false;
|
|
|
|
friend class ByteFlowBase;
|
|
|
|
friend class ByteFlowInplaceBase;
|
|
|
|
};
|
|
|
|
|
|
|
|
class ByteFlowBase : public ByteFlowBaseCommon {
|
|
|
|
public:
|
|
|
|
ByteFlowBase() = default;
|
|
|
|
|
|
|
|
void set_input(ChainBufferReader *input) final {
|
|
|
|
input_ = input;
|
|
|
|
}
|
|
|
|
void set_parent(ByteFlowInterface &other) final {
|
|
|
|
parent_ = &other;
|
|
|
|
parent_->set_input(&output_reader_);
|
|
|
|
}
|
2020-07-22 20:52:00 +02:00
|
|
|
bool loop() override = 0;
|
2018-12-31 20:04:05 +01:00
|
|
|
|
|
|
|
// ChainBufferWriter &get_output() {
|
|
|
|
// return output_;
|
|
|
|
//}
|
|
|
|
|
|
|
|
protected:
|
|
|
|
ChainBufferWriter output_;
|
|
|
|
ChainBufferReader output_reader_ = output_.extract_reader();
|
|
|
|
};
|
|
|
|
|
|
|
|
class ByteFlowInplaceBase : public ByteFlowBaseCommon {
|
|
|
|
public:
|
|
|
|
ByteFlowInplaceBase() = default;
|
|
|
|
|
|
|
|
void set_input(ChainBufferReader *input) final {
|
|
|
|
input_ = input;
|
|
|
|
output_ = ChainBufferReader(input_->begin().clone(), input_->begin().clone(), false);
|
|
|
|
}
|
|
|
|
void set_parent(ByteFlowInterface &other) final {
|
|
|
|
parent_ = &other;
|
|
|
|
parent_->set_input(&output_);
|
|
|
|
}
|
2020-07-22 20:52:00 +02:00
|
|
|
bool loop() override = 0;
|
2018-12-31 20:04:05 +01:00
|
|
|
|
|
|
|
ChainBufferReader &get_output() {
|
|
|
|
return output_;
|
|
|
|
}
|
|
|
|
|
|
|
|
protected:
|
|
|
|
ChainBufferReader output_;
|
|
|
|
};
|
|
|
|
|
|
|
|
inline ByteFlowInterface &operator>>(ByteFlowInterface &from, ByteFlowInterface &to) {
|
|
|
|
from.set_parent(to);
|
|
|
|
return to;
|
|
|
|
}
|
|
|
|
|
|
|
|
class ByteFlowSource : public ByteFlowInterface {
|
|
|
|
public:
|
|
|
|
ByteFlowSource() = default;
|
|
|
|
explicit ByteFlowSource(ChainBufferReader *buffer) : buffer_(buffer) {
|
|
|
|
}
|
|
|
|
ByteFlowSource(ByteFlowSource &&other) : buffer_(other.buffer_), parent_(other.parent_) {
|
|
|
|
other.buffer_ = nullptr;
|
|
|
|
other.parent_ = nullptr;
|
|
|
|
}
|
|
|
|
ByteFlowSource &operator=(ByteFlowSource &&other) {
|
|
|
|
buffer_ = other.buffer_;
|
|
|
|
parent_ = other.parent_;
|
|
|
|
other.buffer_ = nullptr;
|
|
|
|
other.parent_ = nullptr;
|
|
|
|
return *this;
|
|
|
|
}
|
|
|
|
ByteFlowSource(const ByteFlowSource &) = delete;
|
|
|
|
ByteFlowSource &operator=(const ByteFlowSource &) = delete;
|
|
|
|
~ByteFlowSource() override = default;
|
|
|
|
|
|
|
|
void set_input(ChainBufferReader *) final {
|
|
|
|
UNREACHABLE();
|
|
|
|
}
|
|
|
|
void set_parent(ByteFlowInterface &parent) final {
|
|
|
|
CHECK(parent_ == nullptr);
|
|
|
|
parent_ = &parent;
|
|
|
|
parent_->set_input(buffer_);
|
|
|
|
}
|
|
|
|
void close_input(Status status) final {
|
|
|
|
CHECK(parent_);
|
|
|
|
parent_->close_input(std::move(status));
|
|
|
|
parent_ = nullptr;
|
|
|
|
}
|
|
|
|
void wakeup() final {
|
|
|
|
CHECK(parent_);
|
|
|
|
parent_->wakeup();
|
|
|
|
}
|
|
|
|
size_t get_need_size() final {
|
|
|
|
if (parent_ == nullptr) {
|
|
|
|
return 0;
|
|
|
|
}
|
|
|
|
return parent_->get_need_size();
|
|
|
|
}
|
2020-07-22 20:52:00 +02:00
|
|
|
size_t get_read_size() final {
|
|
|
|
UNREACHABLE();
|
|
|
|
return 0;
|
|
|
|
}
|
|
|
|
size_t get_write_size() final {
|
|
|
|
CHECK(parent_);
|
|
|
|
return parent_->get_read_size();
|
|
|
|
}
|
2018-12-31 20:04:05 +01:00
|
|
|
|
|
|
|
private:
|
|
|
|
ChainBufferReader *buffer_ = nullptr;
|
|
|
|
ByteFlowInterface *parent_ = nullptr;
|
|
|
|
};
|
|
|
|
|
|
|
|
class ByteFlowSink : public ByteFlowInterface {
|
|
|
|
public:
|
|
|
|
void set_input(ChainBufferReader *input) final {
|
|
|
|
CHECK(buffer_ == nullptr);
|
|
|
|
buffer_ = input;
|
|
|
|
}
|
2018-08-13 19:15:09 +02:00
|
|
|
void set_parent(ByteFlowInterface & /*parent*/) final {
|
2018-12-31 20:04:05 +01:00
|
|
|
UNREACHABLE();
|
|
|
|
}
|
|
|
|
void close_input(Status status) final {
|
|
|
|
CHECK(active_);
|
|
|
|
active_ = false;
|
|
|
|
status_ = std::move(status);
|
|
|
|
buffer_->sync_with_writer();
|
|
|
|
}
|
|
|
|
void wakeup() final {
|
|
|
|
buffer_->sync_with_writer();
|
|
|
|
}
|
|
|
|
size_t get_need_size() final {
|
|
|
|
UNREACHABLE();
|
|
|
|
return 0;
|
|
|
|
}
|
2020-07-22 20:52:00 +02:00
|
|
|
size_t get_read_size() final {
|
|
|
|
buffer_->sync_with_writer();
|
|
|
|
return buffer_->size();
|
|
|
|
}
|
|
|
|
size_t get_write_size() final {
|
|
|
|
UNREACHABLE();
|
|
|
|
return 0;
|
|
|
|
}
|
2018-12-31 20:04:05 +01:00
|
|
|
bool is_ready() {
|
|
|
|
return !active_;
|
|
|
|
}
|
|
|
|
Status &status() {
|
|
|
|
return status_;
|
|
|
|
}
|
|
|
|
ChainBufferReader *result() {
|
|
|
|
CHECK(is_ready() && status().is_ok());
|
|
|
|
return buffer_;
|
|
|
|
}
|
|
|
|
ChainBufferReader *get_output() {
|
|
|
|
return buffer_;
|
|
|
|
}
|
|
|
|
|
|
|
|
private:
|
|
|
|
bool active_ = true;
|
|
|
|
Status status_;
|
|
|
|
ChainBufferReader *buffer_ = nullptr;
|
|
|
|
};
|
|
|
|
|
|
|
|
class ByteFlowMoveSink : public ByteFlowInterface {
|
|
|
|
public:
|
2018-08-15 14:41:42 +02:00
|
|
|
ByteFlowMoveSink() = default;
|
2018-08-16 00:06:53 +02:00
|
|
|
explicit ByteFlowMoveSink(ChainBufferWriter *output) {
|
2018-08-15 14:41:42 +02:00
|
|
|
set_output(output);
|
|
|
|
}
|
2018-12-31 20:04:05 +01:00
|
|
|
void set_input(ChainBufferReader *input) final {
|
|
|
|
CHECK(!input_);
|
|
|
|
input_ = input;
|
|
|
|
}
|
2018-08-13 19:15:09 +02:00
|
|
|
void set_parent(ByteFlowInterface & /*parent*/) final {
|
2018-12-31 20:04:05 +01:00
|
|
|
UNREACHABLE();
|
|
|
|
}
|
|
|
|
void close_input(Status status) final {
|
|
|
|
CHECK(active_);
|
|
|
|
active_ = false;
|
|
|
|
status_ = std::move(status);
|
|
|
|
wakeup();
|
|
|
|
}
|
|
|
|
void wakeup() final {
|
|
|
|
input_->sync_with_writer();
|
|
|
|
output_->append(*input_);
|
|
|
|
}
|
|
|
|
size_t get_need_size() final {
|
|
|
|
UNREACHABLE();
|
|
|
|
return 0;
|
|
|
|
}
|
2020-07-22 20:52:00 +02:00
|
|
|
size_t get_read_size() final {
|
|
|
|
input_->sync_with_writer();
|
|
|
|
//TODO: must be input_->size() + output_->size()
|
|
|
|
return input_->size();
|
|
|
|
}
|
|
|
|
size_t get_write_size() final {
|
|
|
|
UNREACHABLE();
|
|
|
|
return 0;
|
|
|
|
}
|
2018-12-31 20:04:05 +01:00
|
|
|
void set_output(ChainBufferWriter *output) {
|
|
|
|
CHECK(!output_);
|
|
|
|
output_ = output;
|
|
|
|
}
|
|
|
|
|
|
|
|
bool is_ready() {
|
|
|
|
return !active_;
|
|
|
|
}
|
|
|
|
Status &status() {
|
|
|
|
return status_;
|
|
|
|
}
|
|
|
|
|
|
|
|
private:
|
|
|
|
bool active_ = true;
|
|
|
|
Status status_;
|
|
|
|
ChainBufferReader *input_ = nullptr;
|
|
|
|
ChainBufferWriter *output_ = nullptr;
|
|
|
|
};
|
|
|
|
} // namespace td
|