view release on metacpan or  search on metacpan

lib/AnyEvent/Handle.pm  view on Meta::CPAN

   if (!$self->{_ww} && length $self->{wbuf}) {

      Scalar::Util::weaken $self;

      my $cb = sub {
         my $len = syswrite $self->{fh}, $self->{wbuf};

         if (defined $len) {
            substr $self->{wbuf}, 0, $len, "";

            $self->{_activity} = $self->{_wactivity} = AE::now;

            $self->{on_drain}($self)
               if $self->{low_water_mark} >= (length $self->{wbuf}) + (length $self->{_tls_wbuf})
                  && $self->{on_drain};

            delete $self->{_ww} unless length $self->{wbuf};
         } elsif ($! != EAGAIN && $! != EINTR && $! != EWOULDBLOCK && $! != WSAEWOULDBLOCK) {
            $self->_error ($!, 1);
         }
      };

      # try to write data immediately
      $cb->() unless $self->{autocork};

      # if still data left in wbuf, we need to poll
      $self->{_ww} = AE::io $self->{fh}, 1, $cb
         if length $self->{wbuf};

      if (
         defined $self->{wbuf_max}
         && $self->{wbuf_max} < length $self->{wbuf}
      ) {
         $self->_error (Errno::ENOSPC, 1), return;
      }
   };
}

our %WH;

# deprecated
sub register_write_type($$) {
   $WH{$_[0]} = $_[1];
}

sub push_write {
   my $self = shift;

   if (@_ > 1) {
      my $type = shift;

      @_ = ($WH{$type} ||= _load_func "$type\::anyevent_write_type"
            or Carp::croak "unsupported/unloadable type '$type' passed to AnyEvent::Handle::push_write")
           ->($self, @_);
   }

   # we downgrade here to avoid hard-to-track-down bugs,
   # and diagnose the problem earlier and better.

   if ($self->{tls}) {
      utf8::downgrade $self->{_tls_wbuf} .= $_[0];
      &_dotls ($self)    if $self->{fh};
   } else {
      utf8::downgrade $self->{wbuf}      .= $_[0];
      $self->_drain_wbuf if $self->{fh};
   }
}

=item $handle->push_write (type => @args)

Instead of formatting your data yourself, you can also let this module
do the job by specifying a type and type-specific arguments. You
can also specify the (fully qualified) name of a package, in which
case AnyEvent tries to load the package and then expects to find the
C<anyevent_write_type> function inside (see "custom write types", below).

Predefined types are (if you have ideas for additional types, feel free to
drop by and tell us):

=over 4

=item netstring => $string

Formats the given value as netstring
(http://cr.yp.to/proto/netstrings.txt, this is not a recommendation to use them).

=cut

register_write_type netstring => sub {
   my ($self, $string) = @_;

   (length $string) . ":$string,"
};

=item packstring => $format, $data

An octet string prefixed with an encoded length. The encoding C<$format>
uses the same format as a Perl C<pack> format, but must specify a single
integer only (only one of C<cCsSlLqQiInNvVjJw> is allowed, plus an
optional C<!>, C<< < >> or C<< > >> modifier).

=cut

register_write_type packstring => sub {
   my ($self, $format, $string) = @_;

   pack "$format/a*", $string
};

=item json => $array_or_hashref

Encodes the given hash or array reference into a JSON object. Unless you
provide your own JSON object, this means it will be encoded to JSON text
in UTF-8.

The default encoder might or might not handle every type of JSON value -
it might be limited to arrays and objects for security reasons. See the
C<json> constructor attribute for more details.

JSON objects (and arrays) are self-delimiting, so if you only use arrays
and hashes, you can write JSON at one end of a handle and read them at the
other end without using any additional framing.

The JSON text generated by the default encoder is guaranteed not to
contain any newlines: While this module doesn't need delimiters after or
between JSON texts to be able to read them, many other languages depend on
them.

A simple RPC protocol that interoperates easily with other languages is
to send JSON arrays (or objects, although arrays are usually the better
choice as they mimic how function argument passing works) and a newline
after each JSON text:

   $handle->push_write (json => ["method", "arg1", "arg2"]); # whatever
   $handle->push_write ("\012");
 
An AnyEvent::Handle receiver would simply use the C<json> read type and
rely on the fact that the newline will be skipped as leading whitespace:

   $handle->push_read (json => sub { my $array = $_[1]; ... });

Other languages could read single lines terminated by a newline and pass
this line into their JSON decoder of choice.

=item cbor => $perl_scalar

Encodes the given scalar into a CBOR value. Unless you provide your own
L<CBOR::XS> object, this means it will be encoded to a CBOR string not
using any extensions, if possible.

CBOR values are self-delimiting, so you can write CBOR at one end of
a handle and read them at the other end without using any additional
framing.

A simple nd very very fast RPC protocol that interoperates with
other languages is to send CBOR and receive CBOR values (arrays are
recommended):

   $handle->push_write (cbor => ["method", "arg1", "arg2"]); # whatever
 
An AnyEvent::Handle receiver would simply use the C<cbor> read type:

   $handle->push_read (cbor => sub { my $array = $_[1]; ... });

=cut

sub json_coder() {
   eval { require JSON::XS; JSON::XS->new->utf8 }
      || do { require JSON::PP; JSON::PP->new->utf8 }
}

register_write_type json => sub {
   my ($self, $ref) = @_;

   ($self->{json} ||= json_coder)
      ->encode ($ref)
};

sub cbor_coder() {
   require CBOR::XS;
   CBOR::XS->new
}

register_write_type cbor => sub {
   my ($self, $scalar) = @_;

   ($self->{cbor} ||= cbor_coder)
      ->encode ($scalar)
};

=item storable => $reference

Freezes the given reference using L<Storable> and writes it to the
handle. Uses the C<nfreeze> format.

=cut

register_write_type storable => sub {
   my ($self, $ref) = @_;

   require Storable unless $Storable::VERSION;

   pack "w/a*", Storable::nfreeze ($ref)
};

=back

=item $handle->push_shutdown

Sometimes you know you want to close the socket after writing your data
before it was actually written. One way to do that is to replace your
C<on_drain> handler by a callback that shuts down the socket (and set
C<low_water_mark> to C<0>). This method is a shorthand for just that, and
replaces the C<on_drain> callback with:

   sub { shutdown $_[0]{fh}, 1 }

This simply shuts down the write side and signals an EOF condition to the
the peer.

You can rely on the normal read queue and C<on_eof> handling
afterwards. This is the cleanest way to close a connection.

This method may invoke callbacks (and therefore the handle might be
destroyed after it returns).

=cut

sub push_shutdown {