X-Git-Url: http://git.shadowcat.co.uk/gitweb/gitweb.cgi?a=blobdiff_plain;f=lib%2FCatalyst%2FEngine%2FStomp.pm;h=2752b5e2da992a69fc4cd456b950124ca003642f;hb=b9aa486179cff4fdc2e83ca34c39b83f4c5a246f;hp=b469d83984435a001e2d9a0e52ed023028a963a6;hpb=bf8937b70eb1aabe53e63e2e6c70f569056e609a;p=catagits%2FCatalyst-Engine-STOMP.git diff --git a/lib/Catalyst/Engine/Stomp.pm b/lib/Catalyst/Engine/Stomp.pm index b469d83..2752b5e 100644 --- a/lib/Catalyst/Engine/Stomp.pm +++ b/lib/Catalyst/Engine/Stomp.pm @@ -1,12 +1,13 @@ package Catalyst::Engine::Stomp; use Moose; -extends 'Catalyst::Engine::Embeddable'; - -our $VERSION = '0.03'; - use List::MoreUtils qw/ uniq /; use HTTP::Request; use Net::Stomp; +use namespace::autoclean; + +extends 'Catalyst::Engine::Embeddable'; + +our $VERSION = '0.05'; has connection => (is => 'rw', isa => 'Net::Stomp'); has conn_desc => (is => 'rw', isa => 'Str'); @@ -22,7 +23,7 @@ Catalyst::Engine::Stomp - write message handling apps with Catalyst. BEGIN { $ENV{CATALYST_ENGINE} = 'Stomp'; require Catalyst::Engine::Stomp; - } + } MyApp->config->{Engine::Stomp} = { @@ -46,7 +47,7 @@ Catalyst::Engine::Stomp - write message handling apps with Catalyst. =head1 DESCRIPTION Write a Catalyst app connected to a Stomp messagebroker, not HTTP. You -need a controller that understands messaging, as well as this engine. +need a controller that understands messaging, as well as this engine. This is single-threaded and single process - you need to run multiple instances of this engine to get concurrency, and configure your broker @@ -54,8 +55,8 @@ to load-balance across multiple consumers of the same queue. Controllers are mapped to Stomp queues, and a controller base class is provided, Catalyst::Controller::MessageDriven, which implements -YAML-serialized messages, mapping a top-level YAML "type" key to -the action. +YAML-serialized messages, mapping a top-level YAML "type" key to +the action. =head1 METHODS @@ -71,19 +72,8 @@ sub run { die 'No Engine::Stomp configuration found' unless ref $app->config->{'Engine::Stomp'} eq 'HASH'; - # list the path namespaces that will be mapped as queues. - # - # this is known to use the deprecated - # Dispatcher->action_hash() method, but there doesn't appear - # to be another way to get the relevant strings out. - # - # http://github.com/rafl/catalyst-runtime/commit/5de163f4963d9dbb41d7311ca6f17314091b7af3#L2R644 - # - my @queues = - uniq - grep { length $_ } - map { $_->namespace } - values %{$app->dispatcher->action_hash}; + my @queues = grep { length $_ } + map { $app->controller($_)->action_namespace } $app->controllers; # connect up my %template = %{$app->config->{'Engine::Stomp'}}; @@ -95,8 +85,8 @@ sub run { foreach my $queue (@queues) { my $queue_name = "/queue/$queue"; $self->connection->subscribe({ - destination => $queue_name, - ack => 'client' + destination => $queue_name, + ack => 'client' }); } @@ -117,10 +107,10 @@ client IP address. =cut sub prepare_request { - my ($self, $c, $req, $res_ref) = @_; - shift @_; - $self->next::method(@_); - $c->req->address($self->conn_desc); + my ($self, $c, $req, $res_ref) = @_; + shift @_; + $self->next::method(@_); + $c->req->address($self->conn_desc); } =head2 finalize_headers @@ -131,12 +121,12 @@ Overridden to dump out any errors encountered, since you won't get a =cut sub finalize_headers { - my ($self, $c) = @_; - my $error = join "\n", @{$c->error}; - if ($error) { - $c->log->debug($error); - } - return $self->next::method($c); + my ($self, $c) = @_; + my $error = join "\n", @{$c->error}; + if ($error) { + $c->log->debug($error); + } + return $self->next::method($c); } =head2 handle_stomp_frame @@ -146,18 +136,18 @@ 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"); - } + 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 @@ -167,31 +157,31 @@ Dispatch a Stomp message into the Catalyst app. =cut sub handle_stomp_message { - my ($self, $app, $frame) = @_; - - # queue -> controller - my $queue = $frame->headers->{destination}; - my ($controller) = $queue =~ m!^/queue/(.*)$!; - - # set up request - my $config = $app->config->{'Engine::Stomp'}; - my $url = 'stomp://'.$config->{hostname}.':'.$config->{port}.'/'.$controller; - my $req = HTTP::Request->new(POST => $url); - $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; - $self->connection->send({ destination => $reply_queue, body => $response->content }); - } - - # ack the message off the queue now we've replied / processed - $self->connection->ack( { frame => $frame } ); + my ($self, $app, $frame) = @_; + + # queue -> controller + my $queue = $frame->headers->{destination}; + my ($controller) = $queue =~ m|^/queue/(.*)$|; + + # set up request + my $config = $app->config->{'Engine::Stomp'}; + my $url = 'stomp://'.$config->{hostname}.':'.$config->{port}.'/'.$controller; + my $req = HTTP::Request->new(POST => $url); + $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; + $self->connection->send({ destination => $reply_queue, body => $response->content }); + } + + # ack the message off the queue now we've replied / processed + $self->connection->ack( { frame => $frame } ); } =head2 handle_stomp_error @@ -201,13 +191,11 @@ 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"); + my ($self, $app, $frame) = @_; + + my $error = $frame->headers->{message}; + $app->log->debug("Got Stomp error: $error"); } __PACKAGE__->meta->make_immutable; -1; -