Add support for closing STDIN

Author: Jan Henning Thorsen

Changes

@@ -5,6 +5,7 @@ Revision history for Mojo-IOLoop-ReadWriteFork
        * Add support for write() before child process has started
        * Add support for "drain" callback to write()
        * Add run(). a simpler version start()
+       * Add support for closing STDIN
 
 0.05     Wed Feb 19 13:29:54 2014
        * Fix "read" event cannot change ERRNO from sysread()

lib/Mojo/IOLoop/ReadWriteFork.pm

@@ -100,6 +100,34 @@ has reactor => sub {
 
 =head1 METHODS
 
+=head2 close
+
+  $self = $self->close("stdin");
+
+Close STDIN stream to the child process immediately.
+
+=cut
+
+sub close {
+  my ($self, $what) = @_;
+  $self->{stream}{$what}->close if ref $self->{stream}{$what};
+  $self;
+}
+
+=head2 close_gracefully
+
+  $self = $self->close_gracefully("stdin");
+
+Close STDIN to the child process stream gracefully.
+
+=cut
+
+sub close_gracefully {
+  my ($self, $what) = @_;
+  $self->{stream}{$what}->close_gracefully if ref $self->{stream}{$what};
+  $self;
+}
+
 =head2 run
 
   $self = $self->run($program, @program_args);
@@ -288,8 +316,8 @@ Example:
 sub write {
   my $self = shift;
 
-  if ($self->{stdin}) {
-    $self->{stdin}->write(@_);
+  if ($self->{stream}{stdin}) {
+    $self->{stream}{stdin}->write(@_);
     warn "[${ \$self->pid }] Wrote buffer (" .url_escape($_[0]) .")\n" if DEBUG;
   }
   else {
@@ -326,7 +354,7 @@ sub _cleanup {
   $reactor->remove(delete $self->{stdout_read}) if $self->{stdout_read};
   $reactor->remove(delete $self->{delay}) if $self->{delay};
   $reactor->remove(delete $self->{delay}) if $self->{delay};
-  $self->{stdin}->close if $self->{stdin};
+  delete($self->{stream}{stdin})->close if $self->{stream}{stdin};
 }
 
 sub _read {
@@ -351,7 +379,7 @@ sub _stdin {
   $stream->timeout(0);
   $stream->start;
 
-  $self->{stdin} = $stream;
+  $self->{stream}{stdin} = $stream;
   $self->write(@$_) for @{ delete $self->{stdin_buffer} || [] };
 }
 

t/close.t

@@ -0,0 +1,36 @@
+use Mojo::Base -strict;
+use Mojo::IOLoop::ReadWriteFork;
+use Test::More;
+use Test::Memory::Cycle;
+
+{
+  my $run = Mojo::IOLoop::ReadWriteFork->new;
+  my $output = '';
+
+  $run->on(close => sub { Mojo::IOLoop->stop; });
+  $run->on(read => sub { $output .= $_[1]; });
+  $run->write("line one\nline two\n", sub { shift->close('stdin'); });
+  $run->run(sub { print while <>; print "FORCE\n"; });
+
+  Mojo::IOLoop->timer(3 => sub { Mojo::IOLoop->stop }); # guard
+  Mojo::IOLoop->start;
+
+  like $output, qr/line one\nline two\nFORCE\n/, 'close' or diag $output;
+}
+
+{
+  my $run = Mojo::IOLoop::ReadWriteFork->new;
+  my $output = '';
+
+  $run->on(close => sub { Mojo::IOLoop->stop; });
+  $run->on(read => sub { $output .= $_[1]; shift->close_gracefully('stdin'); });
+  $run->write("line one\nline two\n");
+  $run->run(sub { print while <>; print "GRACEFUL\n" });
+
+  Mojo::IOLoop->timer(3 => sub { Mojo::IOLoop->stop }); # guard
+  Mojo::IOLoop->start;
+
+  like $output, qr/line one\nline two\nGRACEFUL\n/, 'close_gracefully' or diag $output;
+}
+
+done_testing;