AnyEvent-MPRPC

 view release on metacpan or  search on metacpan

Changes  view on Meta::CPAN

Revision history for Perl extension AnyEvent::MPRPC

0.20 2013-02-01T21:28:13

    - Add 2 callbacks (before/after_connect) and 1 option (connect_timeout)
      (hirose31)

0.19

    - Split AnyEvent::MessagePack to it's own distribution.
      (tokuhirom)

0.18

    - AnyEvent::MPRPC::Client 'call' method can take scalar as RPC parameters
      (It has been able to take only ArrayRef).

0.17

    commit 4f6db620d1b0b86f776946f4d8688dc9fc69c966
    Author: Maxime Soulé <btik-git@scoubidou.com>
    Date:   Mon Jul 30 10:17:51 2012 +0200

        A read_type AnyEvent handler must call the callback only once per
        call, else one have a risk of memleak by pushing/shifting too many
        "read" callbacks in the read queue.
        And so each read callback should re-arm itself (if needed) in the read
        queue. Using on_read callback to do that is a mistake, as on_read can
        be called several times before a complete messagepack packet is
        received.

0.16

    - Fixes RT#74518: issue
    Bug #74518 for AnyEvent-MPRPC: AnyEvent::MessagePack read type corrupts responses
    (fixed By Max++, and reported by asguthrie++)

lib/AnyEvent/MPRPC/Client.pm  view on Meta::CPAN

has _next_id => (
    is      => 'ro',
    isa     => 'CodeRef',
    lazy    => 1,
    default => sub {
        my $id = 0;
        sub { ++$id };
    },
);

has _callbacks => (
    is      => 'ro',
    isa     => 'HashRef',
    lazy    => 1,
    default => sub { +{} },
);

has _connection_guard => (
    is  => 'rw',
    isa => 'Object',
);

lib/AnyEvent/MPRPC/Client.pm  view on Meta::CPAN

    ];

    if ($self->handler) {
        $self->handler->push_write( msgpack => $request );
    }
    else {
        push @{ $self->_request_pool }, $request;
    }

    # $msgid is stringified, but $request->{MP_RES_MSGID] is still IV
    $self->_callbacks->{ $msgid } = AnyEvent->condvar;
}

sub _handle_response_cb {
    my $self = shift;

    weaken $self;

    return sub {
        $self || return;

        my ($handle, $res) = @_;

        my $d = delete $self->_callbacks->{ $res->[MP_RES_MSGID] };

        if (my $error = $res->[MP_RES_ERROR]) {
            if ($d) {
                $d->croak($error);
            } else {
                Carp::croak($error);
            }
        }

        $handle->unshift_read(msgpack => $self->_handle_response_cb);

lib/AnyEvent/MPRPC/Server.pm  view on Meta::CPAN

    isa     => 'HashRef',
    default => sub { {} },
);

has _handlers => (
    is      => 'ro',
    isa     => 'ArrayRef',
    default => sub { [] },
);

has _callbacks => (
    is      => 'ro',
    isa     => 'HashRef',
    lazy    => 1,
    default => sub { {} },
);

no Any::Moose;

sub BUILD {
    my $self = shift;

lib/AnyEvent/MPRPC/Server.pm  view on Meta::CPAN

        $handle->unshift_read(msgpack => $self->_dispatch_cb($indicator));

        $self->_handlers->[ fileno($fh) ] = $handle;
    }) unless defined $self->server;
    weaken $self;

    $self;
}

sub reg_cb {
    my ($self, %callbacks) = @_;

    while (my ($method, $callback) = each %callbacks) {
        $self->_callbacks->{ $method } = $callback;
    }
}

sub _dispatch_cb {
    my ($self, $indicator) = @_;

    weaken $self;

    return sub {
        $self || return;

        my ($handle, $request) = @_;
        $self->on_dispatch->($indicator, $handle, $request);
        return if $handle->destroyed;

        $handle->unshift_read(msgpack => $self->_dispatch_cb($indicator));

        return unless $request and ref $request eq 'ARRAY';

        my $target = $self->_callbacks->{ $request->[MP_REQ_METHOD] };

        my $id = $request->[MP_REQ_MSGID];
        $indicator = "$indicator:$id";

        my $res_cb = sub {
            my $type   = shift;
            my $result = @_ > 1 ? \@_ : $_[0];

            $handle->push_write( msgpack => [
                MP_TYPE_RESPONSE,

lib/AnyEvent/MPRPC/Server.pm  view on Meta::CPAN

=item on_accept => $cb->($fh, $host, $port)

=item on_dispatch => $cb->($indicator, $handle, $request);

=item handler_options => 'HashRef'

Hashref options of L<AnyEvent::Handle> that is used to handle client connections.

=back

=head2 reg_cb (%callbacks)

Register MessagePack RPC methods.

    $server->reg_cb(
        echo => sub {
            my ($res_cv, @params) = @_;
            $res_cv->result(@params);
        },
        sum => sub {
            my ($res_cv, @params) = @_;