view release on metacpan or  search on metacpan

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

    }

    # Store config
    my $cfg = {
        bootstrap    => \@bootstrap,
        client_id    => delete $opts{client_id} // 'ev-kafka',
        tls          => delete $opts{tls} // 0,
        tls_ca_file  => delete $opts{tls_ca_file},
        tls_skip_verify => delete $opts{tls_skip_verify} // 0,
        sasl         => delete $opts{sasl},
        on_error     => delete $opts{on_error} // sub { die "EV::Kafka: @_\n" },
        on_connect   => delete $opts{on_connect},
        on_message   => delete $opts{on_message},
        acks         => delete $opts{acks} // -1,
        linger_ms    => delete $opts{linger_ms} // 5,
        batch_size   => delete $opts{batch_size} // 16384,
        partitioner  => delete $opts{partitioner},
        compression      => delete $opts{compression},     # 'lz4', 'gzip', or undef
        idempotent       => delete $opts{idempotent} // 0,
        transactional_id => delete $opts{transactional_id}, # enables transactions
        fetch_max_wait_ms => delete $opts{fetch_max_wait_ms} // 500,
        fetch_max_bytes   => delete $opts{fetch_max_bytes} // 1048576,
        fetch_min_bytes   => delete $opts{fetch_min_bytes} // 1,
        metadata_refresh  => delete $opts{metadata_refresh} // 300,
        flush_timeout     => delete $opts{flush_timeout} // 30,
    };

    Carp::croak("EV::Kafka: unknown option(s): " . join(', ', sort keys %opts))
        if %opts;

    # Internal state
    $cfg->{closed}    = 0;     # set by close()/DESTROY; gates everything
    $cfg->{conns}     = {};    # node_id => EV::Kafka::Conn
    $cfg->{meta}      = undef; # latest metadata response
    $cfg->{leaders}   = {};    # "topic:partition" => node_id
    $cfg->{broker_map}= {};    # node_id => {host, port}
    $cfg->{connected} = 0;
    $cfg->{meta_pending} = 0;
    $cfg->{pending_ops} = [];  # ops waiting for metadata

    # Producer state
    $cfg->{batches}  = {};     # "topic:partition" => [{rec, cb}]
    $cfg->{next_sequence} = {}; # "topic:partition" => next sequence number
    $cfg->{producer_id}    = -1;
    $cfg->{producer_epoch} = -1;
    $cfg->{rr_counter} = 0;
    # InitProducerId state machine: idle | init | ready | failed:$msg.
    # Single-flight: 'init' queues completion callbacks in _pid_cbs.
    $cfg->{_pid}      = 'idle';
    $cfg->{_pid_cbs}  = [];
    $cfg->{_txn_gen}  = 0;     # bumped on every txn state transition
    $cfg->{_conn_state} = {};  # "$conn" => {auto_reconnect, was_ready}

    # Consumer state
    $cfg->{assignments} = [];  # [{topic, partition, offset}]
    $cfg->{fetch_active} = 0;
    $cfg->{group} = undef;

    my $self = bless { cfg => $cfg, loop => $loop }, "${class}::Client";

    # Warn on credentials over plaintext.
    if ($cfg->{sasl} && !$cfg->{tls}) {
        my $mech = $cfg->{sasl}{mechanism} // '';
        if ($mech eq 'PLAIN' || $mech =~ /^SCRAM-/) {
            warn "EV::Kafka: SASL $mech configured without TLS — "
                . "credentials will be sent over plaintext\n";
        }
    }

    return $self;
}

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

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

sub _get_or_create_conn {
    my ($self, $node_id) = @_;
    my $cfg = $self->{cfg};
    return undef if $cfg->{closed};
    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);
    $conn->auto_reconnect(1, 1000);
    $cfg->{_conn_state}{"$conn"}{auto_reconnect} = 1;

    $cfg->{conns}{$node_id} = $conn;
    weaken(my $weak = $self);
    $conn->on_connect(sub {
        my $s = $weak or return;
        my $cfg = $s->{cfg};
        return if $cfg->{closed};
        $cfg->{_conn_state}{"$conn"}{was_ready} = 1;
        $s->_drain_pending_for($node_id);
    });
    $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}) {
        # Omit an undef CA: it would arrive in XS as "" and break
        # SSL_CTX_load_verify_locations (NULL = system trust store).
        my @tls_args = (1);
        if (defined $cfg->{tls_ca_file}) {
            push @tls_args, $cfg->{tls_ca_file}, $cfg->{tls_skip_verify};