end event from pipes doesn't forward $data

[markkimsal]

In Util::pipe() the default behavior is to connect the sources end event with the destination's end event, but it does not forward any potential $data.

        // forward end event from source as $dest->end()
        $end = isset($options['end']) ? $options['end'] : true;
        if ($end) { 
            $source->on('end', $ender = function () use ($dest) { 
                $dest->end();
            });
            $dest->on('close', function () use ($source, $ender) { 
                $source->removeListener('end', $ender);
            });
        } 

I've had to wrap Util::pipe() with my own function that basically mixes this behavior with the forwardEvents util call

    public function pipe(WritableStreamInterface $dest, array $options = array())
    { 
        $options['end'] = FALSE;

        //default pipe behavior throws away $data for end events
        $this->on('end', $ender = function () use($dest) { 
            call_user_func_array([$dest, 'end'], func_get_args());
        });

        $source = $this;
        $dest->on('close', function () use ($source, $ender) { 
            $source->removeListener('end', $ender);
        });

        return Util::pipe($this, $dest, $options);
    } 

Should end() not be used this way? I'm writing a pipe that will collect data from streams and concat into one file, when this "destination" pipe gets end() from its source it will concat all the info into one file and emit its own end with the name of the file as the only value of $data (wrapped in an array). I could rewrite it to emit 'data' one time, but it seems more natural for a pipe component that waits for upstream pipes to send end that it would also send end.

Editing the Util.php file also gives me the result I was expecting:

        // forward end event from source as $dest->end()
        $end = isset($options['end']) ? $options['end'] : true;
        if ($end) { 
            $source->on('end', $ender = function ($data=null) use ($dest) {   //change 1 of 2
                $dest->end($data);                                       //change 2 of 2
            });
            $dest->on('close', function () use ($source, $ender) { 
                $source->removeListener('end', $ender);
            });
        } 

[kelunik]

What's the purpose of the filename as data sent to the 'end' event? How is it used?

[markkimsal]

Sorry, it doesn't actually send the filename as the end, it sends the contents of the concated files, but it waits until the source pipeline sends its end().

$p->src( ['src/**/*.php', 'templates/**/*.html'])
  ->pipe( new ZipPipe() )
  ->pipe(  $p->dest("compressed.zip") );

Where $p->src and $p->dest create readable and writable streams and ZipPipe listens for the end from src, then sends its own end( $allFilesContents ) and then $p->dest() is passed the final filename and contents are read via pipes.

EDIT non-pseudo code:

$p->task('less', function() use($p) {
    $lr = new \Pulp\LiveReload();
    $lr->listen($p->loop);

    $p->watch( ['foo/**/*.less'])->on('change', function($file) use ($p, $lr) {
        $p->output("File changed '".$file->getFilename()."'");
        $p->src(['foo/bootstrap.less'])
          ->pipe(new \Pulp\Less( ['compress'=>TRUE] ))
          ->pipe($p->dest('foo/compressed.css'))
          ->pipe($lr)
        ;
    });
}); 

[clue]

Thank you for this very good question!

I realize that it's easy to confuse the end() method with the end event and its semantics. The former ends the writable side of a stream while the latter reports that the readable side of a stream has been closed.

The end() method accepts a parameter (final write) purely for convenience. Calling end('bye') does exactly the same thing as write('bye');end();.

This means that the end event does never transport any data at all. Accordingly, both variants should emit a final data event with bye and then an end event with no data attached.

I hope this helps 👍

I believe this has been answered, so I'm closing this for now. Please come back with more details if this problem persists and/or you have an idea on how we can improve our documentation and we can reopen this 👍