RMQ Headers Test

1_create_queues.pl

#!/usr/bin/env perl

use strict;
use warnings;
use JSON;

STDOUT->autoflush;

use Net::AMQP::RabbitMQ;

my $rmq = Net::AMQP::RabbitMQ->new;
$rmq->connect(
        "localhost", {
                user      => 'guest',
                password  => 'guest',
                heartbeat => 0,
                vhost     => 'test'
        });

$rmq->channel_open(1);

my $queue_name_any    = "my.transaction.any";
my $queue_name_normal = "my.transaction.normal";
my $queue_name_high   = "my.transaction.high";

$rmq->queue_declare(1, $queue_name_any   , { durable => 0, passive => 0, auto_delete => 0 }, {});
$rmq->queue_declare(1, $queue_name_normal, { durable => 0, passive => 0, auto_delete => 0 }, {});
$rmq->queue_declare(1, $queue_name_high  , { durable => 0, passive => 0, auto_delete => 0 }, {});

# I want all transactions
$rmq->queue_bind(1, $queue_name_any, 'amq.headers', "", {
        table    => 'transaction',
        }
);

# I want normal or low priority transactions
$rmq->queue_bind(1, $queue_name_normal, 'amq.headers', "", {
        priority => 'normal',
        table    => 'transaction',
        }
);

$rmq->queue_bind(1, $queue_name_normal, 'amq.headers', "", {
        priority => 'low',
        table    => 'transaction',
        }
);

# I want only high priority transactions
$rmq->queue_bind(1, $queue_name_high, 'amq.headers', "", {
        priority => 'high',
        table    => 'transaction',
        }
);

2_publish_messages.pl

#!/usr/bin/env perl

use strict;
use warnings;
use JSON;

STDOUT->autoflush;

use Net::AMQP::RabbitMQ;

my $rmq = Net::AMQP::RabbitMQ->new;
$rmq->connect(
        "localhost", {
                user      => 'guest',
                password  => 'guest',
                heartbeat => 0,
                vhost     => 'test',
        });

$rmq->channel_open(1);


my @messages = (
        {
                table => 'transaction',
                priority => 'low',
                body => 'test low',
                when => time,
        },
        {
                table => 'transaction',
                priority => 'high',
                body => 'test high',
                when => time,
        },
        {
                table => 'transaction',
                priority => 'normal',
                body => 'test normal',
                when => time,
        },
);


foreach my $record (@messages) {
        my $body = to_json($record, { canonical => 1 });

        $rmq->publish(1, "", $body, { exchange => 'amq.headers' }, {
                        headers => $record
                }
        );
}

3_consume_messages.pl

#!/usr/bin/env perl

use strict;
use warnings;

STDOUT->autoflush;

use Net::AMQP::RabbitMQ;

my $rmq = Net::AMQP::RabbitMQ->new;
$rmq->connect(
        "localhost", {
                user      => 'guest',
                password  => 'guest',
                heartbeat => 0,
                vhost     => 'test'
        });

$rmq->channel_open(1);

my @queues = (
        "my.transaction.any",
        "my.transaction.normal",
        "my.transaction.high",
);

foreach my $queue (sort @queues) {

        print "$queue\n";
        while (my $message = $rmq->get(1, $queue)) {
                print "  " . $message->{body} . "\n";
        }

        #sleep 1;
}