EV-Kafka

 view release on metacpan or  search on metacpan

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

            warn "EV::Kafka: SASL $mech configured without TLS — "
                . "credentials will be sent over plaintext\n";
        }
    }

    return $self;
}

package EV::Kafka::Client;
use EV;
use Scalar::Util 'weaken';

sub _any_conn {
    my ($self) = @_;
    my $cfg = $self->{cfg};
    my $conn = $cfg->{bootstrap_conn};
    for my $c (values %{$cfg->{conns}}) {
        if ($c->connected) { $conn = $c; last }
    }
    return ($conn && $conn->connected) ? $conn : undef;
}

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

    my $cfg = $self->{cfg};
    return $cfg->{conns}{$node_id} if $cfg->{conns}{$node_id};

    my $info = $cfg->{broker_map}{$node_id};
    return undef unless $info;

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

    $cfg->{conns}{$node_id} = $conn;
    weaken(my $weak = $self);
    $conn->on_connect(sub {
        $weak->_drain_pending_for($node_id) if $weak;
    });
    $conn->connect($info->{host}, $info->{port}, 10.0);

    return $conn;
}

sub _configure_conn {
    my ($self, $conn) = @_;
    my $cfg = $self->{cfg};
    $conn->client_id($cfg->{client_id});
    if ($cfg->{tls}) {
        $conn->tls(1, $cfg->{tls_ca_file}, $cfg->{tls_skip_verify});
    }
    if ($cfg->{sasl}) {
        $conn->sasl($cfg->{sasl}{mechanism}, $cfg->{sasl}{username}, $cfg->{sasl}{password});
    }
    weaken(my $weak_cfg = $cfg);
    $conn->on_error(sub {
        $weak_cfg->{on_error}->($_[0]) if $weak_cfg && $weak_cfg->{on_error};
    });
}

sub _bootstrap_connect {
    my ($self, $cb) = @_;
    my $cfg = $self->{cfg};
    my @bs = @{$cfg->{bootstrap}};
    my $idx = 0;

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

        $self->_arm_metadata_timer;
    });
}

sub _arm_metadata_timer {
    my ($self) = @_;
    my $cfg = $self->{cfg};
    return if $cfg->{_meta_timer};
    my $interval = $cfg->{metadata_refresh} || 0;
    return if $interval <= 0;
    weaken(my $weak = $self);
    $cfg->{_meta_timer} = EV::timer $interval, $interval, sub {
        return unless $weak;
        $weak->_refresh_metadata unless $weak->{cfg}{meta_pending};
    };
}

sub _disarm_metadata_timer {
    my ($self) = @_;
    undef $self->{cfg}{_meta_timer};
}

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

    my $batch_bytes = 0;
    for my $b (@$batch) {
        $batch_bytes += length($b->{rec}{value} // '') + length($b->{rec}{key} // '') + 20;
    }

    if ($batch_bytes >= $cfg->{batch_size}) {
        $self->_flush_batch($topic, $partition, $conn);
    } elsif (!$cfg->{_linger_active}) {
        # start linger timer
        $cfg->{_linger_active} = 1;
        weaken(my $weak = $self);
        $cfg->{_linger_timer} = EV::timer $cfg->{linger_ms} / 1000.0, 0, sub {
            $cfg->{_linger_active} = 0;
            $weak->_flush_all_batches if $weak;
        };
    }
}

sub _flush_batch {
    my ($self, $topic, $partition, $conn) = @_;
    my $cfg = $self->{cfg};

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

        $popts{base_sequence}  = $saved_seq;
        $cfg->{next_sequence}{$bkey} = $saved_seq + scalar @records;
        $cfg->{_inflight}{$bkey} = 1;
    }

    $self->_add_txn_partition($topic, $partition) if $cfg->{_txn_active};

    # retry count persists on the batch across re-queues
    $cfg->{_batch_retries}{$bkey} //= 3;

    weaken(my $weak_self = $self);
    $conn->produce_batch($topic, $partition, \@records, \%popts, sub {
        my ($result, $err) = @_;
        delete $cfg->{_inflight}{$bkey} if $idempotent;

        my $retriable = 0;
        my $fatal_seq = 0;
        if (!$err && $result && ref $result->{topics} eq 'ARRAY') {
            for my $t (@{$result->{topics}}) {
                for my $p (@{$t->{partitions} // []}) {
                    my $ec = $p->{error_code} // 0;

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

        my ($topic, $partition) = split /:/, $bkey, 2;
        my $leader_id = $self->_get_leader($topic, $partition);
        unless (defined $leader_id) { $skipped++; next }
        my $conn = $self->_get_or_create_conn($leader_id);
        unless ($conn && $conn->connected) { $skipped++; next }
        $self->_flush_batch($topic, $partition, $conn);
    }
    # re-arm timer if batches were skipped (connection not yet ready)
    if ($skipped && keys %{$cfg->{batches}}) {
        $cfg->{_linger_active} = 1;
        weaken(my $weak = $self);
        $cfg->{_linger_timer} = EV::timer 0.1, 0, sub {
            $cfg->{_linger_active} = 0;
            $weak->_flush_all_batches if $weak;
        };
    }
}

sub produce_many {
    my ($self, $messages, $cb) = @_;
    my $remaining = scalar @$messages;

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

    undef $g->{heartbeat_timer};
}

sub _start_fetch_loop {
    my ($self) = @_;
    my $cfg = $self->{cfg};
    return if $cfg->{fetch_active};
    $cfg->{fetch_active} = 1;
    $cfg->{_fetch_in_flight} = 0;

    weaken(my $weak = $self);
    $cfg->{fetch_timer} = EV::timer 0, 0.1, sub {
        return unless $weak && $cfg->{fetch_active};
        # Skip ticks while a prior poll round is still in flight to avoid
        # duplicate per-(topic,partition) requests landing on the broker.
        return if $cfg->{_fetch_in_flight};
        $cfg->{_fetch_in_flight} = 1;
        $weak->poll(sub { $cfg->{_fetch_in_flight} = 0 });
    };
}

src/ppport.h  view on Meta::CPAN

sv_resetpvn|5.017005||Viu
SvRMAGICAL|5.003007||Viu
SvRMAGICAL_off|5.003007||Viu
SvRMAGICAL_on|5.003007||Viu
SvROK|5.003007|5.003007|
SvROK_off|5.003007|5.003007|
SvROK_on|5.003007|5.003007|
SvRV|5.003007|5.003007|
SvRV_const|5.010001||Viu
SvRV_set|5.009003|5.003007|p
sv_rvunweaken|5.027004|5.027004|
sv_rvweaken|5.006000|5.006000|
SvRVx|5.003007||Viu
SvRX|5.009005|5.003007|p
SvRXOK|5.009005|5.003007|p
SV_SAVED_COPY|5.009005||Viu
SvSCREAM|5.003007||Viu
SvSCREAM_off|5.003007||Viu
SvSCREAM_on|5.003007||Viu
sv_setbool|5.035004|5.035004|
sv_setbool_mg|5.035004|5.035004|
sv_setgid|5.019001||Viu