view release on metacpan or  search on metacpan

lib/Catalyst/Engine/Stomp.pm  view on Meta::CPAN

package Catalyst::Engine::Stomp;
use Moose;
use List::MoreUtils qw/ uniq /;
use HTTP::Request;
use Net::Stomp;
use MooseX::Types::Moose qw/Str Int HashRef/;
use namespace::autoclean;
use Encode;

extends 'Catalyst::Engine::Embeddable';

our $VERSION = '0.17';

has connection => (is => 'rw', isa => 'Net::Stomp');
has conn_desc => (is => 'rw', isa => Str);
has destination_namespace_map => (is => 'rw', isa => HashRef, default => sub { { } } );

=head1 NAME

Catalyst::Engine::Stomp - write message handling apps with Catalyst.

=head1 SYNOPSIS

  # In a server script:

  BEGIN {
    $ENV{CATALYST_ENGINE} = 'Stomp';
    require Catalyst::Engine::Stomp;
  }

  MyApp->config(
    Engine::Stomp' = {
       tries_per_server => 3,
      'servers' => [
       {
         'hostname' => 'localhost',
         'port' => '61613'
         connect_headers => {
           login => 'myuser',
           passcode => 'mypassword',
         },
       },
       {
         'hostname' => 'stomp.yourmachine.com',
         'port' => '61613'
       },
       ],
       utf8             => 1,
       subscribe_headers => {
         transformation       => 'jms-to-json',
       }
    },
  );
  MyApp->run();

  # In a controller, or controller base class:
  use base qw/ Catalyst::Controller::MessageDriven /;

  # then create actions, which map as message types
  sub testaction : Local {
      my ($self, $c) = @_;

      # Reply with a minimal response message
      my $response = { type => 'testaction_response' };

lib/Catalyst/Engine/Stomp.pm  view on Meta::CPAN

=head2 finalize_headers

Overridden to dump out any errors encountered, since you won't get a #'
"debugging" message as for HTTP.

=cut

sub finalize_headers {
    my ($self, $c) = @_;
    my $error = join "\n", @{$c->error};
    if ($error) {
        $c->log->debug($error);
    }
    return $self->next::method($c);
}

=head2 handle_stomp_frame

Dispatch according to Stomp frame type.

=cut

sub handle_stomp_frame {
    my ($self, $app, $frame) = @_;

    my $command = $frame->command();
    if ($command eq 'MESSAGE') {
        $self->handle_stomp_message($app, $frame);
    }
    elsif ($command eq 'ERROR') {
        $self->handle_stomp_error($app, $frame);
    }
    else {
        $app->log->debug("Got unknown Stomp command: $command");
    }
}

=head2 handle_stomp_message

Dispatch a Stomp message into the Catalyst app.

=cut

sub handle_stomp_message {
    my ($self, $app, $frame) = @_;

    # destination -> controller
    my $destination = $frame->headers->{destination};
    my $subscription = $frame->headers->{subscription};

    $app->log->debug("message from $destination ($subscription)")
        if $app->debug;

    my $controller = $self->destination_namespace_map->{"/subscription/$subscription"}
        || $self->destination_namespace_map->{$destination};

    # set up request
    my $config = $app->config->{'Engine::Stomp'};
    my $url = 'stomp://'.$config->{hostname}.':'.$config->{port}.'/'.$controller;
    my $request_headers = HTTP::Headers->new(%{$frame->headers});
    my $req = HTTP::Request->new(POST => $url, $request_headers);
    $req->content($frame->body);
    $req->content_length(length $frame->body);

    # dispatch
    my $response;
    $app->handle_request($req, \$response);

    # reply, if header set
    if (my $reply_to = $response->headers->header('X-Reply-Address')) {
        my $reply_queue = '/remote-temp-queue/' . $reply_to;
        my $content     = $response->content;

        if ($config->{utf8}) {
            $content = encode("utf8", $response->content); # create octets
        }

        my $reply_headers = $response->headers->clone;
        $reply_headers->remove_content_headers;
        my %reply_hh =
            map {
                lc($_), scalar($reply_headers->header($_)),
            }
            grep { !/^X-/i }
            $reply_headers->header_field_names();

        $self->connection->send({
            %reply_hh,
            destination => $reply_queue,
            body => $content
        });
    }

    # ack the message off the destination now we've replied / processed
    $self->connection->ack( { frame => $frame } );
}

=head2 handle_stomp_error

Log any Stomp error frames we receive.

=cut

sub handle_stomp_error {
    my ($self, $app, $frame) = @_;

    my $error = $frame->headers->{message};
    $app->log->debug("Got Stomp error: $error");
}

__PACKAGE__->meta->make_immutable;

=head1 CONFIGURATION

=head2 subscribe_header

Add additional header key/value pairs to the subscribe message sent to the
message broker.

=cut