log handles pull behaviors, better prefix handling

README.md

@@ -611,6 +611,18 @@ Returns a stream that occurs `ms` milliseconds after `s` occurs.
 Returns a stream that after occurring, ignores the next occurrences in
 `ms` milliseconds.
 
+#### `stream.log(prefix?: string)`
+
+The log method on streams logs the value of every occurrence using
+`console.log`. It is intended to be used for debugging streams during
+development.
+
+The option `prefix` argument will be logged along with every value if specified.
+
+```
+myStream.log("myStream:");
+```
+
 ### Behavior
 
 #### `Behavior.of<A>(a: A): Behavior<A>`
@@ -694,6 +706,22 @@ Integrate behavior with respect to time.
 
 The value of the behavior is treated as a rate of change per millisecond.
 
+#### `behavior.log(prefix?: string, ms: number = 100)`
+
+The log method on behaviors logs the value of the behavior whenever it changes
+using `console.log`. It is intended to be used for debugging behaviors during
+development.
+
+If the behavior is a pull behavior (i.e. it may change infinitely often) then
+changes will only be logged every `ms` milliseconds.
+
+The option `prefix` argument will be logged along with every value if specified.
+
+```
+myBehavior.log("myBehavior:");
+time.map(t => t * t).log("Time squared is:", 1000);
+```
+
 ### Now
 
 The Now monad represents a computation that takes place in a given

src/behavior.ts

@@ -101,8 +101,22 @@ export abstract class Behavior<A> extends Reactive<A, BListener>
       refresh(this, t);
     }
   }
-  log(prefix?: string): Behavior<A> {
-    this.subscribe((a) => console.log(`${prefix || ""} `, a));
+  log(prefix?: string, ms: number = 100): Behavior<A> {
+    this.observe(
+      (a) => (prefix !== undefined ? console.log(prefix, a) : console.log(a)),
+      (pull) => {
+        let stop = false;
+        (function repeat() {
+          if (!stop) {
+            pull();
+            setTimeout(repeat, ms);
+          }
+        })();
+        return () => {
+          stop = true;
+        };
+      }
+    );
     return this;
   }
 }

src/stream.ts

@@ -48,7 +48,9 @@ export abstract class Stream<A> extends Reactive<A, SListener<A>>
     return accumFrom(fn, init, this);
   }
   log(prefix?: string): Stream<A> {
-    this.subscribe((a) => console.log(`${prefix || ""} `, a));
+    this.subscribe((a) =>
+      prefix !== undefined ? console.log(prefix, a) : console.log(a)
+    );
     return this;
   }
   abstract pushS(t: number, value: any): void;

test/behavior.ts

@@ -967,4 +967,48 @@ describe("Behavior and Stream", () => {
       assert.deepEqual(pushSpy.args, [[0], [1], [2], [3]]);
     });
   });
+  describe("log", () => {
+    it("logs every change on push behavior", () => {
+      const origLog = console.log;
+      const strings = [];
+      console.log = (...s: string[]) => strings.push(s);
+      const b = sinkBehavior("hello");
+      b.log();
+      b.push("world");
+      console.log = origLog;
+      assert.deepEqual(strings, [["hello"], ["world"]]);
+    });
+    it("logs with prefix", () => {
+      const origLog = console.log;
+      const strings = [];
+      console.log = (...s: string[]) => strings.push(s);
+      const b = sinkBehavior("hello");
+      b.log("b");
+      b.push("world");
+      console.log = origLog;
+      assert.deepEqual(strings, [["b", "hello"], ["b", "world"]]);
+    });
+    it("logs on pull behavior", () => {
+      const clock = useFakeTimers();
+      const origLog = console.log;
+      const strings = [];
+      console.log = (...s: string[]) => strings.push(s);
+
+      let v = "zero";
+      const b = fromFunction(() => v);
+      b.log("b", 1000);
+
+      clock.tick(500);
+      v = "one";
+      clock.tick(500);
+      v = "two";
+      clock.tick(1000);
+      // Restore
+      console.log = origLog;
+      clock.restore();
+
+      console.log(strings);
+      assert.deepEqual(strings, [["b", "zero"], ["b", "one"], ["b", "two"]]);
+    });
+  });
 });

test/stream.ts

@@ -475,4 +475,29 @@ describe("stream", () => {
       }, /.*pull behavior.*/);
     });
   });
+  describe("log", () => {
+    it("logs every value", () => {
+      const origLog = console.log;
+      const strings = [];
+      console.log = (...s: string[]) => strings.push(s);
+      const s = H.sinkStream();
+      s.log();
+      s.push("hello");
+      s.push("world");
+      s.push("world");
+      console.log = origLog;
+      assert.deepEqual(strings, [["hello"], ["world"], ["world"]]);
+    });
+    it("logs with prefix", () => {
+      const origLog = console.log;
+      const strings = [];
+      console.log = (...s: string[]) => strings.push(s);
+      const s = H.sinkStream();
+      s.log("s");
+      s.push("hello");
+      s.push("world");
+      console.log = origLog;
+      assert.deepEqual(strings, [["s", "hello"], ["s", "world"]]);
+    });
+  });
 });