diff --git a/capnp/async.pyx b/capnp/async.pyx index afe4605..89eb15d 100644 --- a/capnp/async.pyx +++ b/capnp/async.pyx @@ -25,30 +25,81 @@ cdef class EventLoop: Py_INCREF(func) return Promise()._init(async.evalLater(self.thisptr, func)) - cdef wait(self, async.PyPromise * promise): - return self.thisptr.wait(movePromise(deref(promise))) + cpdef wait(self, Promise promise) except+: + if promise.is_consumed: + raise RuntimeError('Promise was already used in a consuming operation. You can no longer use this Promise object') + + ret = self.thisptr.wait(movePromise(deref(promise.thisptr))) + promise.is_consumed = True + + return ret + + cpdef there(self, Promise promise, object func, object error_func=None): + if promise.is_consumed: + raise RuntimeError('Promise was already used in a consuming operation. You can no longer use this Promise object') - cdef there(self, async.PyPromise * promise, object func, object error_func): Py_INCREF(func) Py_INCREF(error_func) - return Promise()._init(async.there(self.thisptr, deref(promise), func, error_func)) + return Promise()._init(async.there(self.thisptr, deref(promise.thisptr), func, error_func)) - -cdef EventLoop c_event_loop = EventLoop() -event_loop = c_event_loop + cpdef yield_end(self): + return Promise()._init(async.yield_end(self.thisptr)) cdef class Promise: cdef async.PyPromise * thisptr + cdef public bint is_consumed + + def __init__(self): + self.is_consumed = True + cdef _init(self, async.PyPromise other): + self.is_consumed = False self.thisptr = new async.PyPromise(movePromise(other)) return self def __dealloc__(self): del self.thisptr - def wait(self): - return c_event_loop.wait(self.thisptr) + cpdef wait(self) except+: + if self.is_consumed: + raise RuntimeError('Promise was already used in a consuming operation. You can no longer use this Promise object') - def then(self, func, error_func=None): - return c_event_loop.there(self.thisptr, func, error_func) + ret = self.thisptr.wait() + self.is_consumed = True + return ret + + cpdef then(self, func, error_func=None) except+: + if self.is_consumed: + raise RuntimeError('Promise was already used in a consuming operation. You can no longer use this Promise object') + + Py_INCREF(func) + Py_INCREF(error_func) + + return Promise()._init(async.then(deref(self.thisptr), func, error_func)) + +# from gevent.event import Event, AsyncResult +# import gevent + +# cdef object _event = Event() +# cdef object _start_loop = AsyncResult() + +# def _start_event_loop(): +# loop = EventLoop() +# _start_loop.set(loop) +# _event.wait() +# _event.clear() + +# _event_loop_greenlet = gevent.spawn(_start_event_loop) + +# event_loop = _start_loop.get() +# _event.set() + +# cdef public void _gevent_eventloop_prepare_to_sleep(): +# _event.clear() + +# cdef public void _gevent_eventloop_sleep(): +# _event.wait() + +# cdef public void _gevent_eventloop_wake(): +# _event.set() diff --git a/capnp/asyncHelper.h b/capnp/asyncHelper.h index f213375..ae115cf 100644 --- a/capnp/asyncHelper.h +++ b/capnp/asyncHelper.h @@ -3,6 +3,9 @@ extern "C" { PyObject * wrap_kj_exception(kj::Exception &); + // void _gevent_eventloop_prepare_to_sleep(); + // void _gevent_eventloop_sleep(); + // void _gevent_eventloop_wake(); } PyObject * wrapPyFunc(PyObject * func, PyObject * arg) { @@ -21,4 +24,32 @@ PyObject * wrapPyFunc(PyObject * func, PyObject * arg) { else return loop.there(kj::mv(promise), [func](PyObject * arg) { return wrapPyFunc(func, arg); } , [error_func](kj::Exception arg) { return wrapPyFunc(error_func, wrap_kj_exception(arg)); } ); -} \ No newline at end of file +} +::kj::Promise then(kj::Promise & promise, PyObject * func, PyObject * error_func) { + if(error_func == Py_None) + return promise.then([func](PyObject * arg) { return wrapPyFunc(func, arg); } ); + else + return promise.then([func](PyObject * arg) { return wrapPyFunc(func, arg); } + , [error_func](kj::Exception arg) { return wrapPyFunc(error_func, wrap_kj_exception(arg)); } ); +} + +::kj::Promise yield_end(kj::EventLoop & loop) { + return loop.there(loop.yield(), []() { Py_RETURN_NONE; } ); +} + +// class PyEventLoop final: public ::kj::EventLoop { +// public: +// PyEventLoop() {} +// ~PyEventLoop() noexcept(false) {} + +// protected: +// void prepareToSleep() noexcept override { +// _gevent_eventloop_prepare_to_sleep(); +// } +// void sleep() override { +// _gevent_eventloop_sleep(); +// } +// void wake() const override { +// _gevent_eventloop_wake(); +// } +// }; \ No newline at end of file diff --git a/capnp/async_cpp.pxd b/capnp/async_cpp.pxd index 4e87530..d26f489 100644 --- a/capnp/async_cpp.pxd +++ b/capnp/async_cpp.pxd @@ -18,9 +18,9 @@ ctypedef Promise[PyObject *] PyPromise cdef extern from "kj/async.h" namespace " ::kj": cdef cppclass EventLoop: EventLoop() - # Promise[void] yieldFrom'yield'() - object wait(PyPromise) - object there(PyPromise) + # Promise[void] yield_end'yield'() + object wait(PyPromise) except+ + object there(PyPromise) except+ PyPromise evalLater(PyObject * func) PyPromise there(PyPromise, PyObject * func) cdef cppclass SimpleEventLoop(EventLoop): @@ -28,4 +28,9 @@ cdef extern from "kj/async.h" namespace " ::kj": cdef extern from "asyncHelper.h": PyPromise evalLater(EventLoop &, PyObject * func) - PyPromise there(EventLoop & loop, PyPromise & promise, PyObject * func, PyObject * error_func) \ No newline at end of file + PyPromise there(EventLoop & loop, PyPromise & promise, PyObject * func, PyObject * error_func) + PyPromise then(PyPromise & promise, PyObject * func, PyObject * error_func) + PyPromise yield_end(EventLoop & loop) + + # cdef cppclass PyEventLoop(EventLoop): + # pass \ No newline at end of file