include/boost/corosio/local_stream_acceptor.hpp

92.5% Lines (74/0/80) 100.0% List of functions (19/0/19)

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

local_stream_acceptor.hpp

f(x) Functions (19)

Function Calls Lines Blocks

Line	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_ACCEPTOR_HPP
11		#define BOOST_COROSIO_LOCAL_STREAM_ACCEPTOR_HPP
12
13		#include <boost/corosio/detail/config.hpp>
14		#include <boost/corosio/detail/except.hpp>
15		#include <boost/corosio/detail/op_base.hpp>
16		#include <boost/corosio/wait_type.hpp>
17		#include <boost/corosio/io/io_object.hpp>
18		#include <boost/capy/io_result.hpp>
19		#include <boost/corosio/local_endpoint.hpp>
20		#include <boost/corosio/local_stream.hpp>
21		#include <boost/corosio/local_stream_socket.hpp>
22		#include <boost/capy/ex/executor_ref.hpp>
23		#include <boost/capy/ex/execution_context.hpp>
24		#include <boost/capy/ex/io_env.hpp>
25		#include <boost/capy/concept/executor.hpp>
26
27		#include <system_error>
28
29		#include <cassert>
30		#include <concepts>
31		#include <coroutine>
32		#include <cstddef>
33		#include <stop_token>
34		#include <type_traits>
35
36		namespace boost::corosio {
37
38		/** Options for @ref local_stream_acceptor::bind().
39
40		Controls filesystem cleanup behavior before binding
41		to a Unix domain socket path.
42		*/
43		enum class bind_option
44		{
45		none,
46		/// Unlink the socket path before binding (ignored for abstract paths).
47		unlink_existing
48		};
49
50		/** An asynchronous Unix domain stream acceptor for coroutine I/O.
51
52		This class provides asynchronous Unix domain stream accept
53		operations that return awaitable types. The acceptor binds
54		to a local endpoint (filesystem path or abstract name) and
55		listens for incoming connections.
56
57		The library does NOT automatically unlink the socket path
58		on close. Callers are responsible for removing the socket
59		file before bind (via @ref bind_option::unlink_existing) or
60		after close.
61
62		@par Thread Safety
63		Distinct objects: Safe.@n
64		Shared objects: Unsafe. An acceptor must not have concurrent
65		accept operations.
66
67		@par Example
68		@code
69		io_context ioc;
70		local_stream_acceptor acc(ioc);
71		acc.open();
72		acc.bind(local_endpoint("/tmp/my.sock"),
73		bind_option::unlink_existing);
74		acc.listen();
75		auto [ec, peer] = co_await acc.accept();
76		@endcode
77		*/
78		class BOOST_COROSIO_DECL local_stream_acceptor : public io_object
79		{
80		struct wait_awaitable
81		: detail::void_op_base<wait_awaitable>
82		{
83		local_stream_acceptor& acc_;
84		wait_type w_;
85
86	4x	wait_awaitable(local_stream_acceptor& acc, wait_type w) noexcept
87	4x	: acc_(acc), w_(w) {}
88
89	4x	std::coroutine_handle<> dispatch(
90		std::coroutine_handle<> h, capy::executor_ref ex) const
91		{
92	4x	return acc_.get().wait(h, ex, w_, token_, &ec_);
93		}
94		};
95
96		struct move_accept_awaitable
97		{
98		local_stream_acceptor& acc_;
99		std::stop_token token_;
100		mutable std::error_code ec_;
101		mutable io_object::implementation* peer_impl_ = nullptr;
102
103	2x	explicit move_accept_awaitable(
104		local_stream_acceptor& acc) noexcept
105	2x	: acc_(acc)
106		{
107	2x	}
108
109	2x	bool await_ready() const noexcept
110		{
111	2x	return token_.stop_requested();
112		}
113
114	2x	[[nodiscard]] capy::io_result<local_stream_socket> await_resume() const noexcept
115		{
116	2x	if (token_.stop_requested())
117	✗	return {make_error_code(std::errc::operation_canceled),
118	✗	local_stream_socket()};
119
120	2x	if (ec_ \|\| !peer_impl_)
121	✗	return {ec_, local_stream_socket()};
122
123	2x	local_stream_socket peer(acc_.ctx_);
124	2x	reset_peer_impl(peer, peer_impl_);
125	2x	return {ec_, std::move(peer)};
126	2x	}
127
128	2x	auto await_suspend(std::coroutine_handle<> h, capy::io_env const* env)
129		-> std::coroutine_handle<>
130		{
131	2x	token_ = env->stop_token;
132	6x	return acc_.get().accept(
133	6x	h, env->executor, token_, &ec_, &peer_impl_);
134		}
135		};
136
137		struct accept_awaitable
138		{
139		local_stream_acceptor& acc_;
140		local_stream_socket& peer_;
141		std::stop_token token_;
142		mutable std::error_code ec_;
143		mutable io_object::implementation* peer_impl_ = nullptr;
144
145	27x	accept_awaitable(
146		local_stream_acceptor& acc, local_stream_socket& peer) noexcept
147	27x	: acc_(acc)
148	27x	, peer_(peer)
149		{
150	27x	}
151
152	27x	bool await_ready() const noexcept
153		{
154	27x	return token_.stop_requested();
155		}
156
157	25x	[[nodiscard]] capy::io_result<> await_resume() const noexcept
158		{
159	25x	if (token_.stop_requested())
160	4x	return {make_error_code(std::errc::operation_canceled)};
161
162	21x	if (!ec_ && peer_impl_)
163	17x	peer_.h_.reset(peer_impl_);
164	21x	return {ec_};
165		}
166
167	27x	auto await_suspend(std::coroutine_handle<> h, capy::io_env const* env)
168		-> std::coroutine_handle<>
169		{
170	27x	token_ = env->stop_token;
171	81x	return acc_.get().accept(
172	81x	h, env->executor, token_, &ec_, &peer_impl_);
173		}
174		};
175
176		public:
177		/** Destructor.
178
179		Closes the acceptor if open, cancelling any pending operations.
180		*/
181		~local_stream_acceptor() override;
182
183		/** Construct an acceptor from an execution context.
184
185		@param ctx The execution context that will own this acceptor.
186		*/
187		explicit local_stream_acceptor(capy::execution_context& ctx);
188
189		/** Construct an acceptor from an executor.
190
191		The acceptor is associated with the executor's context.
192
193		@param ex The executor whose context will own the acceptor.
194
195		@tparam Ex A type satisfying @ref capy::Executor. Must not
196		be `local_stream_acceptor` itself (disables implicit
197		conversion from move).
198		*/
199		template<class Ex>
200		requires(!std::same_as<std::remove_cvref_t<Ex>, local_stream_acceptor>) &&
201		capy::Executor<Ex>
202		explicit local_stream_acceptor(Ex const& ex) : local_stream_acceptor(ex.context())
203		{
204		}
205
206		/** Move constructor.
207
208		Transfers ownership of the acceptor resources.
209
210		@param other The acceptor to move from.
211
212		@pre No awaitables returned by @p other's methods exist.
213		@pre The execution context associated with @p other must
214		outlive this acceptor.
215		*/
216		local_stream_acceptor(local_stream_acceptor&& other) noexcept
217		: local_stream_acceptor(other.ctx_, std::move(other))
218		{
219		}
220
221		/** Move assignment operator.
222
223		Closes any existing acceptor and transfers ownership.
224		Both acceptors must share the same execution context.
225
226		@param other The acceptor to move from.
227
228		@return Reference to this acceptor.
229
230		@pre `&ctx_ == &other.ctx_` (same execution context).
231		@pre No awaitables returned by either `*this` or @p other's
232		methods exist.
233		*/
234		local_stream_acceptor& operator=(local_stream_acceptor&& other) noexcept
235		{
236		assert(&ctx_ == &other.ctx_ &&
237		"move-assign requires the same execution_context");
238		if (this != &other)
239		{
240		close();
241		io_object::operator=(std::move(other));
242		}
243		return *this;
244		}
245
246		local_stream_acceptor(local_stream_acceptor const&) = delete;
247		local_stream_acceptor& operator=(local_stream_acceptor const&) = delete;
248
249		/** Create the acceptor socket.
250
251		@param proto The protocol. Defaults to local_stream{}.
252
253		@throws std::system_error on failure.
254		*/
255		void open(local_stream proto = {});
256
257		/** Bind to a local endpoint.
258
259		@param ep The local endpoint (path) to bind to.
260		@param opt Bind options. Pass bind_option::unlink_existing
261		to unlink the socket path before binding (ignored for
262		abstract sockets and empty endpoints).
263
264		@return An error code on failure, empty on success.
265
266		@throws std::logic_error if the acceptor is not open.
267		*/
268		[[nodiscard]] std::error_code
269		bind(corosio::local_endpoint ep,
270		bind_option opt = bind_option::none);
271
272		/** Start listening for incoming connections.
273
274		@param backlog The maximum pending connection queue length.
275
276		@return An error code on failure, empty on success.
277
278		@throws std::logic_error if the acceptor is not open.
279		*/
280		[[nodiscard]] std::error_code listen(int backlog = 128);
281
282		/** Close the acceptor.
283
284		Cancels any pending accept operations and releases the
285		underlying socket. Has no effect if the acceptor is not
286		open.
287
288		@post is_open() == false
289		*/
290		void close();
291
292		/// Check if the acceptor has an open socket handle.
293	365x	bool is_open() const noexcept
294		{
295	365x	return h_ && get().is_open();
296		}
297
298		/** Initiate an asynchronous accept into an existing socket.
299
300		Completes when a new connection is available. On success
301		@p peer is reset to the accepted connection. Only one
302		accept may be in flight at a time.
303
304		@param peer The socket to receive the accepted connection.
305
306		@par Cancellation
307		Supports cancellation via stop_token or cancel().
308		On cancellation, yields `capy::cond::canceled` and
309		@p peer is not modified.
310
311		@return An awaitable that completes with io_result<>.
312
313		@throws std::logic_error if the acceptor is not open.
314		*/
315	29x	auto accept(local_stream_socket& peer)
316		{
317	29x	if (!is_open())
318	2x	detail::throw_logic_error("accept: acceptor not listening");
319	27x	return accept_awaitable(*this, peer);
320		}
321
322		/** Wait for an incoming connection or readiness condition.
323
324		Suspends until the listen socket is ready in the
325		requested direction. For `wait_type::read`, completion
326		signals that a subsequent @ref accept will succeed
327		without blocking; a connection already queued when the
328		wait begins completes it immediately. No connection is
329		consumed.
330
331		@note `wait_type::write` is not usable on an acceptor:
332		writability carries no meaning for a listening socket, so
333		the wait fails with `errc::operation_not_supported` on
334		every backend.
335
336		@param w The wait direction.
337
338		@return An awaitable that completes with `io_result<>`.
339
340		@par Preconditions
341		The acceptor must be listening.
342		*/
343	4x	[[nodiscard]] auto wait(wait_type w)
344		{
345	4x	if (!is_open())
346	✗	detail::throw_logic_error("wait: acceptor not listening");
347	4x	return wait_awaitable(*this, w);
348		}
349
350		/** Initiate an asynchronous accept, returning the socket.
351
352		Completes when a new connection is available. Only one
353		accept may be in flight at a time.
354
355		@par Cancellation
356		Supports cancellation via stop_token or cancel().
357		On cancellation, yields `capy::cond::canceled` with
358		a default-constructed socket.
359
360		@return An awaitable that completes with
361		io_result<local_stream_socket>.
362
363		@throws std::logic_error if the acceptor is not open.
364		*/
365	4x	auto accept()
366		{
367	4x	if (!is_open())
368	2x	detail::throw_logic_error("accept: acceptor not listening");
369	2x	return move_accept_awaitable(*this);
370		}
371
372		/** Cancel pending asynchronous accept operations.
373
374		Outstanding accept operations complete with
375		@c capy::cond::canceled. Safe to call when no
376		operations are pending (no-op).
377		*/
378		void cancel();
379
380		/** Release ownership of the native socket handle.
381
382		Deregisters the acceptor from the reactor and cancels
383		pending operations without closing the descriptor. The
384		caller takes ownership of the returned handle.
385
386		@return The native handle.
387
388		@throws std::logic_error if the acceptor is not open.
389
390		@post is_open() == false
391		*/
392		native_handle_type release();
393
394		/** Get the native socket handle.
395
396		@return The native socket handle, or -1/INVALID_SOCKET if not
397		open.
398
399		@par Preconditions
400		None. May be called on closed acceptors.
401		*/
402		native_handle_type native_handle() const noexcept;
403
404		/** Assign an existing native socket to this acceptor.
405
406		Adopts a listening socket created outside the library —
407		received from a service manager, inherited, or made natively —
408		and registers it with the backend. The socket must be a
409		listening stream socket in the local IPC family. Adoption
410		never alters the descriptor's flags or options: on POSIX the
411		fd must already be non-blocking, and on Windows the socket
412		must be overlapped-capable.
413
414		Adoption does not verify listen state; @ref accept reports the
415		error if the socket is not listening.
416
417		If this object is already open, pending operations complete
418		with `errc::operation_canceled` and the held socket is closed
419		before the new one is adopted.
420
421		@par Exception Safety
422		Strong guarantee on validation failure: the object is
423		unchanged. If backend registration fails, the object either
424		retains its previous socket or is left closed, depending on
425		the backend. In all failure cases the caller retains
426		ownership of `fd`.
427
428		@param fd The native socket to adopt. On success the object
429		owns it and will close it.
430
431		@throws std::system_error On validation or registration
432		failure.
433		*/
434		void assign(native_handle_type fd);
435
436		/** Return the local endpoint the acceptor is bound to.
437
438		Returns a default-constructed (empty) endpoint if the
439		acceptor is not open or not yet bound. Safe to call in
440		any state.
441		*/
442		corosio::local_endpoint local_endpoint() const noexcept;
443
444		/** Set a socket option on the acceptor.
445
446		Applies a type-safe socket option to the underlying socket.
447		The option type encodes the protocol level and option name.
448
449		@param opt The option to set.
450
451		@tparam Option A socket option type providing static
452		`level()` and `name()` members, and `data()` / `size()`
453		accessors.
454
455		@throws std::logic_error if the acceptor is not open.
456		@throws std::system_error on failure.
457		*/
458		template<class Option>
459	4x	void set_option(Option const& opt)
460		{
461	4x	if (!is_open())
462	2x	detail::throw_logic_error("set_option: acceptor not open");
463	2x	std::error_code ec = get().set_option(
464		Option::level(), Option::name(), opt.data(), opt.size());
465	2x	if (ec)
466	✗	detail::throw_system_error(ec, "local_stream_acceptor::set_option");
467	2x	}
468
469		/** Get a socket option from the acceptor.
470
471		Retrieves the current value of a type-safe socket option.
472
473		@return The current option value.
474
475		@tparam Option A socket option type providing static
476		`level()` and `name()` members, and `data()` / `size()`
477		/ `resize()` members.
478
479		@throws std::logic_error if the acceptor is not open.
480		@throws std::system_error on failure.
481		*/
482		template<class Option>
483	4x	Option get_option() const
484		{
485	4x	if (!is_open())
486	2x	detail::throw_logic_error("get_option: acceptor not open");
487	2x	Option opt{};
488	2x	std::size_t sz = opt.size();
489		std::error_code ec =
490	2x	get().get_option(Option::level(), Option::name(), opt.data(), &sz);
491	2x	if (ec)
492	✗	detail::throw_system_error(ec, "local_stream_acceptor::get_option");
493	2x	opt.resize(sz);
494	2x	return opt;
495		}
496
497		/** Backend hooks for local stream acceptor operations.
498
499		Platform backends derive from this to implement
500		accept, option, and lifecycle management.
501		*/
502		struct implementation : io_object::implementation
503		{
504		/** Initiate an asynchronous accept.
505
506		On completion the backend sets @p *ec and, on
507		success, stores a pointer to the new socket
508		implementation in @p *impl_out.
509
510		@param h Coroutine handle to resume.
511		@param ex Executor for dispatching the completion.
512		@param token Stop token for cancellation.
513		@param ec Output error code.
514		@param impl_out Output pointer for the accepted socket.
515		@return Coroutine handle to resume immediately.
516		*/
517		virtual std::coroutine_handle<> accept(
518		std::coroutine_handle<>,
519		capy::executor_ref,
520		std::stop_token,
521		std::error_code*,
522		io_object::implementation**) = 0;
523
524		/** Initiate an asynchronous wait for acceptor readiness.
525
526		Completes when the listen socket becomes ready for
527		the specified direction. No connection is consumed.
528		*/
529		virtual std::coroutine_handle<> wait(
530		std::coroutine_handle<> h,
531		capy::executor_ref ex,
532		wait_type w,
533		std::stop_token token,
534		std::error_code* ec) = 0;
535
536		/// Return the cached local endpoint.
537		virtual corosio::local_endpoint local_endpoint() const noexcept = 0;
538
539		/// Return whether the underlying socket is open.
540		virtual bool is_open() const noexcept = 0;
541
542		/// Return the native handle, or the platform sentinel if closed.
543		virtual native_handle_type native_handle() const noexcept = 0;
544
545		/// Release and return the native handle without closing.
546		virtual native_handle_type release_socket() noexcept = 0;
547
548		/// Cancel pending accept operations.
549		virtual void cancel() noexcept = 0;
550
551		/// Set a raw socket option.
552		virtual std::error_code set_option(
553		int level,
554		int optname,
555		void const* data,
556		std::size_t size) noexcept = 0;
557
558		/// Get a raw socket option.
559		virtual std::error_code
560		get_option(int level, int optname, void* data, std::size_t* size)
561		const noexcept = 0;
562		};
563
564		protected:
565	12x	local_stream_acceptor(handle h, capy::execution_context& ctx) noexcept
566	12x	: io_object(std::move(h))
567	12x	, ctx_(ctx)
568		{
569	12x	}
570
571		local_stream_acceptor(
572		capy::execution_context& ctx, local_stream_acceptor&& other) noexcept
573		: io_object(std::move(other))
574		, ctx_(ctx)
575		{
576		}
577
578	8x	static void reset_peer_impl(
579		local_stream_socket& peer, io_object::implementation* impl) noexcept
580		{
581	8x	if (impl)
582	8x	peer.h_.reset(impl);
583	8x	}
584
585		private:
586		capy::execution_context& ctx_;
587
588	430x	inline implementation& get() const noexcept
589		{
590	430x	return static_cast<implementation>(h_.get());
591		}
592		};
593
594		} // namespace boost::corosio
595
596		#endif // BOOST_COROSIO_LOCAL_STREAM_ACCEPTOR_HPP
597