NAME

Net::RabbitMQ::Client - RabbitMQ client (XS for librabbitmq)

SYNOPSIS

Simple API:

use utf8;
use strict;

use Net::RabbitMQ::Client;

produce();
consume();

sub produce {
	my $simple = Net::RabbitMQ::Client->sm_new(
		host => "best.host.for.rabbitmq.net",
		login => "login", password => "password",
		exchange => "test_ex", exchange_type => "direct", exchange_declare => 1,
		queue => "test_queue", queue_declare => 1
	);
	die sm_get_error_desc($simple) unless ref $simple;
	
	my $sm_status = $simple->sm_publish('{"say": "hello"}', content_type => "application/json");
	die sm_get_error_desc($sm_status) if $sm_status;
	
	$simple->sm_destroy();
}

sub consume {
	my $simple = Net::RabbitMQ::Client->sm_new(
		host => "best.host.for.rabbitmq.net",
		login => "login", password => "password",
		queue => "test_queue"
	);
	die sm_get_error_desc($simple) unless ref $simple;
	
	my $sm_status = $simple->sm_get_messages(sub {
		my ($self, $message) = @_;
		
		print $message, "\n";
		
		1; # it is important to return 1 (send ask) or 0
	});
	die sm_get_error_desc($sm_status) if $sm_status;
	
	$simple->sm_destroy();
}

Base API:

use utf8;
use strict;

use Net::RabbitMQ::Client;

produce();
consume();

sub produce {
	my $rmq = Net::RabbitMQ::Client->create();
	
	my $exchange = "test";
	my $routingkey = "";
	my $messagebody = "lalala";
	
	my $channel = 1;
	
	my $conn = $rmq->new_connection();
	my $socket = $rmq->tcp_socket_new($conn);
	
	my $status = $rmq->socket_open($socket, "best.host.for.rabbitmq.net", 5672);
	die "Can't create socket" if $status;
	
	$status = $rmq->login($conn, "/", 0, 131072, 0, AMQP_SASL_METHOD_PLAIN, "login", "password");
	die "Can't login on server" if $status != AMQP_RESPONSE_NORMAL;
	
	$status = $rmq->channel_open($conn, $channel);
	die "Can't open chanel" if $status != AMQP_RESPONSE_NORMAL;
    
	$rmq->queue_bind($conn, 1, "test_q", $exchange, $routingkey, 0);
	die "Can't bind queue" if $status != AMQP_RESPONSE_NORMAL;
	
	my $props = $rmq->type_create_basic_properties();
	$rmq->set_prop__flags($props, AMQP_BASIC_CONTENT_TYPE_FLAG|AMQP_BASIC_DELIVERY_MODE_FLAG);
	$rmq->set_prop_content_type($props, "text/plain");
	$rmq->set_prop_delivery_mode($props, AMQP_DELIVERY_PERSISTENT);
	
	$status = $rmq->basic_publish($conn, $channel, $exchange, $routingkey, 0, 0, $props, $messagebody);
	
	if($status != AMQP_STATUS_OK) {
	    print "Can't send message\n";
	}
	
	$rmq->type_destroy_basic_properties($props);
	
	$rmq->channel_close($conn, 1, AMQP_REPLY_SUCCESS);
	$rmq->connection_close($conn, AMQP_REPLY_SUCCESS);
	$rmq->destroy_connection($conn);
}

sub consume {
	my $rmq = Net::RabbitMQ::Client->create();
	
	my $exchange = "test";
	my $routingkey = "";
	my $messagebody = "lalala";
	
	my $channel = 1;
	
	my $conn = $rmq->new_connection();
	my $socket = $rmq->tcp_socket_new($conn);
	
	my $status = $rmq->socket_open($socket, "best.host.for.rabbitmq.net", 5672);
	die "Can't create socket" if $status;
	
	$status = $rmq->login($conn, "/", 0, 131072, 0, AMQP_SASL_METHOD_PLAIN, "login", "password");
	die "Can't login on server" if $status != AMQP_RESPONSE_NORMAL;
	
	$status = $rmq->channel_open($conn, $channel);
	die "Can't open chanel" if $status != AMQP_RESPONSE_NORMAL;
	
	$status = $rmq->basic_consume($conn, 1, "test_q", undef, 0, 1, 0);
	die "Consuming" if $status != AMQP_RESPONSE_NORMAL;
	
	my $envelope = $rmq->type_create_envelope();
	
	while (1)
	{
		$rmq->maybe_release_buffers($conn);
		
		$status = $rmq->consume_message($conn, $envelope, 0, 0);
		last if $status != AMQP_RESPONSE_NORMAL;
		
		print "New message: \n", $rmq->envelope_get_message_body($envelope), "\n";
		
		$rmq->destroy_envelope($envelope);
	}
	
	$rmq->type_destroy_envelope($envelope);
	
	$rmq->channel_close($conn, 1, AMQP_REPLY_SUCCESS);
	$rmq->connection_close($conn, AMQP_REPLY_SUCCESS);
	$rmq->destroy_connection($conn);
}

DESCRIPTION

This is binding for RabbitMQ-C library.

Please, before install this module make RabbitMQ-C library.

See https://github.com/alanxz/rabbitmq-c

https://github.com/lexborisov/perl-net-rabbitmq-client

METHODS

Simple API

sm_new

my $simple = Net::RabbitMQ::Client->sm_new(
	host => '',              # default: 
	port => 5672,            # default: 5672
	channel => 1,            # default: 1
	login => '',             # default: 
	password => '',          # default: 
	exchange => undef,       # default: undef
	exchange_type => undef,  # optional, if exchange_declare == 1; types: topic, fanout, direct, headers; default: undef
	exchange_declare => 0,   # declare exchange or not; 1 or 0; default: 0
	queue => undef,          # default: undef
	routingkey => '',        # default: 
	queue_declare => 0,      # declare queue or not; 1 or 0; default: 0
);

Return: a Simple object if successful, otherwise an error occurred

sm_publish

my $sm_status = $simple->sm_publish($text,
	# default args
	content_type  => "text/plain",
	delivery_mode => AMQP_DELIVERY_PERSISTENT,
	_flags        => AMQP_BASIC_CONTENT_TYPE_FLAG|AMQP_BASIC_DELIVERY_MODE_FLAG
);

Return: 0 if successful, otherwise an error occurred

sm_get_messages

Loop to get messages

my $callback = {
	my ($simple, $message) = @_;
	
	1; # it is important to return 1 (send ask) or 0
}

my $sm_status = $simple->sm_get_messages($callback);

Return: 0 if successful, otherwise an error occurred

sm_get_message

Get one message

my $sm_status = 0;
my $message = $simple->sm_get_message($sm_status);

Return: message if successful

sm_get_rabbitmq

my $rmq = $simple->sm_get_rabbitmq();

Return: RabbitMQ Base object

sm_get_connection

my $conn = $simple->sm_get_connection();

Return: Connection object (from Base API new_connection)

sm_get_socket

my $socket = $simple->sm_get_socket();

Return: Socket object (from Base API tcp_socket_new)

sm_get_config

my $config = $simple->sm_get_config();

Return: Config when creating a Simple object

sm_get_config

my $description = $simple->sm_get_error_desc($sm_error_code);

Return: Error description by Simple error code

sm_destroy

Destroy a Simple object

$simple->sm_destroy();

Base API

Connection and Authorization

create

my $rmq = Net::RabbitMQ::Client->create();

Return: rmq

new_connection

my $amqp_connection_state_t = $rmq->new_connection();

Return: amqp_connection_state_t

tcp_socket_new

my $amqp_socket_t = $rmq->tcp_socket_new($conn);

Return: amqp_socket_t

socket_open

my $status = $rmq->socket_open($socket, $host, $port);

Return: status

socket_open_noblock

my $status = $rmq->socket_open_noblock($socket, $host, $port, $struct_timeout);

Return: status

my $status = $rmq->login($conn, $vhost, $channel_max, $frame_max, $heartbeat, $sasl_method);

Return: status

channel_open

my $status = $rmq->channel_open($conn, $channel);

Return: status

socket_get_sockfd

my $res = $rmq->socket_get_sockfd($socket);

Return: variable

get_socket

my $amqp_socket_t  = $rmq->get_socket($conn);

Return: amqp_socket_t

channel_close

my $status = $rmq->channel_close($conn, $channel, $code);

Return: status

connection_close

my $status = $rmq->connection_close($conn, $code);

Return: status

destroy_connection

my $status = $rmq->destroy_connection($conn);

Return: status

SSL

ssl_socket_new

my $amqp_socket_t = $rmq->ssl_socket_new($conn);

Return: amqp_socket_t

ssl_socket_set_key

my $status = $rmq->ssl_socket_set_key($socket, $cert, $key);

Return: status

set_initialize_ssl_library

$rmq->set_initialize_ssl_library($do_initialize);

ssl_socket_set_cacert

my $status = $rmq->ssl_socket_set_cacert($socket, $cacert);

Return: status

ssl_socket_set_key_buffer

my $status = $rmq->ssl_socket_set_key_buffer($socket, $cert, $key, $n);

Return: status

ssl_socket_set_verify

$rmq->ssl_socket_set_verify($socket, $verify);

Basic Publish/Consume

basic_publish

my $status = $rmq->basic_publish($conn, $channel, $exchange, $routing_key, $mandatory, $immediate, $properties, $body);

Return: status

basic_consume

my $status = $rmq->basic_consume($conn, $channel, $queue, $consumer_tag, $no_local, $no_ack, $exclusive);

Return: status

basic_get

my $status = $rmq->basic_get($conn, $channel, $queue, $no_ack);

Return: status

basic_ack

my $status = $rmq->basic_ack($conn, $channel, $delivery_tag, $multiple);

Return: status

basic_nack

my $status = $rmq->basic_nack($conn, $channel, $delivery_tag, $multiple, $requeue);

Return: status

basic_reject

my $status = $rmq->basic_reject($conn, $channel, $delivery_tag, $requeue);

Return: status

Consume

consume_message

my $status = $rmq->consume_message($conn, $envelope, $struct_timeout, $flags);

Return: status

Queue

queue_declare

my $status = $rmq->queue_declare($conn, $channel, $queue, $passive, $durable, $exclusive, $auto_delete);

Return: status

queue_bind

my $status = $rmq->queue_bind($conn, $channel, $queue, $exchange, $routing_key);

Return: status

queue_unbind

my $status = $rmq->queue_unbind($conn, $channel, $queue, $exchange, $routing_key);

Return: status

Exchange

exchange_declare

my $status = $rmq->exchange_declare($conn, $channel, $exchange, $type, $passive, $durable, $auto_delete, $internal);

Return: status

Envelope

envelope_get_redelivered

my $res = $rmq->envelope_get_redelivered($envelope);

Return: variable

envelope_get_channel

my $res = $rmq->envelope_get_channel($envelope);

Return: variable

envelope_get_exchange

my $res = $rmq->envelope_get_exchange($envelope);

Return: variable

envelope_get_routing_key

my $res = $rmq->envelope_get_routing_key($envelope);

Return: variable

destroy_envelope

$rmq->destroy_envelope($envelope);

envelope_get_consumer_tag

my $res = $rmq->envelope_get_consumer_tag($envelope);

Return: variable

envelope_get_delivery_tag

my $res = $rmq->envelope_get_delivery_tag($envelope);

Return: variable

envelope_get_message_body

my $res = $rmq->envelope_get_message_body($envelope);

Return: variable

Types

type_create_envelope

my $amqp_envelope_t = $rmq->type_create_envelope();

Return: amqp_envelope_t

type_destroy_envelope

$rmq->type_destroy_envelope($envelope);

type_create_timeout

my $struct_timeval = $rmq->type_create_timeout($timeout_sec);

Return: struct_timeval

type_destroy_timeout

$rmq->type_destroy_timeout($timeout);

type_create_basic_properties

my $amqp_basic_properties_t = $rmq->type_create_basic_properties();

Return: amqp_basic_properties_t

type_destroy_basic_properties

my $status = $rmq->type_destroy_basic_properties($props);

Return: status

For a Basic Properties

set_prop_app_id

$rmq->set_prop_app_id($props, $value);

set_prop_content_type

$rmq->set_prop_content_type($props, $value);

set_prop_reply_to

$rmq->set_prop_reply_to($props, $value);

set_prop_priority

$rmq->set_prop_priority($props, $priority);

set_prop__flags

$rmq->set_prop__flags($props, $flags);

set_prop_user_id

$rmq->set_prop_user_id($props, $value);

set_prop_delivery_mode

$rmq->set_prop_delivery_mode($props, $delivery_mode);

set_prop_message_id

$rmq->set_prop_message_id($props, $value);

set_prop_timestamp

$rmq->set_prop_timestamp($props, $timestamp);

set_prop_cluster_id

$rmq->set_prop_cluster_id($props, $value);

set_prop_correlation_id

$rmq->set_prop_correlation_id($props, $value);

set_prop_expiration

$rmq->set_prop_expiration($props, $value);

set_prop_type

$rmq->set_prop_type($props, $value);

set_prop_content_encoding

$rmq->set_prop_content_encoding($props, $value);

Other

data_in_buffer

my $amqp_boolean_t = $rmq->data_in_buffer($conn);

Return: amqp_boolean_t

maybe_release_buffers

$rmq->maybe_release_buffers($conn);

error_string

my $res = $rmq->error_string($error);

Return: variable

DESTROY

undef $rmq;

Free mem and destroy object.

AUTHOR

Alexander Borisov <lex.borisov@gmail.com>

COPYRIGHT AND LICENSE

This is free software; you can redistribute it and/or modify it under the same terms as the Perl 5 programming language system itself.

See librabbitmq license and COPYRIGHT https://github.com/alanxz/rabbitmq-c

To install Net::RabbitMQ::Client, copy and paste the appropriate command in to your terminal.

cpanm

cpanm Net::RabbitMQ::Client

CPAN shell

perl -MCPAN -e shell
install Net::RabbitMQ::Client

For more information on module installation, please visit the detailed CPAN module installation guide.

	Global
`s`	Focus search bar
`?`	Bring up this help dialog

	GitHub
`g` `p`	Go to pull requests
`g` `i`	go to github issues (only if github is preferred repository)

	POD
`g` `a`	Go to author
`g` `c`	Go to changes
`g` `i`	Go to issues
`g` `d`	Go to dist
`g` `r`	Go to repository/SCM
`g` `s`	Go to source
`g` `b`	Go to file browse

	Search terms
module: (e.g. module:Plugin)
distribution: (e.g. distribution:Dancer auth)
author: (e.g. author:SONGMU Redis)
version: (e.g. version:1.00)