Job queue

CMakeLists.txt

@@ -17,7 +17,7 @@ if(CCACHE_PROGRAM)
 endif()
 
 project(libquic
-    VERSION 1.7.1
+    VERSION 1.8.0
     DESCRIPTION "Modular QUIC library for stream and connection management"
     LANGUAGES ${LANGS})
 

include/oxen/quic/btstream.hpp

@@ -247,6 +247,7 @@ namespace oxen::quic
         friend struct sent_request;
         friend class Network;
         friend class Loop;
+        friend class JobQueue;
 
       protected:
         template <typename... Opt>
@@ -281,7 +282,7 @@ namespace oxen::quic
             auto req = std::make_shared<sent_request>(*this, encode_command(ep, rid, body), rid, std::forward<Opt>(opts)...);
 
             if (req->cb)
-                loop.call([this, r = std::move(req)]() mutable {
+                endpoint.job_queue.call([this, r = std::move(req)]() mutable {
                     if (auto* req = add_sent_request(std::move(r)))
                         send(std::move(req->data));
                 });

include/oxen/quic/connection.hpp

@@ -126,7 +126,7 @@ namespace oxen::quic
             // has a forward declaration; the user of this method needs to have the full definition
             // available to call this.
             return std::static_pointer_cast<StreamT>(queue_incoming_stream_impl([&](Connection& c, EndpointDeferred& e) {
-                return e.loop.template make_shared<StreamT>(c, e, std::forward<Args>(args)...);
+                return e.job_queue.template make_shared<StreamT>(c, e, std::forward<Args>(args)...);
             }));
         }
 
@@ -148,7 +148,7 @@ namespace oxen::quic
         std::shared_ptr<StreamT> open_stream(Args&&... args)
         {
             return std::static_pointer_cast<StreamT>(open_stream_impl([&](Connection& c, EndpointDeferred& e) {
-                return e.loop.template make_shared<StreamT>(c, e, std::forward<Args>(args)...);
+                return e.job_queue.template make_shared<StreamT>(c, e, std::forward<Args>(args)...);
             }));
         }
 

include/oxen/quic/datagram.hpp

@@ -329,6 +329,7 @@ namespace oxen::quic
       protected:
         friend class Connection;
         friend class Loop;
+        friend class JobQueue;
         friend struct dgram::rotating_buffer;
         friend class TestHelper;
 

include/oxen/quic/endpoint.hpp

@@ -50,6 +50,7 @@ namespace oxen::quic
         connection_closed_callback connection_close_cb;
 
         Loop& loop;
+        JobQueue job_queue{loop};
 
         template <typename... Opt>
         void listen(Opt&&... opts)
@@ -58,7 +59,7 @@ namespace oxen::quic
                     (0 + ... + std::is_convertible_v<std::remove_cvref_t<Opt>, std::shared_ptr<TLSCreds>>) == 1,
                     "listen() requires exactly one std::shared_ptr<TLSCreds> argument");
 
-            loop.call_get([&opts..., this]() {
+            job_queue.call_get([&opts..., this]() {
                 if (inbound_ctx)
                     throw std::logic_error{"Cannot call listen() more than once"};
 
@@ -82,7 +83,7 @@ namespace oxen::quic
             if (_local.is_ipv6() && !remote.is_ipv6())
                 remote.map_ipv4_as_ipv6();
 
-            return loop.call_get([this, &opts..., remote = std::move(remote)]() mutable {
+            return job_queue.call_get([this, &opts..., remote = std::move(remote)]() mutable {
                 // initialize client context and client tls context simultaneously
                 auto outbound_ctx = std::make_shared<IOContext>(Direction::OUTBOUND, std::forward<Opt>(opts)...);
                 _assign_context_globals(*outbound_ctx);
@@ -159,6 +160,7 @@ namespace oxen::quic
       private:
         friend class Network;
         friend class Loop;
+        friend class JobQueue;
         friend class Connection;
         friend struct connection_callbacks;
         friend class TestHelper;

include/oxen/quic/iochannel.hpp

@@ -121,7 +121,7 @@ namespace oxen::quic
                 typename EP = Endpoint>
         Ret call_get_accessor(T (Class::*getter)() const) const
         {
-            return static_cast<EP&>(endpoint).loop.call_get(
+            return static_cast<EP&>(endpoint).job_queue.call_get(
                     [this, &getter]() -> Ret { return (static_cast<const Class*>(this)->*getter)(); });
         }
     };