[SCM] libmessage-passing-perl Debian packaging branch, master, updated. debian/0.111-3-14-g44f6e88

[Tomas Doran]

The following commit has been merged in the master branch:
commit bccf044b1bb279b65da15aff3432eb39baaa01c6
Author: Tomas Doran <bobtfish at bobtfish.net>
Date:   Wed Jul 25 00:43:46 2012 +0100

    UDP Sockets

diff --git a/Makefile.PL b/Makefile.PL
index e89f043..ac33764 100644
--- a/Makefile.PL
+++ b/Makefile.PL
@@ -18,6 +18,7 @@ requires 'Package::Variant' => '1.001001';
 requires 'namespace::clean' => '0.23';
 requires 'Module::Runtime' => '0.013';
 requires 'AnyEvent';
+requires 'AnyEvent::Handle::UDP';
 requires 'Config::Any';
 requires 'MooX::Types::MooseLike' => '0.08';
 requires 'MooX::Options' => '1.6';
diff --git a/lib/Message/Passing/Input/Socket/UDP.pm b/lib/Message/Passing/Input/Socket/UDP.pm
new file mode 100644
index 0000000..6734548
--- /dev/null
+++ b/lib/Message/Passing/Input/Socket/UDP.pm
@@ -0,0 +1,53 @@
+package Message::Passing::Input::Socket::UDP;
+use Moo;
+use AnyEvent;
+use AnyEvent::Handle::UDP;
+use Scalar::Util qw/ weaken /;
+use namespace::clean -except => 'meta';
+
+with qw/
+    Message::Passing::Role::Input
+    Message::Passing::Role::HasHostnameAndPort
+    Message::Passing::Role::HasErrorChain
+/;
+
+has '+hostname' => (
+    default => sub { 'localhost' },
+);
+
+has '+port' => (
+    required => 1,
+);
+
+sub _default_port { die "You must supply a port #" }
+
+has handle => (
+    is => 'ro',
+    builder => '_build_handle',
+    lazy => 1,
+);
+
+sub BUILD {
+    my $self = shift;
+    $self->handle;
+}
+
+sub _build_handle {
+    my $self = shift;
+    weaken($self);
+    AnyEvent::Handle::UDP->new(
+        bind => [ $self->hostname, $self->port ],
+        on_recv => sub {
+            my ($data, $h, $from_addr) = @_;
+            # The output can optionally drop from addr.
+            $self->output_to->consume($data, $from_addr);
+        },
+        on_error => sub {
+            my ($h, $fatal, $msg) = @_;
+            $self->error->consume($msg);
+        },
+    );
+}
+
+1;
+
diff --git a/lib/Message/Passing/Output/Socket/UDP.pm b/lib/Message/Passing/Output/Socket/UDP.pm
new file mode 100644
index 0000000..4302560
--- /dev/null
+++ b/lib/Message/Passing/Output/Socket/UDP.pm
@@ -0,0 +1,44 @@
+package Message::Passing::Output::Socket::UDP;
+use Moo;
+use IO::Socket::INET;
+use namespace::clean -except => 'meta';
+
+with qw/
+    Message::Passing::Role::Output
+    Message::Passing::Role::HasHostnameAndPort
+    Message::Passing::Role::HasErrorChain
+/;
+
+has '+port' => (
+    required => 1,
+);
+
+sub _default_port { die "You must supply a port #" }
+
+has handle => (
+    is => 'ro',
+    builder => '_build_handle',
+    lazy => 1,
+);
+
+sub BUILD {
+    my $self = shift;
+    $self->handle;
+}
+
+sub _build_handle {
+    my $self = shift;
+    IO::Socket::INET->new(
+        Proto    => 'udp',
+        PeerAddr => $self->hostname,
+        PeerPort => $self->port,
+    ) or die "Could not create UDP socket: $!\n";
+}
+
+sub consume {
+    my ($self, $msg) = @_;
+    $self->handle->send($msg);
+}
+
+1;
+
diff --git a/t/00_compile.t b/t/00_compile.t
index 9c34ee5..e09a3b8 100644
--- a/t/00_compile.t
+++ b/t/00_compile.t
@@ -25,6 +25,7 @@ use_ok('Message::Passing::Filter::Decoder::Null');
 use_ok('Message::Passing::Role::HasHostnameAndPort');
 use_ok('Message::Passing::Role::HasUsernameAndPassword');
 use_ok('Message::Passing::Role::HasErrorChain');
+use_ok('Message::Passing::Input::Socket::UDP');
 
 done_testing;
 
diff --git a/t/socket_udp.t b/t/socket_udp.t
new file mode 100644
index 0000000..c721cdf
--- /dev/null
+++ b/t/socket_udp.t
@@ -0,0 +1,57 @@
+use strict;
+use warnings;
+no warnings 'once';
+use Test::More;
+
+use AnyEvent;
+use Message::Passing::Output::Test;
+use Message::Passing::Input::Socket::UDP;
+use Message::Passing::Output::Socket::UDP;
+
+plan skip_all => "Need Net::Statsd for this test"
+    unless eval { require Net::Statsd; 1; };
+
+my $t = Message::Passing::Output::Test->new;
+my $chain = Message::Passing::Input::Socket::UDP->new(
+    hostname => "localhost",
+    port => "52552",
+    output_to => $t,
+);
+
+$Net::Statsd::PORT = 52552;
+
+is $t->message_count, 0;
+
+Net::Statsd::increment('site.logins');
+
+my $cv = AnyEvent->condvar;
+my $timer = AnyEvent->timer(after => 0.1, cb => sub { $cv->send });
+$cv->recv;
+
+is $t->message_count, 1;
+
+my $out = Message::Passing::Output::Socket::UDP->new(
+    hostname => "localhost",
+    port => '52552',
+);
+
+$cv = AnyEvent->condvar;
+$timer = AnyEvent->timer(after => 0.1, cb => sub { $cv->send });
+$cv->recv;
+
+$out->consume("foo:bar");
+
+$cv = AnyEvent->condvar;
+$timer = AnyEvent->timer(after => 0.1, cb => sub { $cv->send });
+$cv->recv;
+
+is $t->message_count, 2;
+
+is_deeply [$t->messages],
+     [
+          'site.logins:1|c',
+          'foo:bar'
+        ];
+
+done_testing;
+

-- 
libmessage-passing-perl Debian packaging