feat(engine.producer): add clean-up callback for producers

docs/docs/api/producer.md

@@ -42,6 +42,10 @@ body.
 
 <img src='/static/img/producer.jpg' alt='Producer' />
 
+Additionally, the producer function can return a callback for clean-up purposes.
+   This will be called when the producer will be unmounted from the state and thus
+   no longer in operation.
+
 ## Running a producer
 
 A `producer` can not be called directly. Engine considers a `producer` for
@@ -78,6 +82,19 @@ const todoCounter: producer = ({
 );
 ```
 
+## Example with clean-up
+```tsx
+const subscriber: producer = ({
+  something = update.something
+}) => (
+  const stream = subscribeToStream(value => {
+    something.set(value)
+  })
+  return () => {
+    stream.unsubscribe()
+  }
+);
+```
 
 ## Parts
 
@@ -188,6 +205,12 @@ changed through the `update` operation.
 This means, `producer`s will be pushing new data to the Engine which in turn
 trigger other producers to execute and in turn update the state.
 
+#### Clean-up
+
+Producers can be long running by subscribing to streams, initiating long calls, using
+   timers, etc. As such, when the `producer` is unmounted from the state, it is the
+   `producer`'s responsability to provide a callback for clean-up purposes. See the
+   example above.
 
 ## Best practices
 

packages/engine.producer/specs/producer.spec.ts

@@ -589,6 +589,27 @@ test("should support the full api for the get operation", () => {
   jest.runAllTimers();
 });
 
+test("should support unmount lifecycle method", () => {
+  const struct: producer = ({
+    a = update.a,
+  }) => {
+    const id = setInterval(() => {
+      a.push('a')
+    }, 500)
+    return () => {
+      clearInterval(id)
+    }
+  };
+  const result = run(struct, {
+    a: []
+  });
+  jest.advanceTimersByTime(1000);
+  result.producer.unmount()
+  jest.advanceTimersByTime(1000);
+  jest.runOnlyPendingTimers();
+  expect(result.db.get("/a")).toEqual(['a', 'a']);
+});
+
 /*
 test("should allow args composition", () => {
   const state = {

packages/engine.producer/src/graph/graph.ts

@@ -10,12 +10,12 @@ import {
   GraphStructure,
   GraphNodeType,
   ValueTypes,
+  DatastoreInstance,
 } from "@c11/engine.types";
 import { resolveDependencies } from "./resolveDependencies";
 import { getExternalNodes } from "./getExternalNodes";
 import { getInternalNodes } from "./getInternalNodes";
 import { resolveOrder } from "./resolveOrder";
-import { DatastoreInstance } from "@c11/engine.types";
 import { observeOperation } from "./observeOperation";
 import { ComputeType, computeOperation } from "./computeOperation";
 import { pathListener } from "./pathListener";

packages/engine.producer/src/producer.ts

@@ -8,6 +8,7 @@ import {
   StructOperation,
   ProducerMeta,
 } from "@c11/engine.types";
+import isFunction from "lodash/isFunction";
 import shortid from "shortid";
 
 import { Graph } from "./graph";
@@ -30,6 +31,7 @@ export class Producer implements ProducerInstance {
   private debug: boolean;
   private keepReferences: string[];
   private meta: ProducerMeta;
+  private results: any[] = [];
   private stats = {
     executionCount: 0,
   };
@@ -47,13 +49,16 @@ export class Producer implements ProducerInstance {
       this.db,
       this.external,
       this.args,
-      this.debug ? this.fnWrapper.bind(this) : this.fn,
+      this.fnWrapper.bind(this),
       this.keepReferences
     );
   }
   private fnWrapper(...params: any[]) {
     this.stats.executionCount += 1;
-    this.fn.apply(null, params);
+    const result = this.fn.apply(null, params);
+    if (result !== undefined && isFunction(result)) {
+      this.results.push(result)
+    }
   }
   mount() {
     if (this.state === ProducerStates.MOUNTED) {
@@ -69,6 +74,11 @@ export class Producer implements ProducerInstance {
     }
     this.graph.destroy();
     this.state = ProducerStates.UNMOUNTED;
+    this.results.forEach(x => {
+      if (x) {
+        x()
+      }
+    })
     return this;
   }
   updateExternal(props: ProducerContext["props"]) {