Branch data Line data Source code
1 : : /*
2 : : * Copyright 2025 Jinwoo Sung
3 : : *
4 : : * Licensed under the Apache License, Version 2.0 (the "License");
5 : : * you may not use this file except in compliance with the License.
6 : : * You may obtain a copy of the License at
7 : : *
8 : : * http://www.apache.org/licenses/LICENSE-2.0
9 : : *
10 : : * Unless required by applicable law or agreed to in writing, software
11 : : * distributed under the License is distributed on an "AS IS" BASIS,
12 : : * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 : : * See the License for the specific language governing permissions and
14 : : * limitations under the License.
15 : : */
16 : :
17 : : #pragma once
18 : :
19 : : #include <memory>
20 : : #include <optional>
21 : : #include <string>
22 : : #include <string_view>
23 : : #include <vector>
24 : :
25 : : #include "wirestead/base/visibility.hpp"
26 : : #include "wirestead/config/tcp_server_config.hpp"
27 : : #include "wirestead/diagnostics/error_types.hpp"
28 : : #include "wirestead/interface/channel.hpp"
29 : : #include "wirestead/wrapper/fanout_result.hpp"
30 : : #include "wirestead/wrapper/send_result.hpp"
31 : :
32 : : namespace boost {
33 : : namespace asio {
34 : : class io_context;
35 : : }
36 : : } // namespace boost
37 : :
38 : : namespace wirestead {
39 : :
40 : : namespace interface {
41 : : class TcpAcceptorInterface;
42 : : }
43 : :
44 : : namespace wrapper {
45 : : class TcpServer;
46 : : }
47 : : namespace transport {
48 : : class TcpServerSession;
49 : :
50 : : /**
51 : : * @brief Thread-safe TCP Server implementation
52 : : */
53 : : class WIRESTEAD_API TcpServer : public interface::Channel, public std::enable_shared_from_this<TcpServer> {
54 : : public:
55 : : // use_shared_context: opt into the shared IoContextManager singleton
56 : : // instead of the default dedicated io_context + thread. Only meaningful
57 : : // for multi-server-per-process deployments deliberately trading
58 : : // parallelism for reduced thread/memory overhead (#440); most callers
59 : : // should leave this false.
60 : : static std::shared_ptr<TcpServer> create(const config::TcpServerConfig& cfg, bool use_shared_context = false);
61 : : static std::shared_ptr<TcpServer> create(const config::TcpServerConfig& cfg,
62 : : std::unique_ptr<interface::TcpAcceptorInterface> acceptor,
63 : : boost::asio::io_context& ioc);
64 : : ~TcpServer() override;
65 : :
66 : : // Move semantics
67 : : TcpServer(TcpServer&&) noexcept;
68 : : TcpServer& operator=(TcpServer&&) noexcept;
69 : :
70 : : // Non-copyable
71 : : TcpServer(const TcpServer&) = delete;
72 : : TcpServer& operator=(const TcpServer&) = delete;
73 : :
74 : : // Channel implementation
75 : : void start() override;
76 : : void stop() override;
77 : : bool is_connected() const override;
78 : : bool is_backpressure_active() const override;
79 : : bool is_backpressure_active(ClientId client_id) const;
80 : : // Per-session hard queue limit; nullopt if the client ID is absent.
81 : : std::optional<size_t> write_queue_limit(ClientId client_id) const;
82 : : using interface::Channel::write_queue_limit;
83 : : boost::asio::any_io_executor get_executor() override;
84 : : bool async_write_copy(memory::ConstByteSpan data) override;
85 : : bool async_write_move(std::vector<uint8_t>&& data) override;
86 : : bool async_write_shared(std::shared_ptr<const std::vector<uint8_t>> data) override;
87 : : bool async_try_write_copy(memory::ConstByteSpan data) override;
88 : : bool async_try_write_move(std::vector<uint8_t>&& data) override;
89 : : bool async_try_write_shared(std::shared_ptr<const std::vector<uint8_t>> data) override;
90 : : void on_bytes(OnBytes cb) override;
91 : : void on_state(OnState cb) override;
92 : : void on_backpressure(OnBackpressure cb) override;
93 : : wrapper::RuntimeStats stats() const override;
94 : : void reset_stats() override;
95 : : std::optional<diagnostics::ErrorInfo> last_error_info() const override;
96 : :
97 : : // Multi-client support
98 : : bool broadcast(std::string_view message);
99 : : bool broadcast(memory::ConstByteSpan data);
100 : : bool send_to_client(ClientId client_id, std::string_view message);
101 : : bool send_to_client(ClientId client_id, memory::ConstByteSpan data);
102 : : bool try_send_to_client(ClientId client_id, std::string_view message);
103 : : bool try_send_to_client(ClientId client_id, memory::ConstByteSpan data);
104 : : size_t client_count() const;
105 : : std::vector<ClientId> connected_clients() const;
106 : : std::optional<wrapper::RuntimeStats> client_stats(ClientId client_id) const;
107 : :
108 : : void request_stop();
109 : :
110 : : using MultiClientConnectHandler = std::function<void(ClientId client_id, const std::string& client_info)>;
111 : : using MultiClientDataHandler = std::function<void(ClientId client_id, memory::ConstByteSpan data)>;
112 : : using MultiClientDisconnectHandler = std::function<void(ClientId client_id)>;
113 : :
114 : : void on_multi_connect(MultiClientConnectHandler handler);
115 : : void on_multi_data(MultiClientDataHandler handler);
116 : : void on_multi_disconnect(MultiClientDisconnectHandler handler);
117 : :
118 : : void set_client_limit(size_t max_clients);
119 : :
120 : : base::LinkState state() const;
121 : :
122 : : private:
123 : : friend class wrapper::TcpServer;
124 : : void fail_receive(ClientId id);
125 : : wrapper::FanoutResult broadcast_result(memory::ConstByteSpan data, wrapper::SendResult wrapper_state,
126 : : bool append_newline);
127 : : wrapper::SendResult target_state() const;
128 : : std::shared_ptr<TcpServerSession> capture_target(ClientId client_id) const;
129 : : std::optional<boost::asio::any_io_executor> client_executor(ClientId client_id) const;
130 : : std::optional<wrapper::SendResult> poll_target_wait(const std::shared_ptr<TcpServerSession>& session) const;
131 : : void cancel_target_waits();
132 : : wrapper::SendResult write_target(ClientId client_id, memory::ConstByteSpan data, bool try_only,
133 : : const std::shared_ptr<TcpServerSession>& expected = {});
134 : : explicit TcpServer(const config::TcpServerConfig& cfg, bool use_shared_context);
135 : : TcpServer(const config::TcpServerConfig& cfg, std::unique_ptr<interface::TcpAcceptorInterface> acceptor,
136 : : boost::asio::io_context& ioc);
137 : :
138 : : struct Impl;
139 : 2954 : const Impl* get_impl() const { return impl_.get(); }
140 : 10324 : Impl* get_impl() { return impl_.get(); }
141 : : std::unique_ptr<Impl> impl_;
142 : : };
143 : : } // namespace transport
144 : : } // namespace wirestead
|
