include/boost/corosio/local_stream_socket.hpp

94.9% Lines (56/59) 100.0% List of functions (22/22) 65.4% Branches (17/26)

[ ] prev/next file n p prev/next uncovered line

local_stream_socket.hpp

f(x) Functions (22)

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