view release on metacpan or  search on metacpan

local/lib/perl5/IO/Async/Channel.pm  view on Meta::CPAN

May only be set on an async mode channel. If present, will be invoked when the
channel gets closed by the peer.

 $on_eof->( $channel )

=back

=cut

sub _init
{
   my $self = shift;
   my ( $params ) = @_;

   defined $params->{codec} or $params->{codec} = "Storable";

   $self->SUPER::_init( $params );
}

sub configure
{
   my $self = shift;
   my %params = @_;

   foreach (qw( on_recv on_eof )) {
      next unless exists $params{$_};
      $self->{mode} and $self->{mode} eq "async" or
         croak "Can only configure $_ in async mode";

      $self->{$_} = delete $params{$_};
      $self->_build_stream;
   }

   if( my $codec = delete $params{codec} ) {
      @{ $self }{qw( encode decode )} = (
         $self->can( "_make_codec_$codec" ) or croak "Unrecognised codec name '$codec'"
      )->();
   }

   $self->SUPER::configure( %params );
}

sub _make_codec_Storable
{
   require Storable;

   return
      \&Storable::freeze,
      \&Storable::thaw;
}

sub _make_codec_Sereal
{
   require Sereal::Encoder;
   require Sereal::Decoder;

   my $encoder;
   my $decoder;

   # "thread safety" to Sereal::{Encoder,Decoder} means that the variables get
   # reset to undef in new threads. We should defend against that.

   return
      sub { ( $encoder ||= Sereal::Encoder->new )->encode( $_[0] ) },
      sub { ( $decoder ||= Sereal::Decoder->new )->decode( $_[0] ) };
}

=head2 send

   $channel->send( $data )

Pushes the data stored in the given Perl reference into the FIFO of the
Channel, where it can be received by the other end. When called on a
synchronous mode Channel this method may block if a C<write()> call on the
underlying filehandle blocks. When called on an asynchronous mode channel this
method will not block.

=cut

sub send
{
   my $self = shift;
   my ( $data ) = @_;

   $self->send_encoded( $self->{encode}->( $data ) );
}

=head2 send_encoded

   $channel->send_encoded( $record )

A variant of the C<send> method; this method pushes the byte record given.
This should be the result of a call to C<encode>.

=cut

sub send_encoded
{
   my $self = shift;
   my ( $record ) = @_;

   my $bytes = pack( "I", length $record ) . $record;

   defined $self->{mode} or die "Cannot ->send without being set up";

   return $self->_send_sync( $bytes )  if $self->{mode} eq "sync";
   return $self->_send_async( $bytes ) if $self->{mode} eq "async";
}

=head2 encode

   $record = $channel->encode( $data )

Takes a Perl reference and returns a serialised string that can be passed to
C<send_encoded>. The following two forms are equivalent

 $channel->send( $data )
 $channel->send_encoded( $channel->encode( $data ) )

This is provided for the use-case where data needs to be serialised into a
fixed string to "snapshot it" but not sent yet; the returned string can be