package Kafka::MockIO;

=head1 NAME

Kafka::MockIO - object interface to simulate communications with the Apache Kafka
server via socket.

=head1 VERSION

This documentation refers to C<Kafka::MockIO> version 1.08 .

=cut

use 5.010;
use strict;
use warnings;

our $VERSION = '1.08';

use Carp;
use Const::Fast;
use List::Util qw(
    min
);
use Params::Util qw(
    _HASH
    _NONNEGINT
    _NUMBER
    _POSINT
    _STRING
);
use Scalar::Util qw(
    blessed
);
use Sub::Install;

use Kafka qw(
    $COMPRESSION_NONE
    $ERROR_MISMATCH_ARGUMENT
    $ERROR_NO_ERROR
    $ERROR_NOT_BINARY_STRING
    $ERROR_CANNOT_RECV
    $KAFKA_SERVER_PORT
    $RECEIVE_LATEST_OFFSETS    $RECEIVE_EARLIEST_OFFSET
    $REQUEST_TIMEOUT
);
use Kafka::Internals qw(
    $APIKEY_PRODUCE
    $APIKEY_FETCH
    $APIKEY_OFFSET
    $APIKEY_METADATA
    $APIKEY_SASLHANDSHAKE
    $MAX_SOCKET_REQUEST_BYTES
);
use Kafka::IO;
use Kafka::MockProtocol qw(
    decode_fetch_request
    decode_metadata_request
    decode_offset_request
    decode_produce_request
    decode_saslhandshake_request
    encode_fetch_response
    encode_metadata_response
    encode_offset_response
    encode_produce_response
    encode_saslhandshake_response
);
use Kafka::Protocol qw(
    encode_saslhandshake_request
);



=head1 DESCRIPTION

This module is not a user module.

The main features of the C<Kafka::MockIO> class are:

=over 3

=item *

Emulates an object oriented model of communication (L<Kafka::IO|Kafka::IO> class).

=item *

Simplistically emulates interaction with kafka server.

=back

Examples see C<t/*_mock_io.t>.

=cut

const my $MOCKED_PACKAGE            => 'Kafka::IO';

=head2 EXPORT

Use Kafka::MockIO only with the following information.

The following constants are available for export

=cut

=head3 C<$TOPIC>

Topic name.

=cut
const our $TOPIC                    => 'mytopic';

=head3 C<$PARTITION>

0 - Partition number.

=cut
const our $PARTITION                => 0;

=head3 C<$KAFKA_MOCK_SERVER_PORT>

C<$KAFKA_MOCK_SERVER_PORT> is the default Apache Kafka server port
that can be imported from the L<Kafka|Kafka> module and = 9092.

=cut
const our $KAFKA_MOCK_SERVER_PORT   => $KAFKA_SERVER_PORT;

=head3 C<$KAFKA_MOCK_HOSTNAME>

'localhost' - C<$KAFKA_MOCK_HOSTNAME> is the default local host name.

=cut
const our $KAFKA_MOCK_HOSTNAME      => 'localhost';



my %_reinstall = (
    new                         => [ \&Kafka::IO::new,      \&new ],
    send                        => [ \&Kafka::IO::send,     \&send ],
    receive                     => [ \&Kafka::IO::receive,  \&receive ],
    close                       => [ \&Kafka::IO::close,    \&close ],
    _is_alive                   => [ \&Kafka::IO::_is_alive, \&_is_alive ],
    _decoded_topic_partition    => [ sub {},                \&_decoded_topic_partition ],
    _verify_string              => [ sub {},                \&_verify_string ],
    add_special_case            => [ sub {},                \&add_special_case ],
    del_special_case            => [ sub {},                \&del_special_case ],
    special_cases               => [ sub {},                \&special_cases ],
);

my ( $ApiKey, $encoded_response );

our %_received_data;                                        # (
                                                            #   topic   => {
                                                            #       partition   => [
                                                            #           [ Key, Value ],
                                                            #           ...,
                                                            #       ],
                                                            #       ...,
                                                            #   }
                                                            #   ...,
                                                            # )

my %_special_cases;                                         # encoded_request => encoded_response, ...

my $decoded_produce_response = {
    CorrelationId                           => 0,           # for example
    topics                                  => [
        {
            TopicName                       => $TOPIC,      # for example
            partitions                      => [
                {
                    Partition               => $PARTITION,
                    ErrorCode               => 0,
                    Offset                  => 0,
                },
            ],
        },
    ],
};

my $decoded_fetch_response = {
    CorrelationId                           => 0,           # for example
    topics                                  => [
        {
            TopicName                       => $TOPIC,      # for example
            partitions                      => [
                {
                    Partition               => $PARTITION,
                    ErrorCode               => 0,
                    HighwaterMarkOffset     => 2,
                    MessageSet              => [
                        #{
                        #    Offset          => 0,
                        #    MagicByte       => 0,
                        #    Attributes      => 0,
                        #    Key             => q{},
                        #    Value           => 'Hello!',
                        #},
                        #{
                        #    Offset          => 1,
                        #    MagicByte       => 0,
                        #    Attributes      => 0,
                        #    Key             => q{},
                        #    Value           => 'Hello, World!',
                        #},
                    ],
                },
            ],
        },
    ],
};

my $decoded_offset_response = {
    CorrelationId                       => 0,           # for example
    topics                              => [
        {
            TopicName                   => $TOPIC,      # for example
            PartitionOffsets            => [
                {
                    Partition           => $PARTITION,
                    ErrorCode           => 0,
                    Offset              => [
#                                           0,
                    ],
                },
            ],
        },
    ],
};

my $decoded_metadata_response = {
    CorrelationId                       => 0,           # for example
    Broker                              => [
        {
            NodeId                      => 2,
            Host                        => $KAFKA_MOCK_HOSTNAME,
            Port                        => $KAFKA_MOCK_SERVER_PORT + 2,
        },
        {
            NodeId                      => 0,
            Host                        => $KAFKA_MOCK_HOSTNAME,
            Port                        => $KAFKA_MOCK_SERVER_PORT,
        },
        {
            NodeId                      => 1,
            Host                        => $KAFKA_MOCK_HOSTNAME,
            Port                        => $KAFKA_MOCK_SERVER_PORT + 1,
        },
    ],
    TopicMetadata                       => [
        {
            ErrorCode                   => 0,
            TopicName                   => $TOPIC,      # for example
            PartitionMetadata           => [
                {
                    ErrorCode           => 0,
                    Partition           => $PARTITION,
                    Leader              => 2,
                    Replicas            => [    # of ReplicaId
                                           2,
                                           0,
                                           1,
                    ],
                    Isr                 => [    # of ReplicaId
                                           2,
                                           0,
                                           1,
                    ],
                },
            ],
        },
    ],
};

=rem
my $decoded_saslhandshake_request = {
    CorrelationId => $CorrelationId,
    ClientId      => q{},
    ApiVersion    => 0,
    Mechanism     => 'SCRAM-SHA-512',
};
my $normal_encoded_saslhandshake_request = encode_saslhandshake_request( $decoded_saslhandshake_request );
=cut
#-- public functions -----------------------------------------------------------

=head2 FUNCTIONS

The following methods are defined for the C<Kafka::MockIO> class:

=cut

=head3 C<override>

Override the C<Kafka::IO> class methods.

=cut
sub override {
    foreach my $method ( keys %_reinstall ) {
        Sub::Install::reinstall_sub( {
            code    => $_reinstall{ $method }->[1],
            into    => $MOCKED_PACKAGE,
            as      => $method,
        } );
    }

    return;
}

=head3 C<restore>

Restore the C<Kafka::IO> class methods.

=cut
sub restore {
    foreach my $method ( keys %_reinstall ) {
        Sub::Install::reinstall_sub( {
            code    => $_reinstall{ $method }->[0],
            into    => $MOCKED_PACKAGE,
            as      => $method,
        } );
    }

    return;
}

=head3 C<add_special_case( $cases )>

Adds special cases for use in the simulation of interaction with kafka server.

This function takes the following arguments:

=over 3

=item C<$cases>

C<$cases> is a reference to the hash representing
the special cases.

The keys of the hash should be binary-encoded query string to kafka server.
The values of each key must be encoded binary string of the expected response.

=back

=cut
sub add_special_case {
    my ( $cases ) = @_;

    blessed( $cases )
        and confess 'Do not use a class method as a method of the object';
    _HASH( $cases )
        or confess 'requires a hash request-response';
    foreach my $encoded_request ( keys %{ $cases } ) {
        _STRING( $cases->{ $encoded_request } )
            or confess 'hash must contain encoded responses';
    }

    foreach my $encoded_request ( keys %{ $cases } ) {
        $_special_cases{ $encoded_request } = $cases->{ $encoded_request };
    }

    return;
}

=head3 C<del_special_case( $encoded_request )>

Removes the special case.

This function takes the following arguments:

=over 3

=item C<$encoded_request>

Binary string of the encoded request.

=back

=cut
sub del_special_case {
    my ( $encoded_request ) = @_;

    blessed( $encoded_request )
        and confess 'Do not use a class method as a method of the object';

    delete $_special_cases{ $encoded_request };

    return;
}

=head3 C<special_cases>

Returns a reference to a hash of special cases.

The keys of the hash are binary strings encoded requests to kafka server.
The value of each key is encoded binary string of the expected response.

=cut
sub special_cases {
    return \%_special_cases;
}

#-- private functions ----------------------------------------------------------

#-- constructor ----------------------------------------------------------------

=head2 CONSTRUCTOR

=head3 C<new>

Constructor emulation (C<Kafka::IO->new>).

=cut
sub new {
    my ( $class, @args ) = @_;

    my $self = bless {
        host        => q{},
        port        => $KAFKA_SERVER_PORT,
        timeout     => $REQUEST_TIMEOUT,
    }, $class;

    while ( @args ) {
        my $k = shift @args;
        next unless defined $k;
        $self->{ $k } = shift @args if exists $self->{ $k };
    }

    if    ( !( defined( $self->{host} ) && defined( _STRING( $self->{host} ) ) && !utf8::is_utf8( $self->{host} ) ) )   { Kafka::IO::_error( $self, $ERROR_MISMATCH_ARGUMENT, 'Kafka::IO->new - host' ); }
    elsif ( !_POSINT( $self->{port} ) )                                 { Kafka::IO::_error( $self, $ERROR_MISMATCH_ARGUMENT, 'Kafka::IO->new - port' ); }
    elsif ( !( _NUMBER( $self->{timeout} ) && $self->{timeout} > 0 ) )  { Kafka::IO::_error( $self, $ERROR_MISMATCH_ARGUMENT, 'Kafka::IO->new - timeout' ); }
    else  {
        $self->{socket} = 'fake true value';
    }

    return $self;
}

#-- public attributes ----------------------------------------------------------

#-- public methods -------------------------------------------------------------

=head2 METHODS

The following methods are defined for the C<Kafka::MockIO> class:

=cut

=head3 C<send>

Method emulation (C<Kafka::IO::send>).

=cut
use Authen::SCRAM::Server;
use Encode qw/encode_utf8/;
use MIME::Base64 qw/decode_base64/;
use PBKDF2::Tiny qw/derive digest_fcn hmac/;

my %CRED_INPUTS = (
    test_user  => [ 'QSXCR+Q6sek8bf92', 'test_password', 4096 ],
);

my %CRED;
my $SASL_SCRAM_STEP = 0;

sub get_cred {
    my $user = shift;
    return @{ $CRED{$user} || [] };
}

sub _hmac {
    my ( $sha, $sha_block, $key, $data ) = @_;
    $key = $sha->($key) if length($key) > $sha_block;
    return hmac( $data, $key, $sha, $sha_block );
}

my $scram_server;
sub install_cred {
    my ($type) = @_;
    $scram_server = Authen::SCRAM::Server->new(
        digest => $type,
        credential_cb => \&get_cred,
    );
    my ( $sha, $sha_block ) = digest_fcn($type);
    for my $user ( keys %CRED_INPUTS ) {
        my ( $salt, $pw, $i ) = @{ $CRED_INPUTS{$user} };
        $salt = decode_base64($salt);
        my $salted_password = derive($type, encode_utf8($pw), $salt, $i );
        my $client_key = _hmac( $sha, $sha_block, $salted_password, "Client Key" );
        my $stored_key = $sha->($client_key);
        my $server_key = _hmac( $sha, $sha_block, $salted_password, "Server Key" );
        $CRED{$user} = [ $salt, $stored_key, $server_key, $i ];
    }
}

sub send {
    my ( $self, $message ) = @_;

    my $description = 'Kafka::IO->send';
    defined( _STRING( $message ) )
        or Kafka::IO::_error( $self, $ERROR_MISMATCH_ARGUMENT, $description );
    utf8::is_utf8( $message )
        and Kafka::IO::_error( $self, $ERROR_NOT_BINARY_STRING, $description );
    ( my $len = length( $message .= q{} ) ) <= $MAX_SOCKET_REQUEST_BYTES
        or Kafka::IO::_error( $self, $ERROR_MISMATCH_ARGUMENT, $description );
    Kafka::IO::_debug_msg( $self, $message, 'Request to', 'green' )
        if Kafka::IO->debug_level == 1;
    if ( exists $_special_cases{ $message } ) {
        $encoded_response = $_special_cases{ $message };
        return length( $message );
    }

    undef $encoded_response;
    if ($SASL_SCRAM_STEP == 1) {
        my ($first_client_message_len, $first_client_message) = unpack "l>a*", $message;
        my $first_server_message = encode_utf8($scram_server->first_msg($first_client_message));
        $encoded_response = pack "l>a*", length($first_server_message), $first_server_message;
        $SASL_SCRAM_STEP = 2;
        return length($message);
    } elsif ($SASL_SCRAM_STEP == 2) {
        my ($final_client_message_len, $final_client_message) = unpack "l>a*", $message;
        my $final_server_message = $scram_server->final_msg($final_client_message);
        $encoded_response = pack "l>a*", length($final_server_message), $final_server_message;
        $SASL_SCRAM_STEP = 0;
        return length($message);
    }

    $ApiKey = unpack( q{
        x[l]                # Size
        s>                  # ApiKey
    }, $message );

    # Set up the response
    if ( $ApiKey == $APIKEY_PRODUCE ) {
        my $decoded_produce_request = decode_produce_request( \$message )
            or Kafka::IO::_error( $self, $ERROR_MISMATCH_ARGUMENT, $description );
        my ( $topic, $partition ) = $self->_decoded_topic_partition( $decoded_produce_request, $decoded_produce_response );
        $partition // return;

        if ( !exists( $_received_data{ $topic }->{ $partition } ) ) {
            $_received_data{ $topic }->{ $partition } = [];
        }
        my $data = $_received_data{ $topic }->{ $partition };

        $decoded_produce_response->{topics}->[0]->{partitions}->[0]->{Offset} = scalar @{ $data };
        foreach my $Message ( @{ $decoded_produce_request->{topics}->[0]->{partitions}->[0]->{MessageSet} } ) {
            foreach my $key_name ( 'Key', 'Value' ) {
                if ( defined( my $value = $Message->{ $key_name } ) ) {
                    $self->_verify_string( $value, "$description ($key_name)" )
                        or return;
                }
            }
            push @{ $data }, [ $Message->{Key} // q{}, $Message->{Value} // q{} ];
        }

        $encoded_response = encode_produce_response( $decoded_produce_response )
            or Kafka::IO::_error( $self, $ERROR_MISMATCH_ARGUMENT, $description );
    }

    elsif ( $ApiKey == $APIKEY_FETCH ) {
        my $decoded_fetch_request = decode_fetch_request( \$message )
            or Kafka::IO::_error( $self, $ERROR_MISMATCH_ARGUMENT, $description );
        my ( $topic, $partition ) = $self->_decoded_topic_partition( $decoded_fetch_request, $decoded_fetch_response );
        $partition // return;

        my $partition_data = $decoded_fetch_request->{topics}->[0]->{partitions}->[0];
        my $FetchOffset = $partition_data->{FetchOffset};
        my $MaxBytes    = $partition_data->{MaxBytes};

        $partition_data = $decoded_fetch_response->{topics}->[0]->{partitions}->[0];
        my $messages = $partition_data->{MessageSet} = [];
        my $data = $_received_data{ $topic }->{ $partition } // [];
        my $HighwaterMarkOffset = $partition_data->{HighwaterMarkOffset} = scalar @{ $data };
        my $full_message_set_size = 0;
        for ( my $i = $FetchOffset; $i < $HighwaterMarkOffset; ++$i ) {
            my $Key     = $data->[ $i ]->[0];
            my $Value   = $data->[ $i ]->[1];
            my $message_set_size +=
                  8                         # [q]   Offset
                + 4                         # [l]   MessageSize
                + 4                         # [l]   Crc
                + 1                         # [c]   MagicByte
                + 1                         # [c]   Attributes
                + 4                         # [l]   Key length
                + length( $Key )            # Key
                + 4                         # [l]   Value length
                + length( $Value )          # Value
                ;
            if ( $full_message_set_size + $message_set_size <= $MaxBytes ) {
                push @{ $messages }, {
                    Offset      => $i,
                    MagicByte   => 0,
                    Attributes  => $COMPRESSION_NONE,
                    Key         => $Key,
                    Value       => $Value,
                };
            } else {
# NOTE: not all messages can be returned
                last;
            }
            $full_message_set_size += $message_set_size;
        }

        $self->_verify_string( $topic, "$description (TopicName)" )
            or return;
        my $max_response_length = $MaxBytes
            + 4                         # [l]   Size
            + 4                         # [l]   CorrelationId
            + 4                         # [l]   topics array size
            + 2                         # [s]   TopicName length
            + length( $topic )          # TopicName
            + 4                         # [l]   partitions array size
            + 4                         # [l]   Partition
            + 2                         # [s]   ErrorCode
            + 8                         # [q]   HighwaterMarkOffset
            + 4                         # [l]   MessageSetSize
            ;
        $encoded_response = encode_fetch_response( $decoded_fetch_response )
            or Kafka::IO::_error( $self, $ERROR_MISMATCH_ARGUMENT, $description );
        $encoded_response = substr( $encoded_response, 0, $max_response_length );
    }

    elsif ( $ApiKey == $APIKEY_OFFSET ) {
        my $decoded_offset_request = decode_offset_request( \$message )
            or Kafka::IO::_error( $self, $ERROR_MISMATCH_ARGUMENT, $description );
        my ( $topic, $partition ) = $self->_decoded_topic_partition( $decoded_offset_request, $decoded_offset_response );
        $partition // return;

        my $partition_data = $decoded_offset_request->{topics}->[0]->{partitions}->[0];
        my $MaxNumberOfOffsets = $partition_data->{MaxNumberOfOffsets};
        my $Time = $partition_data->{Time};

        my $offsets = $decoded_offset_response->{topics}->[0]->{PartitionOffsets}->[0]->{Offset} = [];
        if ( $Time == $RECEIVE_LATEST_OFFSETS) {
            push( @{ $offsets }, ( exists( $_received_data{ $topic }->{ $partition } )
                ? scalar( @{ $_received_data{ $topic }->{ $partition } } )
                : () ),
                0 );
        }
        elsif ( $Time == $RECEIVE_EARLIEST_OFFSET ) {
            push @{ $offsets }, 0;
        } else {
            if ( exists( $_received_data{ $topic }->{ $partition } ) ) {
                my $max_offset = min $MaxNumberOfOffsets, $#{ $_received_data{ $topic }->{ $partition } };
# NOTE:
# - always return starting at offset 0
# - not verified in practice, the order in which the kafka server returns the offsets
                push @$offsets, ( 0..$max_offset );
            }
        }

        $encoded_response = encode_offset_response( $decoded_offset_response )
            or Kafka::IO::_error( $self, $ERROR_MISMATCH_ARGUMENT, $description );
    }

    elsif ( $ApiKey == $APIKEY_METADATA ) {
        my $decoded_metadata_request = decode_metadata_request( \$message )
            or Kafka::IO::_error( $self, $ERROR_MISMATCH_ARGUMENT, $description );
        my $TopicName = $decoded_metadata_request->{topics}->[0] // $TOPIC;
        $self->_verify_string( $TopicName, "$description (TopicName)" )
            or return;
        $decoded_metadata_response->{TopicMetadata}->[0]->{TopicName} = $TopicName;
        $decoded_metadata_response->{CorrelationId} = $decoded_metadata_request->{CorrelationId};

        $encoded_response = encode_metadata_response( $decoded_metadata_response )
            or Kafka::IO::_error( $self, $ERROR_MISMATCH_ARGUMENT, $description );
    }
    elsif ($ApiKey == $APIKEY_SASLHANDSHAKE){
        my $decoded_saslhandshake_request = decode_saslhandshake_request(\$message);
        my $decoded_saslhandshake_response = {
            'CorrelationId' => 0,
            'Mechanisms' => [
                'PLAIN',
                'SCRAM-SHA-512'
            ],
            'ErrorCode' => 0
        };
        $decoded_saslhandshake_response->{CorrelationId} = $decoded_saslhandshake_request->{CorrelationId};
        if ($decoded_saslhandshake_request->{Mechanism} =~ /^SCRAM-(.*)/) {
            $SASL_SCRAM_STEP = 1;
            install_cred($1);
        } else {
            $decoded_saslhandshake_response->{ErrorCode} = 33; #UNSUPPORTED_SASL_MECHANISM
        }
        $encoded_response = encode_saslhandshake_response( $decoded_saslhandshake_response );
    }
    else {
        Kafka::IO::_error( $self, $ERROR_MISMATCH_ARGUMENT, "Unsupported API key in MockIO::send: $ApiKey" );
    }

    return $len;
}

=head3 C<receive>

Method emulation (C<Kafka::IO::receive>).

=cut
sub receive {
    my ( $self, $length ) = @_;

    _POSINT( $length )
        or Kafka::IO::_error( $self, $ERROR_MISMATCH_ARGUMENT, 'Kafka::IO->receive' );

    unless( defined $encoded_response ) {
        Kafka::IO::_error( $self, $ERROR_CANNOT_RECV, 'No response' );
    }

    my $message = substr( $encoded_response, 0, $length, '' );

    Kafka::IO::_debug_msg( $self, $message, 'Response from', 'yellow' )
        if Kafka::IO->debug_level == 1;

    return \$message;
}

=head3 C<close>

Method emulation (C<Kafka::IO::close>).

=cut
sub close {
    my ( $self ) = @_;

    delete $self->{$_} foreach keys %$self;

    return;
}

# Method emulation (C<Kafka::IO::_is_alive>).
sub _is_alive {
    my ( $self ) = @_;

    return !!$self->{socket};
}

#-- private attributes ---------------------------------------------------------

#-- private methods ------------------------------------------------------------

# Determines the type of request, the topic and partition
sub _decoded_topic_partition {
    my ( $self, $decoded_request, $decoded_response ) = @_;

    my $topic_data = $decoded_request->{topics}->[0];
    my $topic = $topic_data->{TopicName};

    my $partition = $topic_data->{partitions}->[0]->{Partition};
    $partition == $PARTITION
        or Kafka::IO::_error( $self, $ERROR_MISMATCH_ARGUMENT, "Use Kafka::MockIO only with partition = $PARTITION" );

    $topic_data = $decoded_response->{topics}->[0];
    $topic_data->{TopicName} = $topic;
    if ( $ApiKey == $APIKEY_OFFSET ) {
        $topic_data->{PartitionOffsets}->[0]->{Partition} = $partition;
    } else {
        $topic_data->{partitions}->[0]->{Partition} = $partition;
    }

    $decoded_response->{CorrelationId} = $decoded_request->{CorrelationId};

    return $topic, $partition;
}

# Verifies that the first argument is the string does not contain Unicode data
sub _verify_string {
    my ( $self, $string, $description ) = @_;

    return 1
        if defined( $string ) && $string eq q{};
    defined( _STRING( $string ) )
        or Kafka::IO::_error( $self, $ERROR_MISMATCH_ARGUMENT, $description );
    utf8::is_utf8( $string )
        and Kafka::IO::_error( $self, $ERROR_NOT_BINARY_STRING, $description );

    return 1;
}



1;

__END__

=head1 DIAGNOSTICS

Error diagnosis emulated methods corresponds to the work of class L<Kafka::IO|Kafka::IO/"DIAGNOSTICS">.

=head1 SEE ALSO

The basic operation of the Kafka package modules:

L<Kafka|Kafka> - constants and messages used by the Kafka package modules.

L<Kafka::Connection|Kafka::Connection> - interface to connect to a Kafka cluster.

L<Kafka::Producer|Kafka::Producer> - interface for producing client.

L<Kafka::Consumer|Kafka::Consumer> - interface for consuming client.

L<Kafka::Message|Kafka::Message> - interface to access Kafka message
properties.

L<Kafka::Int64|Kafka::Int64> - functions to work with 64 bit elements of the
protocol on 32 bit systems.

L<Kafka::Protocol|Kafka::Protocol> - functions to process messages in the
Apache Kafka's Protocol.

L<Kafka::IO|Kafka::IO> - low-level interface for communication with Kafka server.

L<Kafka::Exceptions|Kafka::Exceptions> - module designated to handle Kafka exceptions.

L<Kafka::Internals|Kafka::Internals> - internal constants and functions used
by several package modules.

A wealth of detail about the Apache Kafka and the Kafka Protocol:

Main page at L<http://kafka.apache.org/>

Kafka Protocol at L<https://cwiki.apache.org/confluence/display/KAFKA/A+Guide+To+The+Kafka+Protocol>

=head1 SOURCE CODE

Kafka package is hosted on GitHub:
L<https://github.com/TrackingSoft/Kafka>

=head1 AUTHOR

Sergey Gladkov

=head1 CONTRIBUTORS

Alexander Solovey

Jeremy Jordan

Sergiy Zuban

Vlad Marchenko

=head1 COPYRIGHT AND LICENSE

Copyright (C) 2012-2016 by TrackingSoft LLC.

This package is free software; you can redistribute it and/or modify it under
the same terms as Perl itself. See I<perlartistic> at
L<http://dev.perl.org/licenses/artistic.html>.

This program is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY;
without even the implied warranty of MERCHANTABILITY or FITNESS FOR A
PARTICULAR PURPOSE.

=cut