EV-Kafka

 view release on metacpan or  search on metacpan

README  view on Meta::CPAN

  Object lifetime
    The connection is torn down when its last reference drops, or when
    "DESTROY" is called explicitly; explicit destruction is idempotent.
    Pending request callbacks are invoked with a 'destroyed' error during
    teardown. Once destroyed, the object is inert: every subsequent method
    call on it (from any copy of the reference) croaks with
    "EV::Kafka::Conn: method called on destroyed connection" -- including
    calls made from callbacks running during the teardown itself (those
    exceptions are caught by the callback dispatcher and reported via
    warn()). A conn passed a custom "EV::Loop" holds a reference on it, so
    the loop cannot be destroyed while the conn is alive.

UTILITY FUNCTIONS
  EV::Kafka::_murmur2($key)
    Kafka-compatible murmur2 hash. Returns a non-negative 31-bit integer.

  EV::Kafka::_crc32c($data)
    CRC32C checksum (Castagnoli). Used internally for RecordBatch integrity.

  EV::Kafka::_error_name($code)
    Convert Kafka error code to string name.

eg/transaction.pl  view on Meta::CPAN

        my @processed = @$batch;
        $batch = [];

        $kafka->send_offsets_to_transaction($group, sub {
            $kafka->commit_transaction(sub {
                printf "committed transaction with %d records\n",
                    scalar @processed;
            });
        });
    };
    # keep ref alive
    $kafka->{_eos_tick} = $tick;

    $kafka->poll;  # kick the fetch loop
});

EV::run;

lib/EV/Kafka.pm  view on Meta::CPAN

        } elsif ($cfg->{on_error}) {
            $cfg->{on_error}->("all bootstrap brokers unreachable");
        }
        return;
    }
    my ($host, $port) = @{$bs[$idx]};

    my $conn = EV::Kafka::Conn::_new('EV::Kafka::Conn', $self->{loop});
    $self->_configure_conn($conn);

    # Keeps the attempt conn alive until on_connect/on_error fires,
    # without a closure cycle.
    $cfg->{_bootstrap_attempt} = $conn;

    weaken(my $weak = $self);
    # Replace the forwarding on_error from _configure_conn with the
    # try-next-broker handler for the duration of the bootstrap.
    $conn->on_error(sub {
        my $s = $weak or return;
        my $cfg = $s->{cfg};
        return if $cfg->{closed};

lib/EV/Kafka.pm  view on Meta::CPAN

The connection is torn down when its last reference drops, or when
C<DESTROY> is called explicitly; explicit destruction is idempotent.
Pending request callbacks are invoked with a C<'destroyed'> error during
teardown. Once destroyed, the object is inert: every subsequent method
call on it (from any copy of the reference) croaks with
C<EV::Kafka::Conn: method called on destroyed connection> -- including
calls made from callbacks running during the teardown itself (those
exceptions are caught by the callback dispatcher and reported via
C<warn()>). A conn passed a custom C<EV::Loop> holds a reference on it,
so the loop cannot be destroyed while the conn is alive.

=head1 UTILITY FUNCTIONS

=head2 EV::Kafka::_murmur2($key)

Kafka-compatible murmur2 hash. Returns a non-negative 31-bit integer.

=head2 EV::Kafka::_crc32c($data)

CRC32C checksum (Castagnoli). Used internally for RecordBatch integrity.

t/20_mock_conn.t  view on Meta::CPAN

sub frame_response {
    my ($corr_id, $body) = @_;
    my $payload = i32($corr_id) . $body;
    return i32(length $payload) . $payload;
}

# Accept the connection and parse the request to recover the correlation id.
my $client_fh;
my @incoming;        # bytes read from client
my $request_corr_id;
my $client_read_w;   # kept alive by closure on the outer lexical

my $accept_w = EV::io fileno($server), EV::READ, sub {
    $client_fh = $server->accept or return;
    $client_fh->blocking(0);
    note "mock broker accepted connection";

    $client_read_w = EV::io fileno($client_fh), EV::READ, sub {
        my $buf;
        my $n = sysread $client_fh, $buf, 4096;
        if (!defined $n || $n == 0) {

t/22_lifecycle.t  view on Meta::CPAN

sub i32 { pack 'N', $_[0] }

my $apis_body =
      i16(0)         # no error
    . i32(2)         # 2 entries
    . i16(0) . i16(0) . i16(7)     # API_PRODUCE, v0..v7
    . i16(18) . i16(0) . i16(0);   # API_API_VERSIONS, v0..v0

# Start a mock broker on a free port. $on_request->($api_key, $corr_id,
# $send, $ctx) runs for each full request frame; $send->($corr, $body)
# frames and writes a response. $ctx keeps watchers/sockets/timers alive
# and offers $ctx->{close_client}. Returns ($port, $ctx).
sub mock_broker {
    my (%opt) = @_;
    my $server = IO::Socket::INET->new(
        LocalAddr => '127.0.0.1',
        LocalPort => 0,
        Listen    => 1,
        Proto     => 'tcp',
        ReuseAddr => 1,
    ) or BAIL_OUT "cannot bind localhost listener: $!";

t/22_lifecycle.t  view on Meta::CPAN

                $on_request->($api, $corr, $send, $ctx);
            }
        };
    };

    return ($server->sockport, $ctx);
}

sub timeout_w { EV::timer 5, 0, sub { diag "test timed out"; EV::break } }

# --- M1: a parsed response must not stay alive past its event -------------
# Broker answers the first Metadata request immediately and the second
# 0.2s later, so the two callbacks run in SEPARATE watcher events.  The
# first callback keeps only a weak ref to its result; if response mortals
# are pinned until EV::run returns (the bug), $saved is still defined when
# the second callback runs.
{
    my $meta_body = i32(0) . i32(1) . i32(0);  # v1: 0 brokers, controller 1, 0 topics
    my $meta_seen = 0;
    my ($port, $broker) = mock_broker(on_request => sub {
        my ($api, $corr, $send, $ctx) = @_;