File Coverage

blib/lib/Net/RabbitMQ/Java.pm

Criterion	Covered	Total	%
statement	13	15	86.6
branch			n/a
condition			n/a
subroutine	5	5	100.0
pod			n/a
total	18	20	90.0

package Net::RabbitMQ::Java;

use strict;
use warnings;

our $VERSION = '2.030102';

use Data::UUID;
use File::ShareDir qw(dist_dir);
use Inline::Java qw(cast);

my %callbacks = ();       # callback_id => [ CallbackCaller, sub {} ]
my %obj_callbacks = ();   # object_ref => [ callback_id, ... ]
my ($Helper);

# callback helpers
sub _callback {
    my ($callback_id, $args) = @_;
    $callbacks{$callback_id}->[1]->(@$args);
}
sub _callback_error {
    print STDERR shift, "\n";
    exit;
}
sub processCallbacks { # class method
    my $class = shift;
    $_->[0]->process for values %callbacks;
}

# prototypes for method overloading
sub method_with_3_args_or_map {
    my $orig = shift;

    # if we were called with more than 3 arguments,
    # last one can be a hash
    splice @_, -1, 1, encode_Map($_[-1]) if @_ > 1+3;

    return $orig->(@_);
}
sub decoding_accessor {
    my $decoder = shift;
    return sub {
        my $orig = shift;
        return $decoder->($orig->(@_));
    };
}
sub encoding_setter {
    my $encoder = shift;
    return sub {
        my $orig = shift;
        return $orig->($encoder->(@_));
    };
}
sub callback_setter {
    my ($type, $decode_coderef) = @_;
    $decode_coderef ||= sub {@_}; # second arg is optional
    return sub {
        my ($orig, $obj, $coderef) = @_;

        # generate the callback unique identifier
        my $callback_id = "${type}_${obj}_" . Data::UUID->new->create_hex;

        # create a Java helper listener
        my $listener_obj = "Net::RabbitMQ::Java::Helper::$type"->new($Helper, $callback_id);
        $orig->($obj, $listener_obj);
        my $callbackCaller = $listener_obj->getCallbackCaller;

        # save the callback in Perl-land
        $coderef ||= sub {};
        $callbacks{$callback_id} = [
            $callbackCaller,
            sub { $coderef->($decode_coderef->(@_)) }
        ];

        # save an index of callbacks added to channels, so that we
        # can remove them when channel objects get destroyed
        $obj_callbacks{"$obj"} ||= [];
        push @{ $obj_callbacks{"$obj"} }, $callback_id;

        return $callbackCaller;
    };
}
sub destroy_callbacks {
    my $self = shift;
    if ($obj_callbacks{"$self"}) {
        delete $callbacks{$_} for @{ $obj_callbacks{"$self"} };
        delete $obj_callbacks{"$self"};
    }
    return Inline::Java::Object::DESTROY($self, @_);
}

# encoding/decoding subroutines
sub encode_ByteArray ($) {
    my $string = shift;
    # this is very inefficient; an idea would be to
    # convert to hexadecimal (base64?) and pass it as a String
    # to a Java helper object to call .getBytes() on
    return [ unpack("c*", $string) ];
}

sub decode_ByteArray ($) {
    my $byteArray = shift;
    return pack("c*", @$byteArray);
}

sub encode_Map ($) {
    my $hash = shift;
    return undef unless $hash;
    return $hash if ref $hash =~ /java::util::HashMap$/;
    my $map_obj = new java::util::HashMap;
    foreach my $key (keys %$hash) {
        $map_obj->put($key, $hash->{$key});
    }
    return $map_obj;
}

sub decode_Map ($) {
    my $map_obj = shift;
    my $hash = {};
    return $hash unless $map_obj;
    my $it = $map_obj->entrySet->iterator;
    while ($it->hasNext) {
        my $entry_obj = cast('java.util.Map$Entry', $it->next);
        # getValue returns a com.rabbitmq.client.impl.LongStringHelper.ByteArrayLongString
        $hash->{ $entry_obj->getKey } = $entry_obj->getValue->toString;
    }
    return $hash;
}

sub encode_Date ($) {
    my $dt = shift;
    return undef unless $dt;
    return $dt if ref $dt =~ /java::util::Date$/;
    return new java::util::Date(ref $dt eq 'DateTime' ? $dt->epoch : $dt);
}

sub encode_BasicProperties ($) {
    return undef unless $_[0];
    return $_[0] if ref $_[0] =~ /AMQP::BasicProperties$/;
    my %args = %{$_[0]};

    $args{headers}   = encode_Map $args{headers};
    $args{timestamp} = encode_Date $args{timestamp};

    my $props_obj = Net::RabbitMQ::Java::Client::AMQP::BasicProperties->new(
        map delete $args{$_},
        qw(contentType contentEncoding headers deliveryMode priority
        correlationId replyTo expiration messageId timestamp type
        userId appId clusterId)
    );
    !%args or die "Unknown properties: " . join(', ', keys %args);
    return $props_obj;
}

# main code
my $inited = 0;
sub init {
    my ($class, %params) = @_;
    return if $inited;
    $inited = 1;

    # load Java code
    my $share_dir = dist_dir('Net-RabbitMQ-Java');
    my $helper_code;
    {
        # TODO: we should pre-compile Helper.java
        # borrowing code from Inline-Java/Makefile.PL
        # (or maybe Inline::Java itself provides helper methods?)
        local $/;
        open(my $fh, '<', "$share_dir/java/Helper.java") or die;
        $helper_code = <$fh>;
        close $fh;
    }

    $params{CLASSPATH} ||= '';
    $params{CLASSPATH} .= ":$share_dir/java/rabbitmq-client.jar:$share_dir/java/commons-io-1.2.jar";
    Inline->bind(
        Java       => $helper_code,
        %params,
        AUTOSTUDY  => 1,
        STUDY      => [qw(
            Helper
            java.util.Date
            java.util.HashMap
            com.rabbitmq.client.ConnectionFactory
            com.rabbitmq.client.AMQP$BasicProperties
            com.rabbitmq.client.impl.AMQConnection
            com.rabbitmq.client.impl.ChannelN
            com.rabbitmq.client.GetResponse
            com.rabbitmq.client.QueueingConsumer
            com.rabbitmq.client.QueueingConsumer$Delivery
        )],
    );

    # alias our namespaces
    $Helper = Net::RabbitMQ::Java::Helper->new;
    *java:: = *Net::RabbitMQ::Java::java::;
    *Net::RabbitMQ::Java::Client:: = *Net::RabbitMQ::Java::com::rabbitmq::client::;

    # override methods that need to be more Perl-friendly
    my %override_subs = (
        'impl::ChannelN::basicPublish' => sub {
            my $orig = shift;

            # last argument is message body
            splice @_, -1, 1, encode_ByteArray $_[-1];

            # next-to-last argument is a basic properties hash
            splice @_, -2, 1, encode_BasicProperties $_[-2];

            return $orig->(@_);
        },
        'impl::ChannelN::basicConsume' => sub {
            my $orig = shift;

            # if called with 7 arguments, 6th is a map
            splice @_, -2, 1, encode_Map $_[-2] if @_ == 1+7;

            return $orig->(@_);
        },
        'impl::ChannelN::exchangeBind'      => \&method_with_3_args_or_map,
        'impl::ChannelN::exchangeDeclare'   => \&method_with_3_args_or_map,
        'impl::ChannelN::exchangeUnbind'    => \&method_with_3_args_or_map,
        'impl::ChannelN::queueBind'         => \&method_with_3_args_or_map,
        'impl::ChannelN::queueDeclare'      => \&method_with_3_args_or_map,
        'impl::ChannelN::queueUnbind'       => \&method_with_3_args_or_map,

        'impl::ChannelN::setReturnListener' => callback_setter('ReturnListener', sub {
            # last argument is message body
            splice @_, -1, 1, decode_ByteArray $_[-1];
            @_;
        }),
        'impl::ChannelN::setConfirmListener'        => callback_setter('ConfirmListener'),
        'impl::ChannelN::setFlowListener'           => callback_setter('FlowListener'),
        'impl::ChannelN::addShutdownListener'       => callback_setter('ShutdownListener'),
        'impl::AMQConnection::addShutdownListener'  => callback_setter('ShutdownListener'),

        'ConnectionFactory::getClientProperties'    => decoding_accessor(\&decode_Map),
        'impl::AMQConnection::getClientProperties'  => decoding_accessor(\&decode_Map),
        'impl::AMQConnection::getServerProperties'  => decoding_accessor(\&decode_Map),
        'QueueingConsumer::Delivery::getBody'       => decoding_accessor(\&decode_ByteArray),
        'GetResponse::getBody'                      => decoding_accessor(\&decode_ByteArray),
        'AMQP::BasicProperties::getHeaders'         => decoding_accessor(\&decode_Map),

        'ConnectionFactory::setClientProperties'    => encoding_setter(\&encode_Map),
    );
    my %new_subs = (
        'impl::ChannelN::DESTROY'       => \&destroy_callbacks,
        'impl::AMQConnection::DESTROY'  => \&destroy_callbacks,
    );
    {
        no strict 'refs';
        no warnings 'redefine';
        foreach my $sub (keys %override_subs) {
            my $fullname = "Net::RabbitMQ::Java::Client::$sub";
            my $orig = *$fullname{CODE} or die "failed to override $fullname";

            *{ $fullname } = sub {
                return $override_subs{$sub}->($orig, @_);
            };
        }
        *{ "Net::RabbitMQ::Java::Client::$_" } = $new_subs{$_}
            for keys %new_subs;
    }
}


1;
__END__