end called while paused for Readable Streams

[joewagner]

@dominictarr I was expecting that the end method would wait if the last write method is paused. I added a test to demonstrate the issue I was having. Please let me know if this doesn't make sense.

Thanks for your time,

[dominictarr]

hmm... this module has been unchanged for a long time, and has 2000 dependent modules (not counting their dependent modules, etc) I'm somewhat reluctant to change anything lest I break one of those things.

What use case do you have that needs end to be paused?

[joewagner]

@dominictarr Not wanting to add this is very understandable, and thanks for your work on the module. My use case is building a reporting tool that pipes a mongodb cursor into a function that does a separate database lookup. I was hoping the end function would be called after the last lookup is done, but it is called after the second to last lookup. The test included in the PR demostrates the issue, but below is a real life snippet.

const usersCursor = Users.find({}).cursor() // go though the entire collection

usersCursor.pipe(through(function (user) {
    this.pause()

    Users.find({$or: [
        // Find users with similar username or email
        {nam: new RegExp(similarUsernameRegexp(user.nam), 'i')},
        {email: new RegExp(similarEmailRegexp(user.email), 'i')}
    ]}, (err, users) => {
        if (err) {
            console.error(err)
            this.emit('error', err)
            return this.resume()
        }

        if (users.length > 1) {
            // Add users who have similarities with at least one other
            const row = {
                USERNAME: user.nam,
                EMAIL: user.email,
                'FIRST NAME': user.firstName,
                'LAST NAME': user.lastName
            }

            this.queue(row)
        }

        this.resume()
    })

}, function (err) {
    if (err) {
        console.log(err)
    }

    this.queue(null)

    // Was hoping to be done with all of the User lookups at this point,
    // also of note the next stage in the pipeline never receives the last user.
}))
// build a document from the results
.pipe(csvBuilder.createTransformStream())