if (!fn) {

return {};

}

// Own the py::function via a shared_ptr with a GIL-acquiring deleter: the last

// reference may be released on a non-Python thread (a reactor pool worker

// erasing an entry after remove()), and ~py::function must run under the GIL.

auto cb = std::shared_ptr<py::function>(new py::function(fn), [](py::function *f) {

py::gil_scoped_acquire gil;

delete f;

});

return [cb](std::vector<uint8_t> &data,

const espp::Socket::Info &sender) -> std::optional<std::vector<uint8_t>> {

py::gil_scoped_acquire gil;

// The handler runs on a reactor pool worker; a Python exception (or a bad

// return type) must not propagate out - it would crash/stall the worker.

// Report it and degrade to "no response".

try {

py::object result =

(*cb)(py::bytes(reinterpret_cast<const char *>(data.data()), data.size()), sender);

if (result.is_none()) {

return std::nullopt;

}

// Accept either bytes or str as the response payload.

std::string s = py::cast<std::string>(result);

return std::vector<uint8_t>(s.begin(), s.end());

} catch (py::error_already_set &e) {

// Reports the traceback via sys.unraisablehook and clears the error.

e.discard_as_unraisable("espp.SocketReactor receive callback");

return std::nullopt;

} catch (const std::exception &e) {

py::print("espp.SocketReactor receive callback error:", e.what());

return std::nullopt;

}

};

py::class_<SocketReactor>(

m, "SocketReactor", py::dynamic_attr(),

"A select()-based event loop that multiplexes many receiver sockets onto a thread pool, "

"instead of one thread per socket.")

// Construct with an owned thread pool of `worker_count` workers. (Sharing an external

// pool is available in C++ but not exposed here.)

.def(py::init(

[](std::size_t worker_count, bool auto_start, espp::Logger::Verbosity log_level) {

SocketReactor::Config config;

config.pool_config.worker_count = worker_count;

config.auto_start = auto_start;

config.log_level = log_level;

return std::make_unique<SocketReactor>(config);

}),

py::arg("worker_count") = 2, py::arg("auto_start") = true,

py::arg("log_level") = espp::Logger::Verbosity::WARN)

// Release the GIL around start()/stop(): stop() blocks waiting for

// in-flight handlers, and a pool worker running a Python receive callback

// needs the GIL - holding it here would deadlock.

.def("start", &SocketReactor::start, py::call_guard<py::gil_scoped_release>(),

"Start the select() loop (and owned pool).")

.def("stop", &SocketReactor::stop, py::call_guard<py::gil_scoped_release>(),

"Stop the loop and wait for in-flight handlers to finish.")

.def("is_running", &SocketReactor::is_running)

.def("num_registered", &SocketReactor::num_registered)

.def("remove",

static_cast<bool (SocketReactor::*)(SocketReactor::Id)>(&SocketReactor::remove),

py::arg("id"), "Unregister a socket by the id returned from add_udp_receiver().")

.def(

"add_udp_receiver",

[](SocketReactor &self, espp::UdpSocket &socket, std::size_t port,

std::size_t buffer_size, const py::function &callback, espp::QosBand band,

std::optional<espp::Dscp> dscp) -> SocketReactor::Id {

espp::UdpSocket::ReceiveConfig rc;

rc.port = port;

rc.buffer_size = buffer_size;

rc.on_receive_callback = wrap_receive_callback(callback);

rc.band = band;

rc.dscp = dscp;

return self.add_udp_receiver(socket, rc);

},

py::arg("socket"), py::arg("port"), py::arg("buffer_size"), py::arg("callback"),

py::arg("band") = espp::QosBand::Normal, py::arg("dscp") = std::optional<espp::Dscp>{},

"Bind `socket` to `port` and receive on it via the reactor. `callback(data: bytes, "

"sender) -> Optional[bytes]`; a returned bytes is sent back to the sender. `band` "

"selects the espp.QosBand this socket's handlers are dispatched at; `dscp` (an "

"espp.Dscp, e.g. Dscp.Ef) optionally marks transmitted replies (IP_TOS, best-effort). "

"Returns a registration id (0 == INVALID_ID on failure).")

.def_property_readonly_static(

"INVALID_ID", [](py::object) { return SocketReactor::INVALID_ID; },

"The id value returned by add_* on failure.");