Fix up async wrappers. Comment out gevent stuff for now

capnp/async.pyx

@@ -25,30 +25,81 @@ cdef class EventLoop:
         Py_INCREF(func)
         return Promise()._init(async.evalLater(self.thisptr, <PyObject *>func))
 
-    cdef wait(self, async.PyPromise * promise):
+    cpdef wait(self, Promise promise) except+:
-        return self.thisptr.wait(movePromise(deref(promise)))
+        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), <PyObject *>func, <PyObject *>error_func))
+        return Promise()._init(async.there(self.thisptr, deref(promise.thisptr), <PyObject *>func, <PyObject *>error_func))
 
-
+    cpdef yield_end(self):
-cdef EventLoop c_event_loop = EventLoop()
+        return Promise()._init(async.yield_end(self.thisptr))
-event_loop = c_event_loop
 
 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):
+    cpdef wait(self) except+:
-        return c_event_loop.wait(self.thisptr)
+        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):
+        ret = <object>self.thisptr.wait()
-        return c_event_loop.there(self.thisptr, func, error_func)
+        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), <PyObject *>func, <PyObject *>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()

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)); } );
-}
+}
+::kj::Promise<PyObject *> then(kj::Promise<PyObject *> & 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<PyObject *> 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();
+//   }
+// };

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'()
+        # Promise[void] yield_end'yield'()
-        object wait(PyPromise)
+        object wait(PyPromise) except+
-        object there(PyPromise)
+        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) 
+    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