package Kafka::Int64;

=head1 NAME

Kafka::Int64 - Functions to work with 64 bit elements of the
protocol on 32 bit systems.

=head1 VERSION

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


use 5.010;
use strict;
use warnings;

use bigint; # this allows integers of practially any size at the cost of significant performance drop

our $VERSION = 'v1.08';

use Exporter qw(

our @EXPORT_OK = qw(

use Carp;

use Kafka qw(
use Kafka::Exceptions;


    use 5.010;
    use strict;
    use warnings;

    use Scalar::Util qw(
    use Try::Tiny;

    use Kafka qw(

    try {

        # Apache Kafka Protocol: FetchOffset, Time

        my $offset = 123;

        my $encoded = $BITS64
            ? pack( 'q>', $offset )
            : Kafka::Int64::packq( $offset );

        my $response = chr( 0 ) x 8;

        $offset = $BITS64
            ? unpack( 'q>', substr( $response, 0, 8 ) )
            : Kafka::Int64::unpackq( substr( $response, 0, 8 ) );

        my $next_offset = $BITS64
            ? $offset + 1
            : Kafka::Int64::intsum( $offset, 1 );

    } catch {
        my $error = $_;
        if ( blessed( $error ) && $error->isa( 'Kafka::Exception' ) ) {
            warn 'Error: (', $error->code, ') ',  $error->message, "\n";
        } else {
            die $error;


This module is not intended to be used by end user.

In order to achieve better performance, functions of this module do not perform
validation of arguments.

Transparent L<BigInteger|bigint> support on 32-bit platforms where native
integer type is limited to 32 bits and slow bigint must be used instead.
Use L<functions|/FUNCTIONS> from this module in such case.

The main features of the C<Kafka::Int64> module are:

=over 3

=item *

Support for working with 64 bit elements of the Kafka protocol
on 32 bit systems.



#-- public functions -----------------------------------------------------------


The following functions are available for the C<Kafka::Int64> module.


=head3 C<intsum( $bint, $int )>

Adds two numbers to emulate bigint adding 64-bit integers in 32-bit systems.

The both arguments must be a number. That is, it is defined and Perl thinks
it's a number. Arguments may be a L<Math::BigInt|Math::BigInt>

Returns the value as a L<Math::BigInt|Math::BigInt> integer.

sub intsum {
    my ( $frst, $scnd ) = @_;

    my $ret = $frst + $scnd + 0;    # bigint coercion
    Kafka::Exception::Int64->throw( throw_args( $ERROR_MISMATCH_ARGUMENT, 'intsum' ) )
        if $ret->is_nan();

    return $ret;

=head3 C<packq( $bint )>

Emulates C<pack( q{qE<gt>}, $bint )> to 32-bit systems - assumes decimal string
or integer input.

An argument must be a positive number. That is, it is defined and Perl thinks
it's a number. The argument may be a L<Math::BigInt|Math::BigInt> integer.

The special values -1, -2 are allowed

Returns the value as a packed binary string.

sub packq {
    my ( $n ) = @_;

    if      ( $n == -1 )    { return pack q{C8}, ( 255 ) x 8; }
    elsif   ( $n == -2 )    { return pack q{C8}, ( 255 ) x 7, 254; }
    elsif   ( $n < 0 )      { Kafka::Exception::Int64->throw( throw_args( $ERROR_MISMATCH_ARGUMENT, 'packq' ) ); }

    return pack q{H16}, substr( '00000000000000000000000000000000'.substr( ( $n + 0 )->as_hex(), 2 ), -16 );

=head3 C<unpackq( $bstr )>

Emulates C<unpack( q{qE<gt>}, $bstr )> to 32-bit systems - assumes binary input.

The argument must be a binary string of 8 bytes length.

Returns the value as a L<Math::BigInt|Math::BigInt> integer.

sub unpackq {
    my ( $s ) = @_;

    my $ret = Math::BigInt->from_hex( '0x'.unpack( q{H16}, $s ) );
    $ret = -1 if $ret == 18446744073709551615;
    $ret = -2 if $ret == 18446744073709551614;

    return $ret;

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




When error is detected, an exception, represented by object of C<Kafka::Exception::Int64> class,
is thrown (see L<Kafka::Exceptions|Kafka::Exceptions>).

Any error L<functions|/FUNCTIONS> is FATAL.
FATAL errors will cause the program to halt (C<confess>), since the
problem is so severe that it would be dangerous to continue.

L<code|Kafka::Exceptions/code> and a more descriptive L<message|Kafka::Exceptions/message> provide
information about thrown exception. Consult documentation of the L<Kafka::Exceptions|Kafka::Exceptions>
for the list of all available methods.

Authors suggest using of L<Try::Tiny|Try::Tiny>'s C<try> and C<catch> to handle exceptions while
working with L<Kafka|Kafka> package.

=over 3

=item C<Invalid argument>

This means that you didn't give the right argument to some of the


=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

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<>

Kafka Protocol at L<>


Kafka package is hosted on GitHub:

=head1 AUTHOR

Sergey Gladkov

Please use GitHub project link above to report problems or contact authors.


Alexander Solovey

Jeremy Jordan

Sergiy Zuban

Vlad Marchenko


Copyright (C) 2012-2017 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

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