py::object await()

in src/moolib.cc [301:374]

• 72 lines of code
• 15 McCabe index (conditional complexity)

  py::object await() {
    if (!moduleState) {
      fatal("moolib module is not loaded");
    }
    if (!moduleState->get_running_loop) {
      moduleState->get_running_loop = py::module::import("asyncio").attr("get_running_loop");
    }
    py::object loop = moduleState->get_running_loop();
    py::object future = loop.attr("create_future")();

    callback = rpc::Function<void()>([me = shared(), loop = GilWrapper<py::object>(std::move(loop)),
                                      future = GilWrapper<py::object>(future)]() mutable noexcept {
                 int f = me->flags.load(std::memory_order_relaxed);
                 py::gil_scoped_acquire gil;
                 if (_Py_IsFinalizing()) {
                   return;
                 }
                 try {
                   keepThreadAlive();
                   if (f & 1) {
                     loop->attr("call_soon_threadsafe")(
                         py::cpp_function([me = std::move(me), future = std::move(future)]() mutable {
                           try {
                             if (!me->value) {
                               future->attr("set_exception")(py::reinterpret_borrow<py::object>(PyExc_RuntimeError)(
                                   "Future value not available. Maybe it has already been retrieved"));
                             } else {
                               future->attr("set_result")(me->value.release());
                             }
                           } catch (const py::error_already_set& e) {
                             // InvalidStateError probably means that the future was cancelled
                             if (!e.matches(py::module::import("asyncio").attr("InvalidStateError"))) {
                               throw;
                             }
                           }
                         }));
                   } else if (f & 2) {
                     try {
                       loop->attr("call_soon_threadsafe")(py::cpp_function([me, future = std::move(future)]() mutable {
                         try {
                           std::lock_guard l(me->errorMutex);
                           future->attr("set_exception")(
                               py::reinterpret_borrow<py::object>(PyExc_RuntimeError)(me->error->what()));
                         } catch (const py::error_already_set& e) {
                           if (!e.matches(py::module::import("asyncio").attr("InvalidStateError"))) {
                             throw;
                           }
                         }
                       }));
                     } catch (const py::error_already_set& e) {
                       std::string s;
                       {
                         std::lock_guard l(me->errorMutex);
                         s = me->error->what();
                       }
                       fatal("Python exception during callback: %s\nOriginal exception: %s", e.what(), s);
                     }
                   } else if (f & 4) {
                     loop->attr("call_soon_threadsafe")(
                         py::cpp_function([future = std::move(future)]() mutable { future->attr("cancel")(); }));
                   } else {
                     fatal("Future callback called in invalid state");
                   }
                 } catch (const py::error_already_set& e) {
                   fatal("Python exception during callback: %s", e.what());
                 }
                 future.reset();
                 loop.reset();
               }).release();
    if (flags.load() != 0) {
      doCallback();
    }
    return future.attr("__await__")();
  }