view release on metacpan or  search on metacpan

lib/MR/IProto/Connection/Async.pm  view on Meta::CPAN

            $self->_clear_no_reply();
            $server->_debug('closing socket') if $server->debug >= 1;
            $handle->destroy();
            $self->_try_to_send();
            $_->(undef, undef, $message, $errno) foreach @callbacks;
            return;
        },
        on_timeout => sub {
            my ($handle) = @_;
            return unless keys %{$self->_callbacks};
            $handle->_error( Errno::ETIMEDOUT ) if keys %{$self->_callbacks};
            return;
        },
    );
}

sub _build__on_drain {
    my ($self) = @_;
    my $server = $self->server;
    weaken($self);
    weaken($server);
    return sub {
        my ($handle) = @_;
        if( $self->_has_no_reply() ) {
            foreach my $sync ( @{$self->_no_reply} ) {
                $server->_recv_finished($sync, undef, undef);
                $self->_in_progress( $self->_in_progress - 1 );
                delete($self->_callbacks->{$sync})->(undef, undef);
            }
            $self->_clear_no_reply();
            $self->_try_to_send();
            $handle->on_drain(undef);
        }
        return;
    };
}

sub _build__queue {
    my ($self) = @_;
    return [];
}

sub _build__callbacks {
    my ($self) = @_;
    return {};
}

sub _build__no_reply {
    my ($self) = @_;
    return [];
}

around _choose_sync => sub {
    my ($orig, $self) = @_;
    my $sync;
    my $callbacks = $self->_callbacks;
    for( 1 .. 50 ) {
        $sync = $self->$orig();
        return $sync unless exists $callbacks->{$sync};
    }
    die "Can't choose sync value after 50 iterations";
};

sub Close { die "This is not what should be done" }

=back

=head1 SEE ALSO

L<MR::IProto::Connection>, L<MR::IProto::Cluster::Server>.

=cut

no Mouse;
__PACKAGE__->meta->make_immutable();

1;