From 73801c8488b30d87e4f1336ba28f80062cf28156 Mon Sep 17 00:00:00 2001
From: Bruno Tavares <eu@brunotavares.net>
Date: Wed, 3 Sep 2014 11:03:14 +0100
Subject: [PATCH] interface methods to message commands on the connection

---
 lib/AnyEvent/NSQ/Reader.pm | 40 ++++++++++++++++++++++++++++----------
 1 file changed, 30 insertions(+), 10 deletions(-)

diff --git a/lib/AnyEvent/NSQ/Reader.pm b/lib/AnyEvent/NSQ/Reader.pm
index b288d5c..a70043a 100644
--- a/lib/AnyEvent/NSQ/Reader.pm
+++ b/lib/AnyEvent/NSQ/Reader.pm
@@ -73,27 +73,47 @@ sub _identified {
 }
 
 sub mark_as_done_msg {
-  my $self    = shift;
-  my $message = shift;
+  my ($self, $msg) = @_;
+
+  my $conn = $self->_find_message_connection($msg);
+  return 0 unless $conn;
+
+  $conn->mark_as_done_msg($msg);
+  return 1;
 } 
 
 sub requeue_msg {
-  my $self    = shift;
-  my $message = shift;
+  my ($self, $msg, $delay) = @_;
+
+  my $conn = $self->_find_message_connection($msg);
+  return 0 unless $conn;
+
+  $conn->requeue_msg($msg, $delay);
+  return 1;
 }
 
 sub touch_message {
-  my $self    = shift;
-  my $message = shift;
+  my ($self, $msg) = @_;
+
+  my $conn = $self->_find_message_connection($msg);
+  return 0 unless $conn;
+  
+  $conn->touch_msg($msg);
+  return 1;
 }
 
 sub _find_message_connection {
-  my $self    = shift;
-  my $message = shift; 
+  my ($self, $msg) = @_;
 
-  my $message_id = ref($message) ? $message->{message_id} : $message;
+  my $id = ref($msg) ? $msg->{message_id} : $msg;
 
-  return $self->{routing}->{$message_id};
+  my $conn = $self->{routing}->{$id};
+ 
+  if ( !$conn ) {
+    warn "WARN: Could not find the connection to route msg $id";
+  }
+
+  return $conn;
 }
 
 1;