flatmap issue?

[yelouafi]

Has anyone tried the flatmap module? i'm giving errors when playing with some examples

javascript var flyd = require("flyd"); var flatmap = require("flyd-flatmap")

var main = flyd.stream()

flatmap(function(v) { return flyd.stream(v); }, main)

main(10)

it gives me the following error

C:\Users\Yassine\Desktop\test (2)\streams\node_modules\flyd\flyd.js:0 (function (exports, require, module, filename, dirname) { (function (root, ^ RangeError: Maximum call stack size exceeded

I hope i'm not doing something dumb? :)

[roobie]

I've also blown the stack with flatmap.

[yelouafi]

upon debugging i've noticed this circular call path, maybe this will help

flatmap(...)
  -> map(...)
        ->stream(...)
              ->flushUpdate(...)
                      -> updateDeps(...)
                             updateStream(...) {
                                  ....
                                  var returnVal = s.fn(s, s.depsChanged);
                             }
                               -> and back to flyd.map inside flatmap

[paldepind]

@yelouafi @raine Can you still reproduce this with the most recent version (0.1.10)? I've just closed #26 and I'm fairly certain this is the same bug.

[yelouafi]

@paldepind tested the above example with v0.1.10 and worked but now switched to an async example and i get an unexpected behavior

to start here is the complete example

var flyd = require("flyd");
var flatmap = require("flyd-flatmap")

function debug(s, prefix) {
  flyd.map(console.log.bind(console, prefix), s)
}

function delay(val, ms) {
  return flyd.stream([], function(self) {
      setTimeout(function() {
        self(val);
        self.end(true);
      }, ms)
  });
}

var main = delay(1, 1000)
var merged = flatmap(function(v) {
    return delay(v, 1000)
}, main);

debug(main, 'main')
debug(merged, 'merged')

This was to ensure that the flat-mapped stream continues yielding from the dynamically generated streams even if the main stream ends. Normally the merged stream shouldn't end with the main stream but it has to wait for all nested streams to complete. I suspected this because in the source of flatmap there is no custom endsOn so i wanted to check if the result stream will ends by default with its only dependency (the main/monadic stream)

but even when i delay further the main stream ending

function delay(val, ms) {
  return flyd.stream([], function(self) {
      setTimeout(function() {
        self(val);
        setTimeout(self.end.bind(null, true), 10000) // ends after 10 seconds
      }, ms)
  });
}

i can't still get any output from the merged stream, even if the main stream will not end until after 10 seconds.

Now if i remove the timeout from the nested stream i get the desired output

var merged = flatmap(function(v) {
    return flyd.stream(v)
}, main);

I hope i'm not missing something here.

[iofjuupasli]

If you will move extensions to one repository, it'll be easier to test and maintain. Now it's probably really hard to check all extensions on some changes in core. The more that API is not stable yet (probably you will add laziness, backpressure, etc.) and more extensions are created. Notice also that this issue created here, not in flyd-flatmap repository. It makes no sense to keep them separated as extensions depends on core and can't be reused by other libraries without core. From user point of view there will berequire('flyd/flatmap') instead require('flyd-flatmap'). And in package.json only "flyd": "x.x.x" instead "flyd": "x.x.x", "flyd-flatmap": "x.x.x", "flyd-filter": "x.x.x", "flyd-foo": "x.x.x", "flyd-bar": "x.x.x", "flyd-more-extensions": "x.x.x", "etcetcetc": "x.x.x"

[roobie]

With v. 0.1.10 I do not get the stack overflow. This is my code:

let flatMap = function(f, s) {
  return flyd.stream([s], function(own) {
    flyd.map(own, f(s()));
  });
};

let getRandomCharStream = () => {
  let timer = every(50);
  let end = timer.end;
  let result = flyd.stream([timer], function (rc$) {
    if (Math.random() < 0.05) {
      end(true);
    }

    rc$(String.fromCharCode(Math.random() * 0xff | 0));
  });

  return result;
};

module.exports = function samaritan(input$) {
  return flyd.scan(function (acc, n) {
    return acc.concat(n);
  }, [], flatMap(function () {
    return getRandomCharStream();
  }, input$));
};

[paldepind]

@iofjuupasli I agree. Will you open a separate issue, thank you?

[yelouafi]

@roobie My point was that the stream returned by samaritan function should follow all the nested streams (generated from the return getRandomCharStream() ) until they all end and not to stop as soon as the main stream (input$) ends.

[yelouafi]

Ok, i was afraid of being doing something dumb, and i was right. Here is the guilty

function delay(val, ms) {
  return flyd.stream([], function(self) {
      setTimeout(function() {
        self(val);
        self.end(true);
      }, ms)
  });
}

the setTimeout never gets called (i guess because the stream has [] dependencies which is curious given it gets invoked if delay is called on a top stream). This explains why the merged stream from my example doesn't yield anything even if i delay enough the end of the main stream.

Remains the problem of the premature end of the flatmapped stream.

[roobie]

@yelouafi this works for me:

flyd.map(function (v) {
  console.log(v)
}, flyd.stream([], function (s) {
  setTimeout(function () {
    s(1);
    s.end(true);
  },100);
}));

Am I missing something?

edit: updated code.

All right, I tested with the flatmap and it's not invoked

[c-dante]

I made this a test case on my fork, the test currently fails: https://github.com/c-dante/flyd/commit/f84985bc3a03d19e623e78ce3b330499d66364bd

So I think for flat map it should follow something like:

var capturedStream = flyd.stream();
var sampleStream = flyd.stream();

var results = sampleStream.flatMap(() => capturedStream)
capturedStream(2);
sampleStream(0);
capturedStream(5);
sampleStream.end();
capturedStream(1).end(true);

// Expectation. X = end
sample:  --0-----X
capture: -2----5---1---X
results:   2---5---1---X

So, flatMap ends when the mapped stream AND all produced streams have ended.