1 //
2 // Copyright Aliaksei Levin (levlam@telegram.org), Arseny Smirnov (arseny30@gmail.com) 2014-2021
3 //
4 // Distributed under the Boost Software License, Version 1.0. (See accompanying
5 // file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
6 //
7 #include "td/net/HttpHeaderCreator.h"
8 #include "td/net/TcpListener.h"
9
10 #include "td/actor/actor.h"
11 #include "td/actor/ConcurrentScheduler.h"
12
13 #include "td/utils/buffer.h"
14 #include "td/utils/logging.h"
15 #include "td/utils/port/detail/PollableFd.h"
16 #include "td/utils/port/SocketFd.h"
17 #include "td/utils/Slice.h"
18 #include "td/utils/Status.h"
19
20 #include <array>
21
22 static int cnt = 0;
23
24 class HelloWorld final : public td::Actor {
25 public:
HelloWorld(td::SocketFd socket_fd)26 explicit HelloWorld(td::SocketFd socket_fd) : socket_fd_(std::move(socket_fd)) {
27 }
28
29 private:
30 td::SocketFd socket_fd_;
31
32 std::array<char, 1024> read_buf;
33 size_t read_new_lines{0};
34
35 td::string hello_;
36 td::string write_buf_;
37 size_t write_pos_{0};
38
start_up()39 void start_up() final {
40 td::Scheduler::subscribe(socket_fd_.get_poll_info().extract_pollable_fd(this));
41 td::HttpHeaderCreator hc;
42 td::Slice content = "hello world";
43 //auto content = td::BufferSlice("hello world");
44 hc.init_ok();
45 hc.set_keep_alive();
46 hc.set_content_size(content.size());
47 hc.add_header("Server", "TDLib/test");
48 hc.add_header("Date", "Thu Dec 14 01:41:50 2017");
49 hc.add_header("Content-Type:", "text/html");
50 hello_ = hc.finish(content).ok().str();
51 }
52
loop()53 void loop() final {
54 auto status = do_loop();
55 if (status.is_error()) {
56 td::Scheduler::unsubscribe(socket_fd_.get_poll_info().get_pollable_fd_ref());
57 stop();
58 LOG(ERROR) << "CLOSE: " << status;
59 }
60 }
do_loop()61 td::Status do_loop() {
62 sync_with_poll(socket_fd_);
63 TRY_STATUS(read_loop());
64 TRY_STATUS(write_loop());
65 if (can_close_local(socket_fd_)) {
66 return td::Status::Error("CLOSE");
67 }
68 return td::Status::OK();
69 }
write_loop()70 td::Status write_loop() {
71 while (can_write_local(socket_fd_) && write_pos_ < write_buf_.size()) {
72 TRY_RESULT(written, socket_fd_.write(td::Slice(write_buf_).substr(write_pos_)));
73 write_pos_ += written;
74 if (write_pos_ == write_buf_.size()) {
75 write_pos_ = 0;
76 write_buf_.clear();
77 }
78 }
79 return td::Status::OK();
80 }
read_loop()81 td::Status read_loop() {
82 while (can_read_local(socket_fd_)) {
83 TRY_RESULT(read_size, socket_fd_.read(td::MutableSlice(read_buf.data(), read_buf.size())));
84 for (size_t i = 0; i < read_size; i++) {
85 if (read_buf[i] == '\n') {
86 read_new_lines++;
87 if (read_new_lines == 2) {
88 read_new_lines = 0;
89 write_buf_.append(hello_);
90 }
91 }
92 }
93 }
94 return td::Status::OK();
95 }
96 };
97
98 const int N = 0;
99 class Server final : public td::TcpListener::Callback {
100 public:
start_up()101 void start_up() final {
102 listener_ =
103 td::create_actor<td::TcpListener>("Listener", 8082, td::ActorOwn<td::TcpListener::Callback>(actor_id(this)));
104 }
accept(td::SocketFd fd)105 void accept(td::SocketFd fd) final {
106 LOG(ERROR) << "ACCEPT " << cnt++;
107 pos_++;
108 auto scheduler_id = pos_ % (N != 0 ? N : 1) + (N != 0);
109 td::create_actor_on_scheduler<HelloWorld>("HelloWorld", scheduler_id, std::move(fd)).release();
110 }
hangup()111 void hangup() final {
112 // may be it should be default?..
113 LOG(ERROR) << "Hanging up..";
114 stop();
115 }
116
117 private:
118 td::ActorOwn<td::TcpListener> listener_;
119 int pos_{0};
120 };
121
main()122 int main() {
123 SET_VERBOSITY_LEVEL(VERBOSITY_NAME(ERROR));
124 auto scheduler = td::make_unique<td::ConcurrentScheduler>();
125 scheduler->init(N);
126 scheduler->create_actor_unsafe<Server>(0, "Server").release();
127 scheduler->start();
128 while (scheduler->run_main(10)) {
129 // empty
130 }
131 scheduler->finish();
132 }
133