include/boost/corosio/local_stream_socket.hpp

100.0% Lines (47/47) 100.0% List of functions (15/15) 100.0% Branches (18/18)

[ ] prev/next file

local_stream_socket.hpp

f(x) Functions (15)

Function Calls Lines Branches Blocks

Line	Branch	Hits	Source Code
1			//
2			// Copyright (c) 2026 Michael Vandeberg
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			// Official repository: https://github.com/cppalliance/corosio
8			//
9
10			#ifndef BOOST_COROSIO_LOCAL_STREAM_SOCKET_HPP
11			#define BOOST_COROSIO_LOCAL_STREAM_SOCKET_HPP
12
13			#include <boost/corosio/detail/config.hpp>
14			#include <boost/corosio/detail/platform.hpp>
15			#include <boost/corosio/detail/except.hpp>
16			#include <boost/corosio/detail/native_handle.hpp>
17			#include <boost/corosio/detail/op_base.hpp>
18			#include <boost/corosio/io/io_stream.hpp>
19			#include <boost/capy/io_result.hpp>
20			#include <boost/corosio/detail/buffer_param.hpp>
21			#include <boost/corosio/local_endpoint.hpp>
22			#include <boost/corosio/local_stream.hpp>
23			#include <boost/corosio/shutdown_type.hpp>
24			#include <boost/corosio/wait_type.hpp>
25			#include <boost/capy/ex/executor_ref.hpp>
26			#include <boost/capy/ex/execution_context.hpp>
27			#include <boost/capy/ex/io_env.hpp>
28			#include <boost/capy/concept/executor.hpp>
29
30			#include <system_error>
31
32			#include <concepts>
33			#include <coroutine>
34			#include <cstddef>
35			#include <stop_token>
36			#include <type_traits>
37
38			namespace boost::corosio {
39
40			/** An asynchronous Unix stream socket for coroutine I/O.
41
42			This class provides asynchronous Unix domain stream socket
43			operations that return awaitable types. Each operation
44			participates in the affine awaitable protocol, ensuring
45			coroutines resume on the correct executor.
46
47			The socket must be opened before performing I/O operations.
48			Operations support cancellation through `std::stop_token` via
49			the affine protocol, or explicitly through the `cancel()`
50			member function.
51
52			@par Thread Safety
53			Distinct objects: Safe.@n
54			Shared objects: Unsafe. A socket must not have concurrent
55			operations of the same type (e.g., two simultaneous reads).
56			One read and one write may be in flight simultaneously.
57
58			@par Semantics
59			Wraps the platform Unix domain socket stack. Operations
60			dispatch to OS socket APIs via the io_context backend
61			(epoll, kqueue, select, or IOCP). Satisfies @ref capy::Stream.
62
63			@par Example
64			@code
65			io_context ioc;
66			local_stream_socket s(ioc);
67
68			auto [ec] = co_await s.connect(local_endpoint("/tmp/my.sock"));
69			if (ec)
70			co_return;
71
72			char buf[1024];
73			auto [read_ec, n] = co_await s.read_some(
74			capy::mutable_buffer(buf, sizeof(buf)));
75			@endcode
76			*/
77			class BOOST_COROSIO_DECL local_stream_socket : public io_stream
78			{
79			public:
80			/// The endpoint type used by this socket.
81			using endpoint_type = corosio::local_endpoint;
82
83			using shutdown_type = corosio::shutdown_type;
84			using enum corosio::shutdown_type;
85
86			/** Define backend hooks for local stream socket operations.
87
88			Platform backends (epoll, kqueue, select) derive from this
89			to implement socket I/O, connection, and option management.
90			*/
91			struct implementation : io_stream::implementation
92			{
93			/** Initiate an asynchronous connect to the given endpoint.
94
95			@param h Coroutine handle to resume on completion.
96			@param ex Executor for dispatching the completion.
97			@param ep The local endpoint (path) to connect to.
98			@param token Stop token for cancellation.
99			@param ec Output error code.
100
101			@return Coroutine handle to resume immediately.
102			*/
103			virtual std::coroutine_handle<> connect(
104			std::coroutine_handle<> h,
105			capy::executor_ref ex,
106			corosio::local_endpoint ep,
107			std::stop_token token,
108			std::error_code* ec) = 0;
109
110			/** Initiate an asynchronous wait for socket readiness.
111
112			Completes when the socket becomes ready for the
113			specified direction, or an error condition is
114			reported. No bytes are transferred.
115
116			@param h Coroutine handle to resume on completion.
117			@param ex Executor for dispatching the completion.
118			@param w The direction to wait on.
119			@param token Stop token for cancellation.
120			@param ec Output error code.
121
122			@return Coroutine handle to resume immediately.
123			*/
124			virtual std::coroutine_handle<> wait(
125			std::coroutine_handle<> h,
126			capy::executor_ref ex,
127			wait_type w,
128			std::stop_token token,
129			std::error_code* ec) = 0;
130
131			/** Shut down the socket for the given direction(s).
132
133			@param what The shutdown direction.
134
135			@return Error code on failure, empty on success.
136			*/
137			virtual std::error_code shutdown(shutdown_type what) noexcept = 0;
138
139			/// Return the platform socket descriptor.
140			virtual native_handle_type native_handle() const noexcept = 0;
141
142			/** Release ownership of the native socket handle.
143
144			Deregisters the socket from the reactor without closing
145			the descriptor. The caller takes ownership.
146
147			@return The native handle.
148			*/
149			virtual native_handle_type release_socket() noexcept = 0;
150
151			/** Request cancellation of pending asynchronous operations.
152
153			All outstanding operations complete with operation_canceled error.
154			Check `ec == cond::canceled` for portable comparison.
155			*/
156			virtual void cancel() noexcept = 0;
157
158			/** Set a socket option.
159
160			@param level The protocol level (e.g. `SOL_SOCKET`).
161			@param optname The option name (e.g. `SO_KEEPALIVE`).
162			@param data Pointer to the option value.
163			@param size Size of the option value in bytes.
164			@return Error code on failure, empty on success.
165			*/
166			virtual std::error_code set_option(
167			int level,
168			int optname,
169			void const* data,
170			std::size_t size) noexcept = 0;
171
172			/** Get a socket option.
173
174			@param level The protocol level (e.g. `SOL_SOCKET`).
175			@param optname The option name (e.g. `SO_KEEPALIVE`).
176			@param data Pointer to receive the option value.
177			@param size On entry, the size of the buffer. On exit,
178			the size of the option value.
179			@return Error code on failure, empty on success.
180			*/
181			virtual std::error_code
182			get_option(int level, int optname, void* data, std::size_t* size)
183			const noexcept = 0;
184
185			/// Return the cached local endpoint.
186			virtual corosio::local_endpoint local_endpoint() const noexcept = 0;
187
188			/// Return the cached remote endpoint.
189			virtual corosio::local_endpoint remote_endpoint() const noexcept = 0;
190			};
191
192			/// Represent the awaitable returned by @ref connect.
193			struct connect_awaitable
194			: detail::void_op_base<connect_awaitable>
195			{
196			local_stream_socket& s_;
197			corosio::local_endpoint endpoint_;
198
199		9x	connect_awaitable(
200			local_stream_socket& s, corosio::local_endpoint ep) noexcept
201		9x	: s_(s), endpoint_(ep) {}
202
203		9x	std::coroutine_handle<> dispatch(
204			std::coroutine_handle<> h, capy::executor_ref ex) const
205			{
206	1/1 ✓ Branch 4 → 5 taken 9 times.	9x	return s_.get().connect(h, ex, endpoint_, token_, &ec_);
207			}
208			};
209
210			/// Represent the awaitable returned by @ref wait.
211			struct wait_awaitable
212			: detail::void_op_base<wait_awaitable>
213			{
214			local_stream_socket& s_;
215			wait_type w_;
216
217		3x	wait_awaitable(local_stream_socket& s, wait_type w) noexcept
218		3x	: s_(s), w_(w) {}
219
220		3x	std::coroutine_handle<> dispatch(
221			std::coroutine_handle<> h, capy::executor_ref ex) const
222			{
223	1/1 ✓ Branch 4 → 5 taken 3 times.	3x	return s_.get().wait(h, ex, w_, token_, &ec_);
224			}
225			};
226
227			public:
228			/** Destructor.
229
230			Closes the socket if open, cancelling any pending operations.
231			*/
232			~local_stream_socket() override;
233
234			/** Construct a socket from an execution context.
235
236			@param ctx The execution context that will own this socket.
237			*/
238			explicit local_stream_socket(capy::execution_context& ctx);
239
240			/** Construct a socket from an executor.
241
242			The socket is associated with the executor's context.
243
244			@param ex The executor whose context will own the socket.
245			*/
246			template<class Ex>
247			requires(!std::same_as<std::remove_cvref_t<Ex>, local_stream_socket>) &&
248			capy::Executor<Ex>
249			explicit local_stream_socket(Ex const& ex) : local_stream_socket(ex.context())
250			{
251			}
252
253			/** Move constructor.
254
255			Transfers ownership of the socket resources.
256
257			@param other The socket to move from.
258
259			@pre No awaitables returned by @p other's methods exist.
260			@pre The execution context associated with @p other must
261			outlive this socket.
262			*/
263		5x	local_stream_socket(local_stream_socket&& other) noexcept
264		5x	: io_object(std::move(other))
265			{
266		5x	}
267
268			/** Move assignment operator.
269
270			Closes any existing socket and transfers ownership.
271
272			@param other The socket to move from.
273
274			@pre No awaitables returned by either `*this` or @p other's
275			methods exist.
276			@pre The execution context associated with @p other must
277			outlive this socket.
278
279			@return Reference to this socket.
280			*/
281		2x	local_stream_socket& operator=(local_stream_socket&& other) noexcept
282			{
283	2/2 ✓ Branch 2 → 3 taken 1 time. ✓ Branch 2 → 6 taken 1 time.	2x	if (this != &other)
284			{
285		1x	close();
286		1x	io_object::operator=(std::move(other));
287			}
288		2x	return *this;
289			}
290
291			local_stream_socket(local_stream_socket const&) = delete;
292			local_stream_socket& operator=(local_stream_socket const&) = delete;
293
294			/** Open the socket.
295
296			Creates a Unix stream socket and associates it with
297			the platform reactor.
298
299			Failures such as descriptor exhaustion are normal runtime
300			conditions and are reported through the returned error code.
301			Opening an already-open socket is a no-op that reports
302			success.
303
304			@param proto The protocol. Defaults to local_stream{}.
305
306			@return The error code, empty on success.
307			*/
308			[[nodiscard]] std::error_code open(local_stream proto = {}) noexcept;
309
310			/** Close the socket.
311
312			Releases socket resources. Any pending operations complete
313			with `errc::operation_canceled`.
314			*/
315			void close() noexcept;
316
317			/** Check if the socket is open.
318
319			@return `true` if the socket is open and ready for operations.
320			*/
321		218x	bool is_open() const noexcept
322			{
323			#if BOOST_COROSIO_HAS_IOCP && !defined(BOOST_COROSIO_MRDOCS)
324	4/4 ✓ Branch 3 → 4 taken 209 times. ✓ Branch 3 → 8 taken 9 times. ✓ Branch 6 → 7 taken 102 times. ✓ Branch 6 → 8 taken 107 times.	218x	return h_ && get().native_handle() != ~native_handle_type(0);
325			#else
326			return h_ && get().native_handle() >= 0;
327			#endif
328			}
329
330			/** Initiate an asynchronous connect operation.
331
332			If the socket is not already open, it is opened automatically.
333
334			@param ep The local endpoint (path) to connect to.
335
336			@return An awaitable that completes with io_result<>.
337
338			If the socket needs to be opened and the open fails, the
339			awaitable completes immediately with that error.
340			*/
341		9x	[[nodiscard]] auto connect(corosio::local_endpoint ep)
342			{
343		9x	connect_awaitable aw(*this, ep);
344	2/2 ✓ Branch 4 → 5 taken 6 times. ✓ Branch 4 → 6 taken 3 times.	9x	if (!is_open())
345		6x	aw.ec_ = open();
346		9x	return aw;
347			}
348
349			/** Wait for the socket to become ready in a given direction.
350
351			Suspends until the socket is ready for the requested
352			direction, or an error condition is reported. No bytes
353			are transferred.
354
355			@param w The wait direction (read, write, or error).
356
357			@return An awaitable that completes with `io_result<>`.
358
359			A closed socket completes with `errc::bad_file_descriptor`.
360
361			@par Preconditions
362			This socket must outlive the returned awaitable.
363			*/
364		3x	[[nodiscard]] auto wait(wait_type w)
365			{
366		3x	return wait_awaitable(*this, w);
367			}
368
369			/** Cancel any pending asynchronous operations.
370
371			All outstanding operations complete with `errc::operation_canceled`.
372			Check `ec == cond::canceled` for portable comparison.
373			*/
374			void cancel() noexcept;
375
376			/** Get the native socket handle.
377
378			Returns the underlying platform-specific socket descriptor.
379			On POSIX systems this is an `int` file descriptor.
380
381			@return The native socket handle, or an invalid sentinel
382			if not open.
383			*/
384			native_handle_type native_handle() const noexcept;
385
386			/** Query the number of bytes available for reading.
387
388			@return The number of bytes that can be read without blocking.
389
390			@throws std::system_error `errc::bad_file_descriptor` if the
391			socket is not open; otherwise thrown on ioctl failure.
392			*/
393			std::size_t available() const;
394
395			/** Release ownership of the native socket handle.
396
397			Deregisters the socket from the backend and cancels pending
398			operations without closing the descriptor. The caller takes
399			ownership of the returned handle.
400
401			@return The native handle.
402
403			@throws std::system_error `errc::bad_file_descriptor` if the
404			socket is not open.
405
406			@post is_open() == false
407			*/
408			native_handle_type release();
409
410			/** Disable sends or receives on the socket.
411
412			Unix stream connections are full-duplex: each direction
413			(send and receive) operates independently. This function
414			allows you to close one or both directions without
415			destroying the socket.
416
417			Failures such as a peer that already disconnected are
418			normal runtime conditions and are reported through the
419			returned error code. A closed socket reports
420			`errc::bad_file_descriptor`.
421
422			@param what Determines what operations will no longer
423			be allowed.
424
425			@return The error code, empty on success.
426			*/
427			[[nodiscard]] std::error_code shutdown(shutdown_type what) noexcept;
428
429			/** Set a socket option.
430
431			Applies a type-safe socket option to the underlying socket.
432			The option type encodes the protocol level and option name.
433
434			@param opt The option to set.
435
436			@throws std::system_error `errc::bad_file_descriptor` if the
437			socket is not open; otherwise thrown on failure.
438			*/
439			template<class Option>
440		3x	void set_option(Option const& opt)
441			{
442	2/2 ✓ Branch 3 → 4 taken 1 time. ✓ Branch 3 → 6 taken 2 times.	3x	if (!is_open())
443		1x	detail::throw_system_error(
444		2x	make_error_code(std::errc::bad_file_descriptor),
445			"local_stream_socket::set_option");
446		2x	std::error_code ec = get().set_option(
447			Option::level(), Option::name(), opt.data(), opt.size());
448	2/2 ✓ Branch 13 → 14 taken 1 time. ✓ Branch 13 → 15 taken 1 time.	2x	if (ec)
449		1x	detail::throw_system_error(ec, "local_stream_socket::set_option");
450		1x	}
451
452			/** Get a socket option.
453
454			Retrieves the current value of a type-safe socket option.
455
456			@return The current option value.
457
458			@throws std::system_error `errc::bad_file_descriptor` if the
459			socket is not open; otherwise thrown on failure.
460			*/
461			template<class Option>
462		3x	Option get_option() const
463			{
464	2/2 ✓ Branch 3 → 4 taken 1 time. ✓ Branch 3 → 6 taken 2 times.	3x	if (!is_open())
465		1x	detail::throw_system_error(
466		2x	make_error_code(std::errc::bad_file_descriptor),
467			"local_stream_socket::get_option");
468		2x	Option opt{};
469		2x	std::size_t sz = opt.size();
470			std::error_code ec =
471		2x	get().get_option(Option::level(), Option::name(), opt.data(), &sz);
472	2/2 ✓ Branch 13 → 14 taken 1 time. ✓ Branch 13 → 15 taken 1 time.	2x	if (ec)
473		1x	detail::throw_system_error(ec, "local_stream_socket::get_option");
474		1x	opt.resize(sz);
475		1x	return opt;
476			}
477
478			/** Assign an existing native socket to this object.
479
480			Adopts a Unix domain stream socket created outside the
481			library — from `socketpair()`, received over `SCM_RIGHTS`,
482			or made natively — and registers it with the backend. The
483			socket must be a stream socket in the `AF_UNIX` family.
484			Adoption never alters the descriptor's flags or options: on
485			POSIX the fd must already be non-blocking, and on Windows
486			the socket must be overlapped-capable.
487
488			If this object is already open, pending operations complete
489			with `errc::operation_canceled` and the held socket is
490			closed before the new one is adopted.
491
492			@par Exception Safety
493			Strong guarantee on validation failure: the object is
494			unchanged. If backend registration fails, the object either
495			retains its previous socket or is left closed, depending on
496			the backend. In all failure cases the caller retains
497			ownership of `fd`.
498
499			@param fd The native socket to adopt. On success the object
500			owns it and will close it.
501
502			@return The error code, empty on success. Validation and
503			registration failures are normal runtime conditions when
504			adopting foreign descriptors.
505			*/
506			[[nodiscard]] std::error_code assign(native_handle_type fd) noexcept;
507
508			/** Get the local endpoint of the socket.
509
510			Returns the local address (path) to which the socket is bound.
511			The endpoint is cached when the connection is established.
512
513			@return The local endpoint, or a default endpoint if the socket
514			is not connected.
515			*/
516			corosio::local_endpoint local_endpoint() const noexcept;
517
518			/** Get the remote endpoint of the socket.
519
520			Returns the remote address (path) to which the socket is connected.
521			The endpoint is cached when the connection is established.
522
523			@return The remote endpoint, or a default endpoint if the socket
524			is not connected.
525			*/
526			corosio::local_endpoint remote_endpoint() const noexcept;
527
528			protected:
529		17x	local_stream_socket() noexcept = default;
530
531			explicit local_stream_socket(handle h) noexcept : io_object(std::move(h)) {}
532
533			private:
534			friend class local_stream_acceptor;
535
536			[[nodiscard]] std::error_code
537			open_for_family(int family, int type, int protocol) noexcept;
538
539		238x	inline implementation& get() const noexcept
540			{
541		238x	return static_cast<implementation>(h_.get());
542			}
543			};
544
545			} // namespace boost::corosio
546
547			#endif // BOOST_COROSIO_LOCAL_STREAM_SOCKET_HPP
548