Ton

1
/*
2
    This file is part of TON Blockchain Library.
3

4
    TON Blockchain Library is free software: you can redistribute it and/or modify
5
    it under the terms of the GNU Lesser General Public License as published by
6
    the Free Software Foundation, either version 2 of the License, or
7
    (at your option) any later version.
8

9
    TON Blockchain Library is distributed in the hope that it will be useful,
10
    but WITHOUT ANY WARRANTY; without even the implied warranty of
11
    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
12
    GNU Lesser General Public License for more details.
13

14
    You should have received a copy of the GNU Lesser General Public License
15
    along with TON Blockchain Library.  If not, see <http://www.gnu.org/licenses/>.
16

17
    Copyright 2017-2020 Telegram Systems LLP
18
*/
19
#include "FileToStreamActor.h"
20

21
namespace td {
22
FileToStreamActor::FileToStreamActor(FileFd fd, StreamWriter writer, Options options)
23
    : fd_(std::move(fd)), writer_(std::move(writer)), options_(options) {
24
}
25

26
void FileToStreamActor::set_callback(td::unique_ptr<Callback> callback) {
27
  callback_ = std::move(callback);
28
  got_more();
29
}
30

31
void FileToStreamActor::got_more() {
32
  if (!callback_) {
33
    return;
34
  }
35
  callback_->got_more();
36
}
37
void FileToStreamActor::loop() {
38
  auto dest = writer_.prepare_write();
39
  if (options_.limit != -1) {
40
    if (static_cast<int64>(dest.size()) > options_.limit) {
41
      dest.truncate(narrow_cast<size_t>(options_.limit));
42
    }
43
  }
44
  if (dest.empty()) {
45
    //NB: Owner of CyclicBufer::Reader should notify this actor after each chunk is readed
46
    return;
47
  }
48

49
  auto r_size = fd_.read(dest);
50
  if (r_size.is_error()) {
51
    writer_.close_writer(r_size.move_as_error());
52
    got_more();
53
    return stop();
54
  }
55
  auto size = r_size.move_as_ok();
56
  writer_.confirm_write(size);
57
  got_more();
58
  if (options_.limit != -1) {
59
    options_.limit -= narrow_cast<int64>(size);
60
  }
61
  if (options_.limit == 0) {
62
    writer_.close_writer(td::Status::OK());
63
    got_more();
64
    return stop();
65
  }
66
  if (size == 0) {
67
    if (options_.read_tail_each < 0) {
68
      writer_.close_writer(td::Status::OK());
69
      got_more();
70
      return stop();
71
    }
72
    alarm_timestamp() = Timestamp::in(options_.read_tail_each);
73
    return;
74
  }
75
  yield();
76
}
77

78
}  // namespace td
79