Fix take operator not supporting take(0) and not respecting Close (#47)

* Fix take operator with max <= 0

* Fix passesSinkClose for take operator

The take operator needs to keep track of ended
actively to avoid confusion with the state.taken
number and the max <= 0 case.

* Add !state.ended condition to take's Push signal

src/wonka_operators.re

···
   switchMap((. x) => x, source);
 
 type takeStateT = {
+  mutable ended: bool,
   mutable taken: int,
   mutable talkback: (. talkbackT) => unit,
 };
···
 let take = (max: int): operatorT('a, 'a) =>
   curry(source =>
     curry(sink => {
-      let state: takeStateT = {taken: 0, talkback: talkbackPlaceholder};
+      let state: takeStateT = {
+        ended: false,
+        taken: 0,
+        talkback: talkbackPlaceholder,
+      };
 
       source((. signal) =>
         switch (signal) {
+        | Start(tb) when max <= 0 =>
+          state.ended = true;
+          sink(. End);
+          tb(. Close);
         | Start(tb) => state.talkback = tb
-        | Push(_) when state.taken < max =>
+        | Push(_) when state.taken < max && !state.ended =>
           state.taken = state.taken + 1;
           sink(. signal);
-
-          if (state.taken === max) {
+          if (!state.ended && state.taken >= max) {
+            state.ended = true;
             sink(. End);
             state.talkback(. Close);
           };
         | Push(_) => ()
-        | End when state.taken < max =>
-          state.taken = max;
+        | End when !state.ended =>
+          state.ended = true;
           sink(. End);
         | End => ()
         }
···
       sink(.
         Start(
           (. signal) =>
-            if (state.taken < max) {
+            if (!state.ended) {
               switch (signal) {
-              | Pull => state.talkback(. Pull)
+              | Pull when state.taken < max => state.talkback(. Pull)
+              | Pull => ()
               | Close =>
-                state.taken = max;
+                state.ended = true;
                 state.talkback(. Close);
               };
             },

src/wonka_operators.test.ts

···
         if (tb === deriving.pull) {
           pulls++;
           sink(deriving.end());
-          sink(deriving.push(0));
+          sink(deriving.push(123));
         }
       }));
     };
···
   const noop = operators.take(10);
   passesPassivePull(noop);
   passesActivePush(noop);
-  // TODO: passesSinkClose(noop);
+  passesSinkClose(noop);
   passesSourceEnd(noop);
   passesSingleStart(noop);
   passesStrictEnd(noop);
   passesAsyncSequence(noop);
 
-  // TODO: take(0) seems to be broken
-  const ending = operators.take(1);
-  passesCloseAndEnd(ending);
+  passesCloseAndEnd(operators.take(0));
 
   it('emits values until a maximum is reached', () => {
     const { source, next } = sources.makeSubject<number>();