dominictarr / map-stream

MapStream

Refactored out of event-stream

##map (asyncFunction[, options])

Create a through stream from an asyncronous function.

var map = require('map-stream')

map(function (data, callback) {
  //transform data
  // ...
  callback(null, data)
})

Each map MUST call the callback. It may callback with data, with an error or with no arguments,

• callback() drop this data.
  this makes the map work like filter,
  note:callback(null,null) is not the same, and will emit null

• callback(null, newData) turn data into newData

• callback(error) emit an error for this item.

Note: if a callback is not called, map will think that it is still being processed,
every call must be answered or the stream will not know when to end.

Also, if the callback is called more than once, every call but the first will be ignored.

##Options

• failures - boolean continue mapping even if error occured. On error map-stream will emit failure event. (default: false)

map-stream's Issues

map-stream never ends when used inside of multipipe

I'm using a Gulp plugin called gulp-svg2png which uses map-stream internally.

When I use it with the multipipe, the stream end never happens (though gulp does exit). I have narrowed it down to map-stream not using the third callback argument of stream.write: https://github.com/dominictarr/map-stream/blob/master/index.js#L89. I'm not a Node stream expert, but it seems that if I accept the callback and call it when required, then everything works as expected.

streaming large files maxes out the call stack

Processing files with millions of records result in a crash on maximum call stack exceeded.

I believe returning queueData recursively just adds to the function stack. Deferring the recursive call to the next tick avoids this issue.

` function queueData (data, number) {
...
// If the next value is in the queue, write it
if (writeQueue.hasOwnProperty(nextToWrite)) {
var dataToWrite = writeQueue[nextToWrite]
delete writeQueue[nextToWrite]
process.nextTick(() => {
return queueData(dataToWrite, nextToWrite);
});
}
...

}`

Calling the mapper function with the context of the stream

Hi @dominictarr,

I am trying to understand whether it is an intended behavior in map-stream for not setting any context when calling the mapper function?

I recently noticed that a gulp plugin using map-stream used this.emit() within the mapper function, but obviously, the context of the stream is not available in the mapper function and thus this.emit() throw errors.

So, shouldn't https://github.com/dominictarr/map-stream/blob/master/index.js#L84 look something like the following?, unless I am missing something here.

function wrappedMapper (input, number, callback) {
+  return mapper.call(stream, input, function(err, data){
-   return mapper.call(null, input, function(err, data){
      callback(err, data, number)
    })
  }

Thanks!

end(data) doesn't seem to fire the 'data' event

I just started looking into your event-stream library and it looks pretty awesome... I ran into this though.

var eventstream = require('event-stream');

var theData;
var endEmitted = false;
var ws = eventstream.map(function(data, callback) {
  return callback(null, data);
});

ws.on('data', function(data) {
  theData = data;
});
ws.on('end', function(err) {
  endEmitted = true;
});

function check() {
  if (endEmitted) {
    console.log('end emitted, theData: %s', theData);
  } else {
    setTimeout(check, 1000);
  }
};

setTimeout(check, 1000);

ws.end('OH HAI');

outputs:

end emitted, theData: undefined

If I change it to use

ws.write('OH HAI');
ws.end();

we get

end emitted, theData: OH HAI

License is ambiguous

The license text is exactly that of The MIT License, but it is not explicitly stated, nor does it appear in the package.json file, which would allow the license to be determined programatically.

when "this" is undefined, this.opts breaks things

happens to me in the browser when i have a module with strict mode that uses map-stream

i don't see opts ever getting used, is it needed?

a simplified version of what happens (in chrome dev tools)

(function blah () { 'use strict'; console.log(this) }())
VM248:1 undefined

output stream does not preserve order of input stream

If the mapper function is asynchronous, the output stream is not guaranteed to have the same order as the input stream. This violates the semantics of "map".

Here is a test to illustrate the problem:

exports ['stream comes back in the correct order'] = function (test) {
  var input = [3, 2, 1]

  var delayer = map(function(data, cb){
    setTimeout(function () {
      cb(null, data)
    }, 100 * data)
  })

  readStream(delayer, function (err, output) {
    it(output).deepEqual(input)
    test.done()
  })

  writeArray(input, delayer)
}

And the resulting error:

test/
  stream comes back in the correct order
    AssertionError: [1,2,3] deepEqual [3,2,1]

Repository contents don't match published NPM version

Hi,

The published 0.1.0 version of map-stream does not have a license field in package.json. This would be very handy to have for automated license checking. I see in your repository that you already added the license field but somehow the version is 0.0.7.

Could you please publish a new version > 0.1.0 that has the license field?

Currently I have this package.json in my node_modules directory:

{
  "name": "map-stream",
  "version": "0.1.0",
  "description": "construct pipes of streams of events",
  "homepage": "http://github.com/dominictarr/map-stream",
  "repository": {
    "type": "git",
    "url": "git://github.com/dominictarr/map-stream.git"
  },
  "dependencies": {
  },
  "devDependencies": {
    "asynct": "*",
    "it-is": "1",
    "ubelt": "~2.9",
    "stream-spec": "~0.2",
    "event-stream": "~2.1",
    "from": "0.0.2"
  },
  "scripts": {
    "test": "asynct test/"
  },
  "author": "Dominic Tarr <[email protected]> (http://dominictarr.com)"
}

Can't Be Paused?

My understanding of node streams is that calling pause() should stop emitting events, though that doesn't seem to be the case with map-stream. Eg.:

var map = mapStream(function(f, cb) {
  cb( null, f );
});

map.on('data', function (x) {
  console.log(x);
});

// Pause the stream, should stop emitting data.
map.pause();
map.write(2);
// emits 2, but shouldn't it be paused?

I imagine in the vast majority of uses of map-stream that this would not be an issue. However, I am writing a library that interacts with Gulp plugins (which very often make use of map-stream) through the Highland stream library, which assumes the pausing functionality works as described above. See caolan/highland#243 for more info on that use case.

Is this a bug/missing feature in map-stream or is my understanding a little off? Assuming I'm right, I'd be happy to take a stab at a pr; I'm not a stream expert, but I have a bit of experience with js so with a small hint in the right direction I'm confident I could implement it.

What happens on 0.1.0?

Some modules depend on 0.1.0 (e.g. gulp-eslint). I cannot find it from the npm page but npm info includes the version.

{ name: 'map-stream',
  description: 'construct pipes of streams of events',
  'dist-tags': { latest: '0.0.5' },
  versions:
   [ '0.0.1',
     '0.0.2',
     '0.0.3',
     '0.0.4',
     '0.0.5',
     '0.1.0' ],
  maintainers: 'dominictarr <[email protected]>',
  time:
   { modified: '2014-11-03T17:03:20.755Z',
     created: '2012-08-18T13:07:31.578Z',
     '0.0.1': '2012-08-18T13:07:33.660Z',
     '0.0.2': '2013-01-17T04:54:50.995Z',
     '0.0.3': '2013-09-24T03:42:27.320Z',
     '0.0.4': '2013-12-07T02:05:06.804Z',
     '0.1.0': '2014-01-09T07:15:35.544Z',
     '0.0.5': '2014-11-03T17:03:20.755Z' },
  author: 'Dominic Tarr <[email protected]> (http://dominictarr.com)',
  repository:
   { type: 'git',
     url: 'git://github.com/dominictarr/map-stream.git' },
  users:
   { brianloveswords: true,
     dukewan: true,
     lapanoid: true },
  homepage: 'http://github.com/dominictarr/map-stream',
  bugs: { url: 'https://github.com/dominictarr/map-stream/issues' },
  license: 'MIT',
  readmeFilename: 'readme.markdown',
  version: '0.0.5',
  dependencies: {},
  devDependencies:
   { asynct: '*',
     'it-is': '1',
     ubelt: '~2.9',
     'stream-spec': '~0.2',
     'event-stream': '~2.1',
     from: '0.0.2' },
  scripts: { test: 'asynct test/' },
  gitHead: '336f8ecdce81f5c66001646871ca73584879b419',
  dist:
   { shasum: '80ffe0f588da32bf006b7f12d9197c5b8725df1c',
     tarball: 'http://registry.npmjs.org/map-stream/-/map-stream-0.0.5.tgz' },
  directories: {} }

Question

Hey Dominic -

This isn't really an issue but more of a question. I've used node for a long time but haven't gotten really heavy into piping streams until recently. Usually when I want to transform input into a different output I do this:

var Transform = require('stream').Transform;

var mapper = new Transform({
  objectMode: true,
  highWaterMark: 100
});

mapper._transform = function(data, enc, cb) {
  //do some mapping on the object
  //e.g. data.fullName = data.firstName + ' ' + data.lastName
  this.push(data)
  cb()
};

input.pipe(mapper).pipe(output)

I thought you'd be doing something similar in this module, but looking at the source it seems you implemented a stream from scratch. Just wondering why you did that? What disadvantages my approach has or what I'm missing?

Thanks for all your awesome contributions & for the talk you gave @ nodeconf 2013 too. 👍

how do I stop the next function from getting executed?

Say, if there is a condition where the whole .map() should just exit without calling the subsequent callback. Is it possible to achieve that?

how to map last null?

I need to know when the stream ends so i can append there my last wrapping up data. I was thinking it would be reasonable to see if the data passed to map function is null to indicate the end of stream, but it' apparently not. Would it be reasonable to be able to map over last null and allow two callbacks like below:

map(function(data, cb) {
  if(data === null) {
    cb(null, "MyVeryLastData");
    cb(null, null); //to notify that the stream has ended.
  } else {
    cb(null, data);
  }
});

This would save from some boilerplate in a nice way. What do you think? Or is there some other convenient ways to append the data within the context of map after the last chunk?

What`s the difference between map-stream and event-stream

What`s the difference between map-stream and event-stream?