callbag-start-with - npm Package Compare versions

Comparing version 2.1.1 to 3.0.0

index.js

 
 const startWith = (...xs) => inputSource => (start, outputSink) => {
   if (start !== 0) return;
+  let disposed = false;
   let inputTalkback;
-  inputSource(0, (it,id) => {
-    if (it === 0){
+  let trackPull = false;
+  let lastPull;
+
+  outputSink(0, (ot, od) => {
+    if (trackPull && ot === 1) {
+      lastPull = [1, od];
+    }
+
+    if (ot === 2) {
+      disposed = true;
+      xs.length = 0;
+    }
+
+    if (!inputTalkback) return;
+    inputTalkback(ot, od);
+  });
+
+  while (xs.length !== 0) {
+    if (xs.length === 1) {
+      trackPull = true;
+    }
+    outputSink(1, xs.shift());
+  }
+
+  if (disposed) return;
+
+  inputSource(0, (it, id) => {
+    if (it === 0) {
       inputTalkback = id;
-      outputSink(0, ot => {
-        if (ot === 2) inputTalkback(ot);
-      });
-      xs.forEach(x => outputSink(1, x));
-    } else {
-      outputSink(it, id);
+      trackPull = false;
+
+      if (lastPull) {
+        inputTalkback(...lastPull);
+        lastPull = null;
+      }
+      return;
     }
-    if (it !== 2) inputTalkback(1);
+    outputSink(it, id);
   });
@@ -17,0 +45,0 @@ };

package.json

 {
   "name": "callbag-start-with",
-  "version": "2.1.1",
+  "version": "3.0.0",
   "description": "Callbag operator that seeds a source with an initial data output",
@@ -21,5 +21,8 @@ "repository": {
     "callbag-from-iter": "~1.0.0",
-    "callbag-mock": "^2.1.0",
+    "callbag-mock": "^2.2.0",
+    "callbag-of": "^1.0.1",
+    "callbag-pipe": "^1.1.1",
+    "callbag-take": "^1.0.0",
     "tape": "^4.8.0"
   }
 }

test.js

@@ -5,2 +5,5 @@ const test = require('tape');
 const makeMockCallbag = require('callbag-mock');
+const of = require('callbag-of');
+const pipe = require('callbag-pipe');
+const take = require('callbag-take');
 const startWith = require('./index');
@@ -29,6 +32,3 @@
 test('it passes requests back up', t => {
-  let history = [];
-  const report = (t,d) => t !== 0 && history.push([t,d]);
-
-  const source = makeMockCallbag(report, true);
+  const source = makeMockCallbag(true);
   const seedWithFoo = startWith('foo');
@@ -39,9 +39,14 @@ const sink = makeMockCallbag();
 
-  sink.emit(1);
+  // emit request with data (request_id) to make sure that this very request is passed to the source
+  sink.emit(1, 'request_id');
   sink.emit(2);
 
-  t.deepEqual(history, [
-    [1, undefined],
-    [2, undefined],
-  ], 'source gets requests from sink');
+  t.deepEqual(
+    source.getMessages().slice(1),
+    [
+      [1, 'request_id'],
+      [2, undefined],
+    ],
+    'source gets requests from sink'
+  );
 
@@ -52,9 +57,9 @@ t.end();
 test('it supports iterables', t => {
-  const seededSrc = startWith('a')(fromIter(['b', 'c']));
-
   const expected = [];
 
-  forEach((v) => {
-    expected.push(v);
-  })(seededSrc);
+  pipe(
+    fromIter(['b', 'c']),
+    startWith('a'),
+    forEach((v) => expected.push(v))
+  );
 
@@ -76,3 +81,3 @@ t.deepEqual(expected, ['a', 'b', 'c'], 'sink gets data in the correct order');
   t.deepEqual(
-    sink.getReceivedData(), 
+    sink.getReceivedData(),
     ['foo','bar','baz','qu'],
@@ -84,1 +89,99 @@ 'sink gets seed and subsequent data'
 });
+
+test('it queues sync completion', t => {
+  const seededSrc = startWith('a', 'b', 'c')(of('d'));
+
+  const sink = makeMockCallbag();
+
+  seededSrc(0, sink);
+
+  t.deepEqual(
+    sink.getMessages().slice(1),
+    [
+      [1, 'a'],
+      [1, 'b'],
+      [1, 'c'],
+      [1, 'd'],
+      [2, undefined],
+    ],
+    'sink gets data in the correct order'
+  );
+
+  t.end();
+});
+
+test('it doesn\'t request data when receiving uknown type', t => {
+  const autoPull = () => {};
+
+  const source = makeMockCallbag(true);
+
+  pipe(
+    source,
+    startWith('foo'),
+    forEach(autoPull)
+  )
+
+  source.emit(1, 'a');
+  source.emit(1, 'b');
+  source.emit('unknown', 'c');
+  source.emit(1, 'd');
+  source.emit('unknown', 'e');
+
+  t.deepEqual(
+    source.getMessages().slice(1),
+    [
+      [1, undefined], // request sent up on 0
+      [1, undefined],
+      [1, undefined],
+      [1, undefined],
+    ],
+    'source gets correct number of requests from sink'
+  );
+
+  t.end();
+});
+
+test('it passes sink errors up (& data for unknown types too)', t => {
+  const source = makeMockCallbag(true);
+  const seedWithFoo = startWith('foo');
+  const sink = makeMockCallbag();
+
+  seedWithFoo(source)(0, sink);
+
+  sink.emit(1);
+  sink.emit('unknown', 'payload');
+  sink.emit(11, 'other_data');
+  sink.emit(2, 'err');
+
+  t.deepEqual(
+    source.getMessages().slice(1),
+    [
+      [1, undefined],
+      ['unknown', 'payload'],
+      [11, 'other_data'],
+      [2, 'err'],
+    ],
+    'source gets type & data from sink'
+  );
+
+  t.end();
+});
+
+test('it stops emitting after receiving unsubscription request', t => {
+  const sink = makeMockCallbag();
+
+  pipe(
+    fromIter(['d']),
+    startWith('a', 'b', 'c'),
+    take(2),
+    source => source(0, sink)
+  );
+
+  t.deepEqual(
+    sink.getReceivedData(),
+    ['a', 'b'],
+    'sink stops receiving data after unsubscribing'
+  );
+
+  t.end();
+});

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

Improved metrics

Total package byte prevSize

7082

increased by54.87%

Lines of code

181

increased by141.33%

Worsened metrics

Dev dependency count

7

increased by75%

No dependency changes