Add Close talkback signal to tapAll

src/operators/wonka_operator_tapAll.re

···
 
 let tapAll = (~onStart, ~onPush, ~onEnd) => {
   curry(source => curry(sink => {
+    let ended = ref(false);
+
     source((.signal) => {
       switch (signal) {
-      | Start(_) => onStart(.)
-      | Push(x) => onPush(.x)
-      | End => onEnd(.)
+      | Start(talkback) => {
+        onStart(.)
+
+        sink(.Start((.signal) => {
+          switch (signal) {
+          | Close when !ended^ => {
+            ended := true;
+            onEnd(.);
+          }
+          | Close => ()
+          | _ => talkback(.signal)
+          }
+        }));
+      }
+      | Push(x) => {
+        onPush(.x)
+        sink(.signal);
+      }
+      | End when !ended^ => {
+        ended := true;
+        onEnd(.)
+      }
+      | _ => ()
       };
 
-      sink(.signal);
     });
   }));
 };