Clojure -> Python Async

example.clj

(defn sleep [s]
    (-> (asyncio/sleep s)
        (pp/then (fn [_] s))))

  (defn fun []
    (pp/let [x (sleep 1)
             z (pp/resolved 5)
             y (sleep 2)]
      (+ x y z)))

  @(fun)

impl.clj

(ns ch.python.impl
  (:require [libpython-clj2.python.bridge-as-jvm :as py-bridge-jvm]
            [libpython-clj2.python.protocols :as pyprot]
            [libpython-clj2.require :refer [require-python]]
            [libpython-clj2.python :refer [py.] :as py])
  (:import
   java.util.concurrent.CompletionStage
   java.util.concurrent.CompletionException
   java.util.concurrent.CompletableFuture
   java.util.function.BiConsumer))

(require-python 'asyncio)

(defn event-loop []
  (try
    (asyncio/get_running_loop)
    (catch Exception e nil)))

(let [{{:strs [report_future_done]} :globals}
      (py/run-simple-string "
import asyncio
def report_future_done(future, fn):
    def done(fut):
        fn()
    asyncio.ensure_future(future).add_done_callback(done)
")]
  (def ^:private report-future-done report_future_done))

(def ^:dynamic *keep-generic-python-object* false)

(defmacro with-preserve-python [& body]
  `(binding [*keep-generic-python-object* true] ~@body))

(defn coroutine->promise [coro]
  (if *keep-generic-python-object*
    coro
    (let [completable-future (CompletableFuture.)]
      (if (event-loop)
        (let [py-future (with-preserve-python
                          (asyncio/ensure_future coro))]
          (with-preserve-python
            (report-future-done
             py-future
             (fn []
               (if-let [ex (py. py-future exception)]
                 (.completeExceptionally completable-future ex)
                 (let [result (py. py-future result)]
                   (.complete completable-future result)))))))
        (future
          (try
            (let [async-result (asyncio/run coro)]
              (.complete completable-future async-result))
            (catch Exception e
              (.completeExceptionally completable-future e)))))
      completable-future)))

(defn promise->coroutine [^CompletionStage it]
  (let [loop (asyncio/get_running_loop)
        future (with-preserve-python (py. loop create_future))]
    (.whenComplete it
                   ^BiConsumer
                   (reify BiConsumer
                     (accept [_ v e]
                       (if e
                         (if (instance? CompletionException e)
                           (py. future set_exception (.getCause ^Exception e))
                           (py. future set_exception e))
                         (py. future set_result v)))))
    future))

(extend-protocol pyprot/PBridgeToPython
  java.util.concurrent.CompletableFuture
  (as-python [item _]
    (promise->coroutine item)))

; Defmethod py
(defmethod pyprot/pyobject-as-jvm :coroutine [pyobj & [_]]
  (let [generic-ob (py-bridge-jvm/generic-python-as-jvm (delay pyobj))]
    (coroutine->promise generic-ob)))

(defmethod pyprot/pyobject-as-jvm :future [pyobj & [_]]
  (let [generic-ob (py-bridge-jvm/generic-python-as-jvm (delay pyobj))]
    (coroutine->promise generic-ob)))

(defmethod pyprot/pyobject-as-jvm :task [pyobj & [_]]
  (let [generic-ob (py-bridge-jvm/generic-python-as-jvm (delay pyobj))]
    (coroutine->promise generic-ob)))