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 <functional>
20 : : #include <future>
21 : : #include <memory>
22 : : #include <optional>
23 : : #include <string_view>
24 : : #include <vector>
25 : :
26 : : #include "wirestead/base/visibility.hpp"
27 : : #include "wirestead/diagnostics/logger.hpp"
28 : : #include "wirestead/framer/iframer.hpp"
29 : : #include "wirestead/wrapper/context.hpp"
30 : : #include "wirestead/wrapper/fanout_result.hpp"
31 : : #include "wirestead/wrapper/runtime_stats.hpp"
32 : : #include "wirestead/wrapper/send_result.hpp"
33 : :
34 : : namespace wirestead {
35 : : namespace wrapper {
36 : :
37 : : /**
38 : : * @brief Interface for 1:N server communication (e.g., TcpServer)
39 : : */
40 : : class WIRESTEAD_API ServerInterface {
41 : : public:
42 : : using MessageHandler = std::function<void(const MessageContext&)>;
43 : : using BatchMessageHandler = std::function<void(const std::vector<MessageContext>&)>;
44 : : using ConnectionHandler = std::function<void(const ConnectionContext&)>;
45 : : using ErrorHandler = std::function<void(const ErrorContext&)>;
46 : : using FramerFactory = std::function<std::unique_ptr<framer::IFramer>()>;
47 : :
48 : 614 : virtual ~ServerInterface() = default;
49 : :
50 : : // Lifecycle
51 : : /**
52 : : * @brief Start the server asynchronously.
53 : : *
54 : : * @return A future that resolves to true when listening, or false on failure
55 : : * (e.g. bind failure, port already in use). This future always
56 : : * resolves - including on a restart after stop() (#444) - never
57 : : * blocks indefinitely.
58 : : */
59 : : [[nodiscard]] virtual std::future<bool> start() = 0;
60 : :
61 : : /**
62 : : * @brief Synchronously start the channel/server and wait for the result.
63 : : */
64 : 49 : [[nodiscard]] virtual bool start_sync() { return start().get(); }
65 : :
66 : : /**
67 : : * @brief Request server shutdown and wait when the calling thread permits it.
68 : : *
69 : : * Outside callers wait for shutdown completion, including concurrent callers
70 : : * that find a stop already in progress. No user callback remains running or can
71 : : * start from that run when a waiting stop returns. Repeated completed stops are no-ops.
72 : : *
73 : : * On an executor needed by shutdown (including this object's own callback),
74 : : * stop() only requests shutdown and returns without waiting. Such a return does
75 : : * not permit destruction, restart, or stopped-only configuration. An outside
76 : : * caller must observe shutdown completion first. Destroying the object from its
77 : : * own callback and concurrent destruction/use or start/stop are unsupported.
78 : : * Externally managed executors must keep making progress while callers wait;
79 : : * user callbacks must return. Internal handlers may remain if they own their
80 : : * lifetime and cannot affect a later run.
81 : : *
82 : : * Restart contract (#444): completed shutdown fully tears down the underlying transport
83 : : * (acceptor, sessions, timers) rather than leaving it in a reusable
84 : : * half-alive state. Every on_*() callback and every config setter ever
85 : : * called on this wrapper remains in force across any number of stop()/
86 : : * start() cycles - a subsequent start() reconstructs the transport from
87 : : * this wrapper's retained configuration and re-registers all callbacks
88 : : * automatically, including config changes made while stopped. Connected-
89 : : * client state and stats() reset on restart. Callers never need to
90 : : * re-register callbacks or reconfigure after a stop()/start() cycle.
91 : : */
92 : : virtual void stop() = 0;
93 : : virtual bool listening() const = 0;
94 : :
95 : : /**
96 : : * @brief Snapshot of this server's runtime counters.
97 : : *
98 : : * The cumulative fields (`bytes_*`, `messages_*`, `failed_sends`, `dropped_*`,
99 : : * `backpressure_events`, `max_queued_bytes`) cover every session this server
100 : : * has accepted, including ones that have since disconnected or stopped.
101 : : * Native TCP/UDS send_accounting also covers those contributors. During stop,
102 : : * retiring sessions remain in the aggregate until their final totals transfer.
103 : : * The
104 : : * instantaneous fields (`queued_bytes`, `pending_bytes`,
105 : : * `backpressure_active`) describe only the sessions that are live right now.
106 : : *
107 : : * Cleared by reset_stats(), and by a stop()/start() cycle per the restart
108 : : * contract above. See docs/error_model.md.
109 : : */
110 : : virtual RuntimeStats stats() const = 0;
111 : :
112 : : /**
113 : : * @brief Counters for one connected client.
114 : : *
115 : : * Returns nullopt when the id is unknown - including for a client that has
116 : : * already disconnected, since a session's counters are folded into the
117 : : * server-wide stats() and the session itself is gone. Sample this while the
118 : : * client is connected if you need its numbers in isolation.
119 : : *
120 : : * Built-in UDP servers expose per-virtual-session traffic and accounting.
121 : : * Their backpressure state/events describe shared socket capacity. Custom
122 : : * servers may return nullopt when per-client statistics are unsupported.
123 : : */
124 : 0 : virtual std::optional<RuntimeStats> client_stats(ClientId /*client_id*/) const { return std::nullopt; }
125 : :
126 : : virtual void reset_stats() = 0;
127 : :
128 : : // Transmission
129 : : //
130 : : // Strategy-aware API (recommended):
131 : : //
132 : : // send_to() / broadcast()
133 : : // Behaviour depends on the configured backpressure strategy:
134 : : // BestEffort — non-blocking; drops data when the target client's queue is full.
135 : : // Reliable — blocks the calling thread until queue pressure is relieved,
136 : : // then enqueues for send_to().
137 : : // broadcast() always performs non-blocking fan-out to connected clients.
138 : : // It does not wait for one slow client to relieve backpressure.
139 : : //
140 : : // Explicit API (escape hatch):
141 : : //
142 : : // try_send_to() / try_broadcast()
143 : : // Always non-blocking and always drops on full queue, regardless of strategy.
144 : : // Rejects payloads that would exceed the current non-blocking queue
145 : : // threshold; payloads below MAX_BUFFER_SIZE can still be rejected when
146 : : // they exceed the current backpressure high-water budget.
147 : : //
148 : : // send_to_blocking()
149 : : // Always blocks until queue pressure is relieved, regardless of strategy.
150 : : // This call may block indefinitely while the server remains active and
151 : : // backpressure does not clear. stop() from another thread is expected to
152 : : // unblock waiting senders.
153 : :
154 : : /**
155 : : * @brief Send to a specific client, honouring the backpressure strategy.
156 : : *
157 : : * BestEffort: non-blocking, drops if the client's send queue is full.
158 : : * Reliable: blocks until queue pressure is relieved, then enqueues.
159 : : *
160 : : * @return Local queue acceptance or a synchronous rejection reason; not delivery.
161 : : */
162 : : [[nodiscard]] virtual SendResult send_to(ClientId client_id, std::string_view data) = 0;
163 : :
164 : : /**
165 : : * @brief Send to all connected clients using non-blocking fan-out.
166 : : *
167 : : * Does not wait for slow clients to relieve backpressure. Use send_to_blocking()
168 : : * for per-client blocking queue admission.
169 : : *
170 : : * @return Counts over the fixed target set; empty() identifies zero targets.
171 : : */
172 : : [[nodiscard]] virtual FanoutResult broadcast(std::string_view data) = 0;
173 : :
174 : : /**
175 : : * @brief Block until queue pressure is relieved, then send to a client. Ignores strategy.
176 : : *
177 : : * This call may block indefinitely while the server remains active and
178 : : * backpressure does not clear. stop() from another thread is expected to unblock
179 : : * waiting senders.
180 : : *
181 : : * @return Local queue acceptance or a synchronous rejection reason. A stopped wait returns
182 : : * CancelledWhileWaiting; a lost target returns NotReady.
183 : : */
184 : : [[nodiscard]] virtual SendResult send_to_blocking(ClientId client_id, std::string_view data) = 0;
185 : :
186 : : /**
187 : : * @brief Non-blocking send_to that always drops on a full queue, ignoring strategy.
188 : : *
189 : : * Use as an escape hatch when you need drop-on-full behaviour on a Reliable channel.
190 : : * Rejects payloads that would exceed the current non-blocking queue threshold.
191 : : * A payload may be rejected even if it is below MAX_BUFFER_SIZE when it is larger
192 : : * than the current backpressure high-water budget. Use send_to() or
193 : : * send_to_blocking() when Reliable enqueue semantics are required for large
194 : : * payloads.
195 : : *
196 : : * @return Local queue acceptance or a synchronous rejection reason; not delivery.
197 : : */
198 : : [[nodiscard]] virtual SendResult try_send_to(ClientId client_id, std::string_view data) = 0;
199 : :
200 : : /**
201 : : * @brief Non-blocking broadcast that always drops on full queues, ignoring strategy.
202 : : *
203 : : * Uses the same non-blocking queue threshold policy as try_send_to().
204 : : *
205 : : * @return Counts over the fixed target set; empty() identifies zero targets.
206 : : */
207 : : [[nodiscard]] virtual FanoutResult try_broadcast(std::string_view data) = 0;
208 : :
209 : : /**
210 : : * @brief Send a line (data + "\n") to all clients without waiting for capacity.
211 : : */
212 : : [[nodiscard]] virtual FanoutResult broadcast_line(std::string_view line) = 0;
213 : :
214 : : /**
215 : : * @brief Send a line (data + "\n") to a specific client, honouring the backpressure strategy.
216 : : */
217 : : [[nodiscard]] virtual SendResult send_to_line(ClientId client_id, std::string_view line) = 0;
218 : :
219 : : /**
220 : : * @brief Non-blocking broadcast_line that always drops on a full queue, ignoring strategy.
221 : : */
222 : : [[nodiscard]] virtual FanoutResult try_broadcast_line(std::string_view line) = 0;
223 : :
224 : : /**
225 : : * @brief Non-blocking send_to_line that always drops on a full queue, ignoring strategy.
226 : : */
227 : : [[nodiscard]] virtual SendResult try_send_to_line(ClientId client_id, std::string_view line) = 0;
228 : :
229 : : // Event handlers
230 : :
231 : : /**
232 : : * @brief Fires when a connection is accepted.
233 : : *
234 : : * Accepted, not usable. On a TLS server this runs before the handshake, so a
235 : : * client that fails it still produces one on_connect - followed by an
236 : : * on_disconnect, with no on_data in between. The two callbacks stay paired
237 : : * either way, so connection bookkeeping keyed on them does not leak; treat
238 : : * the first byte of data, not this callback, as proof the peer got through.
239 : : */
240 : : virtual ServerInterface& on_connect(ConnectionHandler handler) = 0;
241 : : virtual ServerInterface& on_disconnect(ConnectionHandler handler) = 0;
242 : : virtual ServerInterface& on_data(MessageHandler handler) = 0;
243 : :
244 : : /** @brief Register a callback for batched data reception */
245 : : virtual ServerInterface& on_data_batch(BatchMessageHandler handler) = 0;
246 : :
247 : : virtual ServerInterface& on_error(ErrorHandler handler) = 0;
248 : :
249 : : /**
250 : : * @brief Register a callback to be notified when send queue congestion changes.
251 : : * @param handler Callback receiving the current number of queued bytes.
252 : : *
253 : : * Default implementation is a no-op. Concrete implementations that support
254 : : * backpressure reporting must override this method; otherwise the handler
255 : : * is silently discarded.
256 : : */
257 : 0 : virtual ServerInterface& on_backpressure(std::function<void(size_t)> handler) {
258 [ # # ]: 0 : if (handler) {
259 : 0 : WIRESTEAD_LOG_WARNING("server_interface", "on_backpressure",
260 : : "Backpressure reporting is not supported by this server; handler is discarded.");
261 : : }
262 : 0 : return *this;
263 : : }
264 : :
265 : : /**
266 : : * @brief Set a factory function to create a new framer for each client connection.
267 : : * @param factory Function that returns a unique_ptr to a new framer.
268 : : */
269 : : virtual ServerInterface& framer(FramerFactory factory) = 0;
270 : :
271 : : /**
272 : : * @brief Set a handler for complete messages extracted by the framer.
273 : : * @param handler callback taking MessageContext (where data is the framed payload).
274 : : */
275 : : virtual ServerInterface& on_message(MessageHandler handler) = 0;
276 : :
277 : : /** @brief Register a callback for batched framed message reception */
278 : : virtual ServerInterface& on_message_batch(BatchMessageHandler handler) = 0;
279 : :
280 : : // Management
281 : : virtual ServerInterface& auto_start(bool manage = true) = 0;
282 : : virtual size_t client_count() const = 0;
283 : : virtual std::vector<ClientId> connected_clients() const = 0;
284 : : };
285 : :
286 : : } // namespace wrapper
287 : : } // namespace wirestead
|
