[Finalizer] WARN manifold.deferred - unconsumed deferred in error state, make sure you're using `catch`.
java.lang.NullPointerException: Cannot invoke "manifold.stream.core.IEventSource.take(Object, Object)" because "this.s" is null
	at manifold.stream$reduce$fn__31304$this__29295__auto____31308$fn__31309.invoke(stream.clj:738)
	at manifold.stream$reduce$fn__31304$this__29295__auto____31308.invoke(stream.clj:737)
	at manifold.stream$reduce$fn__31304.invoke(stream.clj:737)
	at manifold.deferred$eval29130$chain_SINGLEQUOTE____29151.invoke(deferred.clj:883)
	at manifold.stream$reduce.invokeStatic(stream.clj:733)
	at manifold.stream$reduce.invoke(stream.clj:722)
	at aleph.http.client_middleware$wrap_exceptions$fn__38826$fn__38873.invoke(client_middleware.clj:281)
	at clojure.lang.AFn.applyToHelper(AFn.java:154)
	at clojure.lang.AFn.applyTo(AFn.java:144)
	at clojure.core$apply.invokeStatic(core.clj:667)
	at clojure.core$with_bindings_STAR_.invokeStatic(core.clj:1990)
	at clojure.core$with_bindings_STAR_.doInvoke(core.clj:1990)
	at clojure.lang.RestFn.applyTo(RestFn.java:145)
	at clojure.core$apply.invokeStatic(core.clj:671)
	at clojure.core$bound_fn_STAR_$fn__5839.doInvoke(core.clj:2020)
	at clojure.lang.RestFn.invoke(RestFn.java:411)
	at manifold.deferred$eval29130$chain_SINGLEQUOTE____29151.invoke(deferred.clj:883)
	at manifold.deferred$eval29130$subscribe__29131$fn__29136.invoke(deferred.clj:849)
	at manifold.deferred.Listener.onSuccess(deferred.clj:345)
	at manifold.deferred.Deferred$fn__29000$fn__29001.invoke(deferred.clj:524)
	at clojure.lang.AFn.run(AFn.java:22)
	at io.aleph.dirigiste.Executor$3.run(Executor.java:325)
	at io.aleph.dirigiste.Executor$Worker$1.run(Executor.java:61)
	at manifold.executor$thread_factory$reify__28099$f__28100.invoke(executor.clj:71)
	at clojure.lang.AFn.run(AFn.java:22)
	at java.base/java.lang.Thread.run(Thread.java:1583)

(defn get-retry-429
  "Performs an HTTP get, but if it receives a 429 error, it retries
  after 15 seconds, for up to five times."
  ([url]
   (get-retry-429 url {}))
  ([url options]
   (d/loop [retries 5]
     (-> (http/get url options)
         (d/catch clojure.lang.ExceptionInfo
             (fn [e] (let [info (ex-data e)]
                       (if (= (:status info) 429)
                         (if (pos? retries)
                           ;; Get delay
                           (let [seconds 15 #_(-> (get-in info [:headers "retry-after"])
                                                  parse-long)
                                 millis (* seconds 1000)]
                             (-> (d/timeout! (d/deferred) millis nil)
                                 (d/chain (fn [_] (d/recur (dec retries))))))
                           (throw e))
                         (throw e)))))))))