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 <boost/asio/ip/udp.hpp>
20 : : #include <functional>
21 : : #include <memory>
22 : : #include <optional>
23 : : #include <vector>
24 : :
25 : : #include "wirestead/base/visibility.hpp"
26 : : #include "wirestead/config/udp_config.hpp"
27 : : #include "wirestead/diagnostics/error_types.hpp"
28 : : #include "wirestead/interface/result_channel.hpp"
29 : : #include "wirestead/wrapper/send_result.hpp"
30 : :
31 : : namespace boost {
32 : : namespace asio {
33 : : class io_context;
34 : : }
35 : : } // namespace boost
36 : :
37 : : namespace wirestead {
38 : : namespace wrapper {
39 : : class UdpClient;
40 : : class UdpServer;
41 : : } // namespace wrapper
42 : : namespace transport {
43 : : namespace detail {
44 : : struct UdpWriteWait;
45 : : }
46 : :
47 : : /**
48 : : * @brief UDP Transport implementation with 1:N support
49 : : */
50 : : class WIRESTEAD_API UdpChannel : public interface::ResultChannel, public std::enable_shared_from_this<UdpChannel> {
51 : : public:
52 : : using OnBytesFrom = std::function<void(memory::ConstByteSpan, const boost::asio::ip::udp::endpoint&)>;
53 : :
54 : : static std::shared_ptr<UdpChannel> create(const config::UdpConfig& cfg);
55 : : static std::shared_ptr<UdpChannel> create(const config::UdpConfig& cfg, boost::asio::io_context& ioc);
56 : : ~UdpChannel() override;
57 : :
58 : : // Move semantics
59 : : UdpChannel(UdpChannel&&) noexcept;
60 : : UdpChannel& operator=(UdpChannel&&) noexcept;
61 : :
62 : : // Non-copyable
63 : : UdpChannel(const UdpChannel&) = delete;
64 : : UdpChannel& operator=(const UdpChannel&) = delete;
65 : :
66 : : // Channel implementation
67 : : void start() override;
68 : : void stop() override;
69 : : bool is_connected() const override;
70 : : bool is_backpressure_active() const override;
71 : : std::optional<size_t> write_queue_limit() const override;
72 : : wrapper::RuntimeStats stats() const override;
73 : : void reset_stats() override;
74 : : std::optional<diagnostics::ErrorInfo> last_error_info() const override;
75 : :
76 : : // 1:1 writes (using configured remote_endpoint_)
77 : : [[nodiscard]] wrapper::SendResult async_write_copy_result(memory::ConstByteSpan data) override;
78 : : [[nodiscard]] wrapper::SendResult async_write_move_result(std::vector<uint8_t>&& data) override;
79 : : [[nodiscard]] wrapper::SendResult async_write_shared_result(
80 : : std::shared_ptr<const std::vector<uint8_t>> data) override;
81 : : [[nodiscard]] wrapper::SendResult async_try_write_copy_result(memory::ConstByteSpan data) override;
82 : : [[nodiscard]] wrapper::SendResult async_try_write_move_result(std::vector<uint8_t>&& data) override;
83 : : [[nodiscard]] wrapper::SendResult async_try_write_shared_result(
84 : : std::shared_ptr<const std::vector<uint8_t>> data) override;
85 : :
86 : : // 1:N writes (explicit destination)
87 : : virtual bool async_write_to(memory::ConstByteSpan data, const boost::asio::ip::udp::endpoint& destination);
88 : : virtual bool async_try_write_to(memory::ConstByteSpan data, const boost::asio::ip::udp::endpoint& destination);
89 : :
90 : : // Callbacks
91 : : void on_bytes(OnBytes cb) override;
92 : : virtual void on_bytes_from(OnBytesFrom cb);
93 : : void on_state(OnState cb) override;
94 : : void on_backpressure(OnBackpressure cb) override;
95 : :
96 : : void set_backpressure_strategy(base::constants::BackpressureStrategy strategy);
97 : :
98 : : /**
99 : : * @brief Get the local endpoint the socket is bound to.
100 : : */
101 : : boost::asio::ip::udp::endpoint local_endpoint() const;
102 : :
103 : : /**
104 : : * @brief Get the strand shared by channel I/O and wrapper timers.
105 : : */
106 : : boost::asio::any_io_executor get_executor() override;
107 : :
108 : : private:
109 : : friend class wrapper::UdpClient;
110 : : friend class wrapper::UdpServer;
111 : : std::shared_ptr<detail::UdpWriteWait> capture_write_wait(bool require_remote = true, bool track_session = false);
112 : : std::optional<wrapper::SendResult> poll_write_wait(const std::shared_ptr<detail::UdpWriteWait>& wait);
113 : : void end_write_wait(const std::shared_ptr<detail::UdpWriteWait>& wait, wrapper::SendRejection reason);
114 : : void cancel_write_waits();
115 : : void expire_session(const std::shared_ptr<detail::UdpWriteWait>& wait);
116 : : wrapper::RuntimeStats session_stats(const std::shared_ptr<detail::UdpWriteWait>& wait) const;
117 : : std::optional<uint64_t> write_connection() const;
118 : : wrapper::SendResult write_state(bool require_remote = true);
119 : : wrapper::SendResult write_copy(memory::ConstByteSpan data, std::optional<uint64_t> expected_run = std::nullopt);
120 : : wrapper::SendResult try_write_copy(memory::ConstByteSpan data, std::optional<uint64_t> expected_run = std::nullopt);
121 : : wrapper::SendResult write_move(std::vector<uint8_t>&& data, std::optional<uint64_t> expected_run = std::nullopt);
122 : : wrapper::SendResult try_write_move(std::vector<uint8_t>&& data, std::optional<uint64_t> expected_run = std::nullopt);
123 : : wrapper::SendResult write_shared(std::shared_ptr<const std::vector<uint8_t>> data,
124 : : std::optional<uint64_t> expected_run = std::nullopt);
125 : : wrapper::SendResult try_write_shared(std::shared_ptr<const std::vector<uint8_t>> data,
126 : : std::optional<uint64_t> expected_run = std::nullopt);
127 : : wrapper::SendResult write_to(memory::ConstByteSpan data, const boost::asio::ip::udp::endpoint& destination,
128 : : std::optional<uint64_t> expected_run = std::nullopt,
129 : : std::shared_ptr<detail::UdpWriteWait> session = {});
130 : : wrapper::SendResult try_write_to(memory::ConstByteSpan data, const boost::asio::ip::udp::endpoint& destination,
131 : : std::optional<uint64_t> expected_run = std::nullopt,
132 : : std::shared_ptr<detail::UdpWriteWait> session = {});
133 : : explicit UdpChannel(const config::UdpConfig& cfg);
134 : : UdpChannel(const config::UdpConfig& cfg, boost::asio::io_context& ioc);
135 : :
136 : : struct Impl;
137 : 5271 : const Impl* get_impl() const { return impl_.get(); }
138 : 554511 : Impl* get_impl() { return impl_.get(); }
139 : : std::unique_ptr<Impl> impl_;
140 : : };
141 : :
142 : : } // namespace transport
143 : : } // namespace wirestead
|
