Fix dropped errors in manifold.time/in

The result deferred `d` may end up in an error state when `(f)` throws an exception or returns an
error deferred. Now since the cancellation callback was only attached via `d/chain` for
side-effect (i.e. the resulting deferred is not returned), the dropped error detection would be
triggered in this case. The fix is simply to also attach an error handler with `d/on-realized`.

Furthermore, putting the result deferred in an error state would not have cancelled the timer
because there was no error handler attached to it. This is now also fixed and the tests are extended
to cover the cancellation API, too.

The latter uncovered an oversight in the mock clock implementation where the `in` method didn't
return a cancellation function, resulting in an NPE. This is now also fixed.

Author: DerGuteMoritz

src/manifold/time.clj

@@ -125,7 +125,7 @@
 ;;;
 
 (in-ns 'manifold.deferred)
-(clojure.core/declare success! error! deferred realized? chain connect)
+(clojure.core/declare success! error! deferred realized? chain connect on-realized)
 (in-ns 'manifold.time)
 
 ;;;
@@ -191,8 +191,14 @@
      (reify
        IClock
        (in [this interval-millis f]
-         (swap! events update-in [(p/+ ^double @now interval-millis)] #(conj (or % []) f))
-         (advance this 0))
+         (let [t (p/+ ^double @now interval-millis)
+               cancel-fn (fn []
+                           (swap! events #(cond-> %
+                                            (contains? % t)
+                                            (update t disj f))))]
+           (swap! events update t (fnil conj #{}) f)
+           (advance this 0)
+           cancel-fn))
        (every [this delay-millis period-millis f]
          (assert (p/< 0.0 period-millis))
          (let [period    (atom period-millis)
@@ -263,7 +269,9 @@
                         (catch Throwable e
                           (manifold.deferred/error! d e)))))
         cancel-fn (.in *clock* interval f)]
-    (manifold.deferred/chain d (fn [_] (cancel-fn)))
+    (manifold.deferred/on-realized d
+                                   (fn [_] (cancel-fn))
+                                   (fn [_] (cancel-fn)))
     d))
 
 (defn every

test/manifold/time_test.clj

@@ -24,7 +24,23 @@
   (testing "delayed function returns failed deferred"
     (let [d (d/deferred)]
       (d/error! d (Exception. "BOOM"))
-      (is (thrown? Exception @(t/in 1 (fn [] d)))))))
+      (is (thrown? Exception @(t/in 1 (fn [] d))))))
+
+  (testing "cancelling by completing result deferred"
+    (let [c (t/mock-clock 0)]
+      (t/with-clock c
+        (testing "with success"
+          (let [n (atom 0)
+                d (t/in 1 #(swap! n inc))]
+            (d/success! d :cancel)
+            (t/advance c 1)
+            (is (= 0 @n))))
+        (testing "with error"
+          (let [n (atom 0)
+                d (t/in 1 #(swap! n inc))]
+            (d/error! d (Exception. "cancel"))
+            (t/advance c 1)
+            (is (= 0 @n))))))))
 
 (deftest test-every
   (let [n (atom 0)