Skip to content

HTTPS clone URL

Subversion checkout URL

You can clone with HTTPS or Subversion.

Download ZIP
Browse files

Parameterize events the roles can emit. Create eg-15-ipc-run.pl and a…

… supporting role and class.
  • Loading branch information...
commit 2ea8fb890f7eec4bdfe31124e059bab4aebd1dcf 1 parent 5b2d561
Rocco Caputo authored
16 eg/Runner.pm
View
@@ -0,0 +1,16 @@
+package Runner;
+use Moose;
+extends 'Reflex::Base';
+
+has [qw(stdin stdout stderr)] => ( isa => 'FileHandle', is => 'ro' );
+has pid => ( isa => 'Int', is => 'ro' );
+
+with 'RunnerRole' => {
+ stdin => 'stdin',
+ stdout => 'stdout',
+ stderr => 'stderr',
+ pid => 'pid',
+ ev_exit => 'exit',
+};
+
+1;
226 eg/RunnerRole.pm
View
@@ -0,0 +1,226 @@
+package RunnerRole;
+use Reflex::Role;
+
+attribute_parameter stdin => "stdin";
+attribute_parameter stdout => "stdin";
+attribute_parameter stderr => "stdin";
+attribute_parameter pid => "pid";
+
+callback_parameter cb_stdout_data => qw( on stdout data );
+callback_parameter cb_stdout_error => qw( on stdout error );
+
+callback_parameter cb_stdout_closed => qw( on stdout closed );
+event_parameter ev_stdout_closed => qw( _ stdout closed );
+
+callback_parameter cb_stderr_data => qw( on stderr data );
+callback_parameter cb_stderr_error => qw( on stderr error );
+
+callback_parameter cb_stderr_closed => qw( on stderr closed );
+event_parameter ev_stderr_closed => qw( _ stderr closed );
+
+callback_parameter cb_exit => qw( on pid exit );
+event_parameter ev_exit => qw( _ pid exit );
+
+method_parameter method_put => qw( put stdin _ );
+
+role {
+ my $p = shift;
+
+ with 'Reflex::Role::OutStreaming' => {
+ handle => $p->stdin(),
+ method_put => $p->method_put(),
+ };
+
+ my $m_stdout_stop = "stop_" . $p->stdout();
+ my $cb_stdout_closed = $p->cb_stdout_closed();
+ my $ev_stdout_closed = $p->ev_stdout_closed();
+
+ after $cb_stdout_closed => sub {
+ my ($self, $args) = @_;
+ $self->$m_stdout_stop();
+ };
+
+ with 'Reflex::Role::InStreaming' => {
+ handle => $p->stdout(),
+ cb_data => $p->cb_stdout_data(),
+ cb_error => $p->cb_stdout_error(),
+ cb_closed => $cb_stdout_closed,
+ };
+
+ my $m_stderr_stop = "stop_" . $p->stderr();
+ my $cb_stderr_closed = $p->cb_stderr_closed();
+ my $ev_stderr_closed = $p->ev_stderr_closed();
+
+ after $cb_stderr_closed => sub {
+ my ($self, $args) = @_;
+ $self->$m_stderr_stop();
+ $self->emit(event => $ev_stderr_closed, args => $args);
+ };
+
+ with 'Reflex::Role::InStreaming' => {
+ handle => $p->stderr(),
+ cb_data => $p->cb_stderr_data(),
+ cb_error => $p->cb_stderr_error(),
+ cb_closed => $cb_stderr_closed,
+ };
+
+ with 'Reflex::Role::PidCatcher' => {
+ pid => 'pid',
+ cb_exit => $p->cb_exit(),
+ ev_exit => $p->ev_exit(),
+ };
+};
+
+1;
+
+__END__
+
+extends 'Reflex::Base';
+
+use Reflex::Trait::Observed;
+use Reflex::PID;
+
+use Carp qw(croak);
+use IPC::Run qw(start);
+use Symbol qw(gensym);
+
+
+__END__
+
+observes process => ( isa => 'Maybe[Reflex::PID]', is => 'rw' );
+
+has [qw(stdin stdout stderr)] => (
+ isa => 'Maybe[FileHandle]',
+ is => 'rw',
+);
+
+has ipc_run => ( isa => 'IPC::Run', is => 'rw' );
+
+has cmd => (
+ isa => 'ArrayRef',
+ is => 'ro',
+ required => 1,
+);
+
+### Reap the child process.
+
+sub on_process_exit {
+ my ($self, $args) = @_;
+ $self->emit(event => 'exit', args => $args);
+}
+
+sub kill {
+ my ($self, $signal) = @_;
+ croak "no process to kill" unless $self->process();
+ $signal ||= 'TERM';
+ kill $signal, $self->process()->pid();
+}
+
+### Write to standard input.
+
+sub on_stdin_error {
+ my ($self, $args) = @_;
+ $self->emit(event => 'stdin_error', args => $args);
+}
+
+with 'Reflex::Role::Writing' => { handle => 'stdin' };
+
+sub on_stdin_writable {
+ my ($self, $arg) = @_;
+ my $octets_left = $self->flush_stdin();
+ return if $octets_left;
+ $self->flush_stdin();
+}
+
+with 'Reflex::Role::Writable' => { handle => 'stdin' };
+
+### Read from standard output.
+
+sub on_stdout_readable {
+ my ($self, $arg) = @_;
+ my $octets_read = $self->read_stdout($arg);
+ warn $octets_read;
+ return if $octets_read;
+ if (defined $octets_read) {
+ warn 111;
+ $self->pause_stdout_readable();
+ return;
+ }
+ $self->stop_stdout_readable();
+}
+
+sub on_stdout_error {
+ my ($self, $args) = @_;
+ $self->emit(event => 'stdout_error', args => $args);
+ $self->stop_stdout_readable();
+}
+
+with 'Reflex::Role::Reading' => {
+ handle => 'stdout',
+ cb_data => 'on_stdout',
+ ev_data => 'stdout',
+};
+
+with 'Reflex::Role::Readable' => {
+ handle => 'stdout',
+ cb_ready => 'on_stdout_readable',
+};
+
+### Read from standard error.
+
+sub on_stderr_error {
+ my ($self, $args) = @_;
+ $self->emit(event => 'stderr_error', args => $args);
+ $self->stop_stderr_readable();
+}
+
+sub on_stderr_readable {
+ my ($self, $arg) = @_;
+ my $octets_read = $self->read_stderr($arg);
+ warn $octets_read;
+ return if $octets_read;
+ if (defined $octets_read) {
+ warn 111;
+ $self->pause_stderr_readable();
+ return;
+ }
+ $self->stop_stderr_readable();
+}
+
+with 'Reflex::Role::Reading' => {
+ handle => 'stderr',
+ cb_data => 'on_stderr',
+ ev_data => 'stderr',
+};
+
+with 'Reflex::Role::Readable' => {
+ handle => 'stderr',
+ cb_ready => 'on_stderr_readable',
+};
+
+sub BUILD {
+ my $self = shift;
+
+ my ($fh_in, $fh_out, $fh_err) = (gensym(), gensym(), gensym());
+
+ $self->ipc_run(
+ start(
+ $self->cmd(),
+ '<pipe', $fh_in,
+ '>pipe', $fh_out,
+ '2>pipe', $fh_err,
+ )
+ ) or die "IPC::Run start() failed: $? ($!)";
+
+ $self->process(
+ Reflex::PID->new(
+ pid => $self->ipc_run->{KIDS}[0]{PID}
+ )
+ );
+
+ $self->stdin($fh_in);
+ $self->stdout($fh_out);
+ $self->stderr($fh_err);
+}
+
+1;
83 eg/eg-15-ipc-run.pl
View
@@ -0,0 +1,83 @@
+#!/usr/bin/perl
+
+# Demonstrate subprocesses without POE::Wheel::Run.
+# Test case for upcoming Reflex::Run, which drives IPC::Run.
+
+use warnings;
+use strict;
+use lib qw(../lib);
+
+use Runner;
+
+my $cmd = [
+ $^X, '-MTime::HiRes=sleep', '-wle',
+ q($|=1;) .
+ q(for (1..3) { $_ = qq[pid($$) moo($_)]; print; warn "$_\n"; sleep rand; })
+];
+
+sub my_start {
+ my $cmd = shift;
+
+ use IPC::Run qw(start);
+ use Symbol qw(gensym);
+
+ my ($fh_in, $fh_out, $fh_err) = (gensym(), gensym(), gensym());
+
+ my $ipc_run = start(
+ $cmd,
+ '<pipe', $fh_in,
+ '>pipe', $fh_out,
+ '2>pipe', $fh_err,
+ ) or die "IPC::Run start() failed: $? ($!)";
+
+ return($ipc_run, $fh_in, $fh_out, $fh_err);
+}
+
+my ($ipc_run_1, $runner_1);
+{
+ ($ipc_run_1, my($fh_in, $fh_out, $fh_err)) = my_start($cmd);
+
+ $runner_1 = Runner->new(
+ stdin => $fh_in,
+ stdout => $fh_out,
+ stderr => $fh_err,
+ pid => $ipc_run_1->{KIDS}[0]{PID},
+
+ on_stdout_closed => sub { print "runner_1 stdout closed\n" },
+ on_stderr_closed => sub { print "runner_1 stderr closed\n" },
+ on_stdout_data => sub { print "runner_1 stdout: $_[1]{data}" },
+ on_stderr_data => sub { print "runner_1 stderr: $_[1]{data}" },
+
+ on_exit => sub {
+ my ($self, $args) = @_;
+ warn "runner_1 child $args->{pid} exited: $args->{exit}\n";
+ $runner_1 = undef;
+ },
+ );
+}
+
+my ($ipc_run_2, $runner_2);
+{
+ ($ipc_run_2, my($fh_in, $fh_out, $fh_err)) = my_start($cmd);
+
+ $runner_2 = Runner->new(
+ stdin => $fh_in,
+ stdout => $fh_out,
+ stderr => $fh_err,
+ pid => $ipc_run_2->{KIDS}[0]{PID},
+
+ on_stdout_closed => sub { print "runner_2 stdout closed\n" },
+ on_stderr_closed => sub { print "runner_2 stderr closed\n" },
+ on_stdout_data => sub { print "runner_2 stdout: $_[1]{data}" },
+ on_stderr_data => sub { print "runner_2 stderr: $_[1]{data}" },
+
+ on_exit => sub {
+ my ($self, $args) = @_;
+ warn "runner_2 child $args->{pid} exited: $args->{exit}\n";
+ $runner_2 = undef;
+ },
+ );
+}
+
+Reflex->run_all();
+exit;
1  lib/Reflex/Interval.pm
View
@@ -12,6 +12,7 @@ with 'Reflex::Role::Interval' => {
auto_start => "auto_start",
auto_repeat => "auto_repeat",
cb_tick => "on_tick",
+ ev_tick => "tick",
method_start => "start",
method_stop => "stop",
method_repeat => "repeat",
1  lib/Reflex/PID.pm
View
@@ -19,6 +19,7 @@ with 'Reflex::Role::PidCatcher' => {
pid => 'pid',
active => 'active',
cb_exit => 'on_exit',
+ ev_exit => 'exit',
method_start => 'start',
method_stop => 'stop',
method_pause => 'pause',
2  lib/Reflex/POE/Wheel/Run.pm
View
@@ -98,7 +98,7 @@ sub BUILD {
}
# Rethrow our signal event.
-sub on_sigchld_pid {
+sub on_sigchld_exit {
my ($self, $args) = @_;
$self->emit(
event => 'signal',
8 lib/Reflex/Role.pm
View
@@ -10,7 +10,7 @@ use Moose::Exporter;
Moose::Exporter->setup_import_methods(
with_caller => [ qw(
- attribute_parameter method_parameter callback_parameter
+ attribute_parameter method_parameter callback_parameter event_parameter
method_emit_and_stop method_emit
) ],
also => 'MooseX::Role::Parameterized',
@@ -115,6 +115,8 @@ sub method_emit {
# parameter flags to automatically generate those methods.
BEGIN { *callback_parameter = *method_parameter; }
+BEGIN { *event_parameter = *method_parameter; }
+
1;
__END__
@@ -125,6 +127,8 @@ Reflex::Role - define a Reflex paramaterized role
=head1 SYNOPSIS
+TODO - Changed again;
+
package Reflex::Role::Streaming;
use Reflex::Role;
@@ -145,7 +149,7 @@ Reflex::Role - define a Reflex paramaterized role
with 'Reflex::Role::Collectible';
- method_emit_and_stop $cb_error => "error";
+ method_emit_and_stop $cb_error => $p->ev_error();
with 'Reflex::Role::Reading' => {
handle => $h,
8 lib/Reflex/Role/Accepting.pm
View
@@ -4,7 +4,9 @@ use Reflex::Role;
attribute_parameter listener => "listener";
callback_parameter cb_accept => qw( on listener accept );
-callback_parameter cb_error => qw( on listener error );
+event_parameter ev_accept => qw( _ listener accept );
+callback_parameter cb_error => qw( on listener error );
+event_parameter ev_error => qw( _ listener error );
method_parameter method_pause => qw( pause listener _ );
method_parameter method_resume => qw( resume listener _ );
method_parameter method_stop => qw( stop listener _ );
@@ -44,8 +46,8 @@ role {
return;
};
- method_emit $cb_accept => "accept";
- method_emit $cb_error => "error"; # TODO - Retryable ones.
+ method_emit $cb_accept => $p->ev_accept();
+ method_emit $cb_error => $p->ev_error(); # TODO - Retryable ones.
with 'Reflex::Role::Readable' => {
handle => $listener,
7 lib/Reflex/Role/Connecting.pm
View
@@ -11,6 +11,9 @@ attribute_parameter port => "port";
callback_parameter cb_success => qw( on socket success );
callback_parameter cb_error => qw( on socket error );
+event_parameter ev_success => qw( _ socket success );
+event_parameter ev_error => qw( _ socket error );
+
role {
my $p = shift;
@@ -103,8 +106,8 @@ role {
return;
};
- method_emit $cb_success => "success";
- method_emit $cb_error => "error";
+ method_emit $cb_success => $p->ev_success();
+ method_emit $cb_error => $p->ev_error();
with 'Reflex::Role::Writable' => {
handle => $socket,
5 lib/Reflex/Role/Interval.pm
View
@@ -9,7 +9,9 @@ attribute_parameter auto_start => "auto_start";
method_parameter method_start => qw( start name _ );
method_parameter method_stop => qw( stop interval _ );
method_parameter method_repeat => qw( repeat interval _ );
+
callback_parameter cb_tick => qw( on interval tick );
+event_parameter ev_tick => qw( _ interval tick );
role {
my $p = shift;
@@ -76,9 +78,10 @@ role {
$self->$timer_id_name(undef);
};
+ my $ev_tick = $p->ev_tick();
method $cb_tick => sub {
my ($self, $args) = @_;
- $self->emit(event => "tick", args => $args);
+ $self->emit(event => $ev_tick, args => $args);
$self->$method_repeat() if $self->$auto_repeat();
};
};
4 lib/Reflex/Role/PidCatcher.pm
View
@@ -11,6 +11,8 @@ parameter active => (
);
callback_parameter cb_exit => qw( on pid exit );
+event_parameter ev_exit => qw( _ pid exit );
+
method_parameter method_start => qw( start pid _ );
method_parameter method_stop => qw( stop pid _ );
method_parameter method_pause => qw( pause pid _ );
@@ -143,7 +145,7 @@ role {
}
};
- method_emit $cb_exit => "pid";
+ method_emit $cb_exit => $p->ev_exit();
};
__END__
1  lib/Reflex/Role/Reactive.pm
View
@@ -415,6 +415,7 @@ sub emit {
}
$deliver_event = "promise";
+ #warn $event unless exists $self->watchers_by_event()->{$deliver_event};
return unless exists $self->watchers_by_event()->{$deliver_event};
# Fall through.
}
12 lib/Reflex/Role/Readable.pm
View
@@ -47,18 +47,24 @@ role {
$POE::Kernel::poe_kernel->select_pause_read($self->$h());
};
- method $p->method_pause() => sub {
+ my $method_pause = $p->method_pause();
+ method $method_pause => sub {
my $self = shift;
+ return unless $self->call_gate($method_pause);
$POE::Kernel::poe_kernel->select_pause_read($self->$h());
};
- method $p->method_resume() => sub {
+ my $method_resume = $p->method_resume();
+ method $p->method_resume => sub {
my $self = shift;
+ return unless $self->call_gate($method_resume);
$POE::Kernel::poe_kernel->select_resume_read($self->$h());
};
- method $p->method_stop() => sub {
+ my $method_stop = $p->method_stop();
+ method $method_stop => sub {
my $self = shift;
+ return unless $self->call_gate($method_stop);
$POE::Kernel::poe_kernel->select_read($self->$h(), undef);
};
24 lib/Reflex/Role/Reading.pm
View
@@ -7,6 +7,9 @@ callback_parameter cb_data => qw( on handle data );
callback_parameter cb_error => qw( on handle error );
callback_parameter cb_closed => qw( on handle closed );
+event_parameter ev_data => qw( _ handle data );
+event_parameter ev_closed => qw( _ handle closed );
+
# Matches Reflex::Role::Readable's default callback.
# TODO - Any way we can coordinate this so it's obvious in the code
# but not too verbose?
@@ -31,13 +34,13 @@ role {
# Got data.
if ($octet_count) {
$self->$cb_data({ data => $buffer });
- return;
+ return $octet_count;
}
# EOF
if (defined $octet_count) {
$self->$cb_closed({ });
- return;
+ return $octet_count;
}
# Quelle erreur!
@@ -48,11 +51,20 @@ role {
errfun => "sysread",
}
);
+ return; # Nothing.
};
# Default callbacks that re-emit their parameters.
- method_emit $cb_data => "data";
- method_emit_and_stop $cb_closed => "closed";
+
+ method_emit $cb_data => $p->ev_data();
+ method_emit_and_stop $cb_closed => $p->ev_closed();
+
+# my $ev_closed = $p->ev_closed();
+# method $cb_closed => sub {
+# my ($self, $args) = @_;
+# $self->emit(event => $ev_closed, args => $args);
+# $self->stopped();
+# };
};
1;
@@ -65,6 +77,8 @@ Reflex::Role::Reading - add standard sysread() behavior to a class
=head1 SYNOPSIS
+TODO - Changed again.
+
package InputStreaming;
use Reflex::Role;
@@ -81,7 +95,7 @@ Reflex::Role::Reading - add standard sysread() behavior to a class
my $cb_error = $p->cb_error();
my $method_read = "on_${h}_readable";
- method_emit_and_stop $cb_error => "error";
+ method_emit_and_stop $cb_error => $p->ev_error();
with 'Reflex::Role::Reading' => {
handle => $h,
7 lib/Reflex/Role/Recving.pm
View
@@ -6,6 +6,9 @@ attribute_parameter handle => "socket";
callback_parameter cb_datagram => qw( on handle datagram );
callback_parameter cb_error => qw( on handle error );
+callback_parameter ev_datagram => qw( _ handle datagram );
+callback_parameter ev_error => qw( _ handle error );
+
method_parameter method_send => qw( send handle _ );
method_parameter method_stop => qw( stop handle _ );
@@ -89,8 +92,8 @@ role {
};
# Default callbacks that re-emit their parameters.
- method_emit $cb_datagram => "datagram";
- method_emit_and_stop $cb_error => "error";
+ method_emit $cb_datagram => $p->ev_datagram();
+ method_emit_and_stop $cb_error => $p->ev_error();
};
1;
3  lib/Reflex/Role/SigCatcher.pm
View
@@ -11,6 +11,7 @@ parameter active => (
);
callback_parameter cb_signal => qw( on signal caught );
+event_parameter ev_signal => qw( _ signal caught );
method_parameter method_start => qw( start signal _ );
method_parameter method_stop => qw( stop signal _ );
method_parameter method_pause => qw( pause signal _ );
@@ -152,7 +153,7 @@ role {
}
};
- method_emit $cb_signal => "signal";
+ method_emit $cb_signal => $p->ev_signal();
};
__END__
11 lib/Reflex/Role/Streaming.pm
View
@@ -8,11 +8,16 @@ callback_parameter cb_closed => qw( on handle closed );
method_parameter method_put => qw( put handle _ );
method_parameter method_stop => qw( stop handle _ );
+event_parameter ev_data => qw( _ handle data );
+event_parameter ev_error => qw( _ handle error );
+event_parameter ev_closed => qw( _ handle closed );
+
role {
my $p = shift;
my $h = $p->handle();
my $cb_error = $p->cb_error();
+ my $ev_error = $p->ev_error();
my $method_read = "_on_${h}_readable";
my $method_put = $p->method_put();
@@ -24,13 +29,16 @@ role {
with 'Reflex::Role::Collectible';
- method_emit_and_stop $cb_error => "error";
+ method_emit_and_stop $cb_error => $ev_error;
with 'Reflex::Role::Reading' => {
handle => $h,
cb_data => $p->cb_data(),
+ ev_data => $p->ev_data(),
cb_error => $cb_error,
+ ev_error => $ev_error,
cb_closed => $p->cb_closed(),
+ ev_closed => $p->ev_closed(),
method_read => $method_read,
};
@@ -43,6 +51,7 @@ role {
with 'Reflex::Role::Writing' => {
handle => $h,
cb_error => $cb_error,
+ ev_error => $ev_error,
method_put => $internal_put,
};
3  lib/Reflex/Role/Timeout.pm
View
@@ -10,6 +10,7 @@ method_parameter method_start => qw( start delay _ );
method_parameter method_reset => qw( reset delay _ );
callback_parameter cb_timeout => qw( on delay done );
+event_parameter ev_timeout => qw( _ delay done );
role {
my $p = shift;
@@ -82,7 +83,7 @@ role {
$self->$timer_id_name(undef);
};
- method_emit $cb_timeout => "done";
+ method_emit $cb_timeout => $p->ev_timeout();
};
1;
2  lib/Reflex/Role/Writable.pm
View
@@ -79,7 +79,7 @@ role {
# Turn off watcher during destruction.
after DEMOLISH => sub {
my $self = shift;
- $self->method_stop();
+ $self->$method_stop();
};
};
1  lib/Reflex/Signal.pm
View
@@ -19,6 +19,7 @@ with 'Reflex::Role::SigCatcher' => {
signal => 'signal',
active => 'active',
cb_signal => 'on_signal',
+ ev_signal => 'signal',
method_start => 'start',
method_stop => 'stop',
method_pause => 'pause',
3  lib/Reflex/Stream.pm
View
@@ -16,6 +16,9 @@ with 'Reflex::Role::Streaming' => {
cb_error => 'on_error',
cb_data => 'on_data',
cb_closed => 'on_closed',
+ ev_error => 'error',
+ ev_data => 'data',
+ ev_closed => 'closed',
};
1;
3  lib/Reflex/Timeout.pm
View
@@ -9,7 +9,7 @@ has auto_start => ( isa => 'Bool', is => 'ro', default => 1 );
# TODO - There is a flaw in the design.
#
# Reflex::Timeout = cb_timeout => "on_done"
-# Reflex::Role::Timeout = method_emit $cb_timeout => "done"
+# Reflex::Role::Timeout = method_emit $cb_timeout => $p->ev_done()
#
# However, the user's on_done => callback() only works because the
# emitted event is "done". And this "done" is a constant, which means
@@ -22,6 +22,7 @@ has auto_start => ( isa => 'Bool', is => 'ro', default => 1 );
with 'Reflex::Role::Timeout' => {
delay => "delay",
cb_timeout => "on_done",
+ ev_timeout => "done",
auto_start => "auto_start",
method_start => "start",
method_stop => "stop",
2  lib/Reflex/UdpPeer.pm
View
@@ -14,6 +14,8 @@ with 'Reflex::Role::Recving' => {
method_stop => 'stop',
cb_datagram => 'on_datagram',
cb_error => 'on_error',
+ ev_datagram => 'datagram',
+ ev_error => 'error',
};
1;
Please sign in to comment.
Something went wrong with that request. Please try again.