@cycle/run - npm Package Compare versions

Comparing version 1.0.0-rc.8 to 1.0.0-rc.9

CHANGELOG.md

@@ -0,1 +1,20 @@
+<a name="1.0.0-rc.9"></a>
+# 1.0.0-rc.9 (2017-02-08)
+
+
+### Bug Fixes
+
+* **run:** fix race condition for drivers that subscribe late ([58b7991](https://github.com/cyclejs/cyclejs/tree/master/run/commit/58b7991))
+* **run:** sink proxy completes on dispose, not with setTimeout ([47931fc](https://github.com/cyclejs/cyclejs/tree/master/run/commit/47931fc))
+
+
+### BREAKING CHANGES
+
+* run: if you are using sources or sinks from the output of
+Cycle setup(), you may see different behavior of the complete
+notification, which now happens always when run's dispose() is called.
+The complete notifications from main's sinks are ignored.
+
+
+
 <a name="1.0.0-rc.8"></a>
@@ -2,0 +21,0 @@ # 1.0.0-rc.8 (2017-02-03)

dist/cycle-run.js

@@ -64,7 +64,7 @@ (function(f){if(typeof exports==="object"&&typeof module!=="undefined"){module.exports=f()}else if(typeof define==="function"&&define.amd){define([],f)}else{var g;if(typeof window!=="undefined"){g=window}else if(typeof global!=="undefined"){g=global}else if(typeof self!=="undefined"){g=self}else{g=this}g.Cycle = f()}})(function(){var define,module,exports;return (function e(t,n,r){function s(o,u){if(!n[o]){if(!t[o]){var a=typeof require=="function"&&require;if(!u&&a)return a(o,!0);if(i)return i(o,!0);var f=new Error("Cannot find module '"+o+"'");throw f.code="MODULE_NOT_FOUND",f}var l=n[o]={exports:{}};t[o][0].call(l.exports,function(e){var n=t[o][1][e];return s(n?n:e)},l,l.exports,e,t,n,r)}return n[o].exports}var i=typeof require=="function"&&require;for(var o=0;o<r.length;o++)s(r[o]);return s})({1:[function(require,module,exports){
     sinkNames.forEach(function (name) {
-        buffers[name] = { _n: [], _e: [], _c: [] };
+        buffers[name] = { _n: [], _e: [] };
         replicators[name] = {
             next: function (x) { return buffers[name]._n.push(x); },
             error: function (err) { return buffers[name]._e.push(err); },
-            complete: function () { return buffers[name]._c.push(null); },
+            complete: function () { },
         };
@@ -74,7 +74,2 @@ });
         .map(function (name) { return xstream_1.default.fromObservable(sinks[name]).subscribe(replicators[name]); });
-    // A sink proxy should not complete before 500 milliseconds.
-    // This is to allow late drivers (drivers that subscribe to the sink proxy
-    // asynchronously later, not immediately when the driver is setup) to
-    // have time to receive the 'next' values from the MemoryStream sink proxy.
-    var EARLIEST_SINK_COMPLETE = 500; // milliseconds
     sinkNames.forEach(function (name) {
@@ -84,12 +79,10 @@ var listener = sinkProxies[name];
         var error = function (err) { logToConsoleError(err); listener._e(err); };
-        var complete = function () { setTimeout(function () { listener._c(); }, EARLIEST_SINK_COMPLETE); };
         buffers[name]._n.forEach(next);
         buffers[name]._e.forEach(error);
-        buffers[name]._c.forEach(complete);
         replicators[name].next = next;
         replicators[name].error = error;
-        replicators[name].complete = complete;
+        // because sink.subscribe(replicator) had mutated replicator to add
+        // _n, _e, _c, we must also update these:
         replicators[name]._n = next;
         replicators[name]._e = error;
-        replicators[name]._c = complete;
     });
@@ -99,2 +92,3 @@ buffers = null; // free up for GC
         subscriptions.forEach(function (s) { return s.unsubscribe(); });
+        sinkNames.forEach(function (name) { return sinkProxies[name]._c(); });
     };
@@ -101,0 +95,0 @@ }

dist/cycle-run.min.js

@@ -1,1 +0,1 @@
-(function(f){if(typeof exports==="object"&&typeof module!=="undefined"){module.exports=f()}else if(typeof define==="function"&&define.amd){define([],f)}else{var g;if(typeof window!=="undefined"){g=window}else if(typeof global!=="undefined"){g=global}else if(typeof self!=="undefined"){g=self}else{g=this}g.Cycle=f()}})(function(){var define,module,exports;return function e(t,n,r){function s(o,u){if(!n[o]){if(!t[o]){var a=typeof require=="function"&&require;if(!u&&a)return a(o,!0);if(i)return i(o,!0);var f=new Error("Cannot find module '"+o+"'");throw f.code="MODULE_NOT_FOUND",f}var l=n[o]={exports:{}};t[o][0].call(l.exports,function(e){var n=t[o][1][e];return s(n?n:e)},l,l.exports,e,t,n,r)}return n[o].exports}var i=typeof require=="function"&&require;for(var o=0;o<r.length;o++)s(r[o]);return s}({1:[function(require,module,exports){"use strict";var adaptStream=function(x){return x};function setAdapt(f){adaptStream=f}exports.setAdapt=setAdapt;function adapt(stream){return adaptStream(stream)}exports.adapt=adapt},{}],2:[function(require,module,exports){(function(global){"use strict";var xstream_1=typeof window!=="undefined"?window["xstream"]:typeof global!=="undefined"?global["xstream"]:null;var adapt_1=require("./adapt");function logToConsoleError(err){var target=err.stack||err;if(console&&console.error){console.error(target)}else if(console&&console.log){console.log(target)}}function makeSinkProxies(drivers){var sinkProxies={};for(var name_1 in drivers){if(drivers.hasOwnProperty(name_1)){sinkProxies[name_1]=xstream_1.default.createWithMemory()}}return sinkProxies}function callDrivers(drivers,sinkProxies){var sources={};for(var name_2 in drivers){if(drivers.hasOwnProperty(name_2)){sources[name_2]=drivers[name_2](sinkProxies[name_2],name_2);if(sources[name_2]&&typeof sources[name_2]==="object"){sources[name_2]._isCycleSource=name_2}}}return sources}function adaptSources(sources){for(var name_3 in sources){if(sources.hasOwnProperty(name_3)&&sources[name_3]&&typeof sources[name_3]["shamefullySendNext"]==="function"){sources[name_3]=adapt_1.adapt(sources[name_3])}}return sources}function replicateMany(sinks,sinkProxies){var sinkNames=Object.keys(sinks).filter(function(name){return!!sinkProxies[name]});var buffers={};var replicators={};sinkNames.forEach(function(name){buffers[name]={_n:[],_e:[],_c:[]};replicators[name]={next:function(x){return buffers[name]._n.push(x)},error:function(err){return buffers[name]._e.push(err)},complete:function(){return buffers[name]._c.push(null)}}});var subscriptions=sinkNames.map(function(name){return xstream_1.default.fromObservable(sinks[name]).subscribe(replicators[name])});var EARLIEST_SINK_COMPLETE=500;sinkNames.forEach(function(name){var listener=sinkProxies[name];var next=function(x){listener._n(x)};var error=function(err){logToConsoleError(err);listener._e(err)};var complete=function(){setTimeout(function(){listener._c()},EARLIEST_SINK_COMPLETE)};buffers[name]._n.forEach(next);buffers[name]._e.forEach(error);buffers[name]._c.forEach(complete);replicators[name].next=next;replicators[name].error=error;replicators[name].complete=complete;replicators[name]._n=next;replicators[name]._e=error;replicators[name]._c=complete});buffers=null;return function disposeReplication(){subscriptions.forEach(function(s){return s.unsubscribe()})}}function disposeSources(sources){for(var k in sources){if(sources.hasOwnProperty(k)&&sources[k]&&sources[k].dispose){sources[k].dispose()}}}function isObjectEmpty(obj){return Object.keys(obj).length===0}function setup(main,drivers){if(typeof main!=="function"){throw new Error("First argument given to Cycle must be the 'main' "+"function.")}if(typeof drivers!=="object"||drivers===null){throw new Error("Second argument given to Cycle must be an object "+"with driver functions as properties.")}if(isObjectEmpty(drivers)){throw new Error("Second argument given to Cycle must be an object "+"with at least one driver function declared as a property.")}var sinkProxies=makeSinkProxies(drivers);var sources=callDrivers(drivers,sinkProxies);var adaptedSources=adaptSources(sources);var sinks=main(adaptedSources);if(typeof window!=="undefined"){window.Cyclejs=window.Cyclejs||{};window.Cyclejs.sinks=sinks}function run(){var disposeReplication=replicateMany(sinks,sinkProxies);return function dispose(){disposeSources(sources);disposeReplication()}}return{sinks:sinks,sources:sources,run:run}}exports.setup=setup;function run(main,drivers){var _a=setup(main,drivers),run=_a.run,sinks=_a.sinks;if(typeof window!=="undefined"&&window["CyclejsDevTool_startGraphSerializer"]){window["CyclejsDevTool_startGraphSerializer"](sinks)}return run()}exports.run=run;Object.defineProperty(exports,"__esModule",{value:true});exports.default=run}).call(this,typeof global!=="undefined"?global:typeof self!=="undefined"?self:typeof window!=="undefined"?window:{})},{"./adapt":1}]},{},[2])(2)});
+(function(f){if(typeof exports==="object"&&typeof module!=="undefined"){module.exports=f()}else if(typeof define==="function"&&define.amd){define([],f)}else{var g;if(typeof window!=="undefined"){g=window}else if(typeof global!=="undefined"){g=global}else if(typeof self!=="undefined"){g=self}else{g=this}g.Cycle=f()}})(function(){var define,module,exports;return function e(t,n,r){function s(o,u){if(!n[o]){if(!t[o]){var a=typeof require=="function"&&require;if(!u&&a)return a(o,!0);if(i)return i(o,!0);var f=new Error("Cannot find module '"+o+"'");throw f.code="MODULE_NOT_FOUND",f}var l=n[o]={exports:{}};t[o][0].call(l.exports,function(e){var n=t[o][1][e];return s(n?n:e)},l,l.exports,e,t,n,r)}return n[o].exports}var i=typeof require=="function"&&require;for(var o=0;o<r.length;o++)s(r[o]);return s}({1:[function(require,module,exports){"use strict";var adaptStream=function(x){return x};function setAdapt(f){adaptStream=f}exports.setAdapt=setAdapt;function adapt(stream){return adaptStream(stream)}exports.adapt=adapt},{}],2:[function(require,module,exports){(function(global){"use strict";var xstream_1=typeof window!=="undefined"?window["xstream"]:typeof global!=="undefined"?global["xstream"]:null;var adapt_1=require("./adapt");function logToConsoleError(err){var target=err.stack||err;if(console&&console.error){console.error(target)}else if(console&&console.log){console.log(target)}}function makeSinkProxies(drivers){var sinkProxies={};for(var name_1 in drivers){if(drivers.hasOwnProperty(name_1)){sinkProxies[name_1]=xstream_1.default.createWithMemory()}}return sinkProxies}function callDrivers(drivers,sinkProxies){var sources={};for(var name_2 in drivers){if(drivers.hasOwnProperty(name_2)){sources[name_2]=drivers[name_2](sinkProxies[name_2],name_2);if(sources[name_2]&&typeof sources[name_2]==="object"){sources[name_2]._isCycleSource=name_2}}}return sources}function adaptSources(sources){for(var name_3 in sources){if(sources.hasOwnProperty(name_3)&&sources[name_3]&&typeof sources[name_3]["shamefullySendNext"]==="function"){sources[name_3]=adapt_1.adapt(sources[name_3])}}return sources}function replicateMany(sinks,sinkProxies){var sinkNames=Object.keys(sinks).filter(function(name){return!!sinkProxies[name]});var buffers={};var replicators={};sinkNames.forEach(function(name){buffers[name]={_n:[],_e:[]};replicators[name]={next:function(x){return buffers[name]._n.push(x)},error:function(err){return buffers[name]._e.push(err)},complete:function(){}}});var subscriptions=sinkNames.map(function(name){return xstream_1.default.fromObservable(sinks[name]).subscribe(replicators[name])});sinkNames.forEach(function(name){var listener=sinkProxies[name];var next=function(x){listener._n(x)};var error=function(err){logToConsoleError(err);listener._e(err)};buffers[name]._n.forEach(next);buffers[name]._e.forEach(error);replicators[name].next=next;replicators[name].error=error;replicators[name]._n=next;replicators[name]._e=error});buffers=null;return function disposeReplication(){subscriptions.forEach(function(s){return s.unsubscribe()});sinkNames.forEach(function(name){return sinkProxies[name]._c()})}}function disposeSources(sources){for(var k in sources){if(sources.hasOwnProperty(k)&&sources[k]&&sources[k].dispose){sources[k].dispose()}}}function isObjectEmpty(obj){return Object.keys(obj).length===0}function setup(main,drivers){if(typeof main!=="function"){throw new Error("First argument given to Cycle must be the 'main' "+"function.")}if(typeof drivers!=="object"||drivers===null){throw new Error("Second argument given to Cycle must be an object "+"with driver functions as properties.")}if(isObjectEmpty(drivers)){throw new Error("Second argument given to Cycle must be an object "+"with at least one driver function declared as a property.")}var sinkProxies=makeSinkProxies(drivers);var sources=callDrivers(drivers,sinkProxies);var adaptedSources=adaptSources(sources);var sinks=main(adaptedSources);if(typeof window!=="undefined"){window.Cyclejs=window.Cyclejs||{};window.Cyclejs.sinks=sinks}function run(){var disposeReplication=replicateMany(sinks,sinkProxies);return function dispose(){disposeSources(sources);disposeReplication()}}return{sinks:sinks,sources:sources,run:run}}exports.setup=setup;function run(main,drivers){var _a=setup(main,drivers),run=_a.run,sinks=_a.sinks;if(typeof window!=="undefined"&&window["CyclejsDevTool_startGraphSerializer"]){window["CyclejsDevTool_startGraphSerializer"](sinks)}return run()}exports.run=run;Object.defineProperty(exports,"__esModule",{value:true});exports.default=run}).call(this,typeof global!=="undefined"?global:typeof self!=="undefined"?self:typeof window!=="undefined"?window:{})},{"./adapt":1}]},{},[2])(2)});

lib/index.d.ts

 import { MemoryStream } from 'xstream';
 export interface FantasyObserver {
-    next: (x: any) => void;
-    error: (err: any) => void;
-    complete: (c?: any) => void;
+    next(x: any): void;
+    error(err: any): void;
+    complete(c?: any): void;
 }
 export interface FantasySubscription {
-    unsubscribe: () => void;
+    unsubscribe(): void;
 }
@@ -47,3 +47,3 @@ export interface FantasyObservable {
     sinks: Si;
-    run: () => DisposeFunction;
+    run(): DisposeFunction;
 }
@@ -50,0 +50,0 @@ /**

lib/index.js

@@ -50,7 +50,7 @@ "use strict";
     sinkNames.forEach(function (name) {
-        buffers[name] = { _n: [], _e: [], _c: [] };
+        buffers[name] = { _n: [], _e: [] };
         replicators[name] = {
             next: function (x) { return buffers[name]._n.push(x); },
             error: function (err) { return buffers[name]._e.push(err); },
-            complete: function () { return buffers[name]._c.push(null); },
+            complete: function () { },
         };
@@ -60,7 +60,2 @@ });
         .map(function (name) { return xstream_1.default.fromObservable(sinks[name]).subscribe(replicators[name]); });
-    // A sink proxy should not complete before 500 milliseconds.
-    // This is to allow late drivers (drivers that subscribe to the sink proxy
-    // asynchronously later, not immediately when the driver is setup) to
-    // have time to receive the 'next' values from the MemoryStream sink proxy.
-    var EARLIEST_SINK_COMPLETE = 500; // milliseconds
     sinkNames.forEach(function (name) {
@@ -70,12 +65,10 @@ var listener = sinkProxies[name];
         var error = function (err) { logToConsoleError(err); listener._e(err); };
-        var complete = function () { setTimeout(function () { listener._c(); }, EARLIEST_SINK_COMPLETE); };
         buffers[name]._n.forEach(next);
         buffers[name]._e.forEach(error);
-        buffers[name]._c.forEach(complete);
         replicators[name].next = next;
         replicators[name].error = error;
-        replicators[name].complete = complete;
+        // because sink.subscribe(replicator) had mutated replicator to add
+        // _n, _e, _c, we must also update these:
         replicators[name]._n = next;
         replicators[name]._e = error;
-        replicators[name]._c = complete;
     });
@@ -85,2 +78,3 @@ buffers = null; // free up for GC
         subscriptions.forEach(function (s) { return s.unsubscribe(); });
+        sinkNames.forEach(function (name) { return sinkProxies[name]._c(); });
     };
@@ -87,0 +81,0 @@ }

package.json

 {
   "name": "@cycle/run",
-  "version": "1.0.0-rc.8",
+  "version": "1.0.0-rc.9",
   "description": "The Cycle.js run() function to use with xstream",
@@ -47,3 +47,3 @@ "license": "MIT",
   "scripts": {
-    "lint": "../node_modules/.bin/tslint -c ../tslint.json src/**/*.ts",
+    "lint": "../node_modules/.bin/tslint --config ../tslint.json --project tsconfig.json",
     "prelib": "rm -rf lib/ && mkdir -p lib/",
@@ -50,0 +50,0 @@ "lib": "../node_modules/.bin/tsc",

src/index.ts

@@ -5,9 +5,9 @@ import xs, {Stream, MemoryStream} from 'xstream';
 export interface FantasyObserver {
-  next: (x: any) => void;
-  error: (err: any) => void;
-  complete: (c?: any) => void;
+  next(x: any): void;
+  error(err: any): void;
+  complete(c?: any): void;
 }
 
 export interface FantasySubscription {
-  unsubscribe: () => void;
+  unsubscribe(): void;
 }
@@ -59,3 +59,3 @@
   sinks: Si;
-  run: () => DisposeFunction;
+  run(): DisposeFunction;
 }
@@ -74,3 +74,3 @@
   const sinkProxies: SinkProxies = {};
-  for (let name in drivers) {
+  for (const name in drivers) {
     if (drivers.hasOwnProperty(name)) {
@@ -85,3 +85,3 @@ sinkProxies[name] = xs.createWithMemory<any>();
   const sources = {};
-  for (let name in drivers) {
+  for (const name in drivers) {
     if (drivers.hasOwnProperty(name)) {
@@ -99,3 +99,3 @@ sources[name] = drivers[name](sinkProxies[name], name);
 function adaptSources<So extends Object>(sources: So): So {
-  for (let name in sources) {
+  for (const name in sources) {
     if (sources.hasOwnProperty(name)
@@ -110,10 +110,15 @@ && sources[name]
 
+/**
+ * Notice that we do not replicate 'complete' from real sinks, in
+ * SinksReplicators and ReplicationBuffers.
+ * Complete is triggered only on disposeReplication. See discussion in #425
+ * for details.
+ */
 interface SinkReplicators {
   [name: string]: {
-    next: (x: any) => void;
-    _n?: (x: any) => void;
-    error: (err: any) => void;
-    _e?: (err: any) => void;
-    complete: () => void;
-    _c?: () => void;
+    next(x: any): void;
+    _n?(x: any): void;
+    error(err: any): void;
+    _e?(err: any): void;
+    complete(): void;
   };
@@ -126,3 +131,2 @@ }
     _e: Array<any>;
-    _c: Array<null>;
   };
@@ -137,7 +141,7 @@ }
   sinkNames.forEach((name) => {
-    buffers[name] = {_n: [], _e: [], _c: []};
+    buffers[name] = {_n: [], _e: []};
     replicators[name] = {
       next: (x: any) => buffers[name]._n.push(x),
       error: (err: any) => buffers[name]._e.push(err),
-      complete: () => buffers[name]._c.push(null),
+      complete: () => {},
     };
@@ -149,8 +153,2 @@ });
 
-  // A sink proxy should not complete before 500 milliseconds.
-  // This is to allow late drivers (drivers that subscribe to the sink proxy
-  // asynchronously later, not immediately when the driver is setup) to
-  // have time to receive the 'next' values from the MemoryStream sink proxy.
-  const EARLIEST_SINK_COMPLETE = 500; // milliseconds
-
   sinkNames.forEach((name) => {
@@ -160,12 +158,10 @@ const listener = sinkProxies[name];
     const error = (err: any) => { logToConsoleError(err); listener._e(err); };
-    const complete = () => { setTimeout(() => { listener._c(); }, EARLIEST_SINK_COMPLETE); };
     buffers[name]._n.forEach(next);
     buffers[name]._e.forEach(error);
-    buffers[name]._c.forEach(complete);
     replicators[name].next = next;
     replicators[name].error = error;
-    replicators[name].complete = complete;
+    // because sink.subscribe(replicator) had mutated replicator to add
+    // _n, _e, _c, we must also update these:
     replicators[name]._n = next;
     replicators[name]._e = error;
-    replicators[name]._c = complete;
   });
@@ -176,2 +172,3 @@ buffers = null as any; // free up for GC
     subscriptions.forEach(s => s.unsubscribe());
+    sinkNames.forEach((name) => sinkProxies[name]._c());
   };
@@ -181,3 +178,3 @@ }
 function disposeSources<So>(sources: So) {
-  for (let k in sources) {
+  for (const k in sources) {
     if (sources.hasOwnProperty(k) && sources[k] && (sources[k] as any).dispose) {
@@ -184,0 +181,0 @@ (sources[k] as any).dispose();

test/index.ts

@@ -99,7 +99,6 @@ import 'mocha';
         assert.strictEqual(x, 97);
-        dispose();
-        done();
+        dispose(); // will trigger this listener's complete
       },
       error: err => done(err),
-      complete: () => done('complete should not be called'),
+      complete: () => done(),
     });
@@ -200,10 +199,7 @@ dispose = run();
         if (x === 'x2') {
-          dispose();
-          setTimeout(() => {
-            done();
-          }, 100);
+          dispose(); // will trigger this listener's complete
         }
       },
       error: err => done(err),
-      complete: () => done('complete should not be called'),
+      complete: () => done(),
     });
@@ -210,0 +206,0 @@ dispose = run();

tsconfig.json

 {
+  "extends": "../tsconfig.common.json",
   "compilerOptions": {
-    "removeComments": false,
-    "preserveConstEnums": true,
-    "sourceMap": true,
-    "declaration": true,
-    "noImplicitAny": true,
-    "strictNullChecks": true,
-    "suppressImplicitAnyIndexErrors": true,
-    "module": "commonjs",
-    "target": "es5",
     "outDir": "lib/"
   },
-  "formatCodeOptions": {
-    "indentSize": 2,
-    "tabSize": 2
-  },
-   "files": [
-     "src/index.ts",
-     "src/adapt.ts"
-   ]
+  "files": [
+    "src/index.ts",
+    "src/adapt.ts"
+  ]
 }

New alerts

License Policy Violation

License

This package is not allowed per your license policy. Review the package's license to ensure compliance.

Found 1 instance in 1 package

Fixed alerts

License Policy Violation

License

This package is not allowed per your license policy. Review the package's license to ensure compliance.

Found 1 instance in 1 package

Worsened metrics

Total package byte prevSize

55823

decreased by-2.44%

Lines of code

1079

decreased by-2.71%

No dependency changes