rough async UDP API client

1 files changed, 205 insertions, 0 deletions

diff --git a/lib/AniDB/APIClient.pm b/lib/AniDB/APIClient.pm
new file mode 100644
index 0000000..3453840
--- /dev/null
+++ b/lib/AniDB/APIClient.pm
@@ -0,0 +1,205 @@
+package AniDB::APIClient;
+use 5.024;
+use Moo;
+use experimental 'signatures';
+use experimental 'postderef';
+use IO::Async::Socket;
+use Log::Any '$log';
+use curry::weak;
+use Encode;
+use namespace::clean;
+
+has [qw(username password)] => (
+    is => 'ro',
+    required => 1,
+);
+
+has client_name => (
+    is => 'ro',
+    default => 'dakkarenamer',
+);
+
+has client_version => (
+    is => 'ro',
+    default => 1,
+);
+
+has host => (
+    is => 'ro',
+    default => 'api.anidb.info',
+);
+
+has port => (
+    is => 'ro',
+    default => 9000,
+);
+
+has local_port => (
+    is => 'ro',
+    default => 4568,
+);
+
+sub login($self) {
+    $self->_send(
+        AUTH => {
+            user => $self->username,
+            pass => $self->password,
+            protover => 3,
+            client => $self->client_name,
+            clientver => $self->client_version,
+            nat => 1,
+            enc => 'UTF-8',
+        }
+    )->then(
+        sub($code,$payload) {
+            my ($skey,$addr,$message) = split /\s+/,$payload,3;
+            $log->debugf('login response: %s',$message);
+            $self->_session_key($skey);
+            return Future->done($message);
+        }
+    );
+}
+
+sub logout($self) {
+    $self->_send('LOGOUT')->then_done();
+}
+
+has _loop => (
+    is => 'ro',
+    init_arg => 'loop',
+    default => sub {
+        require IO::Async::Loop;
+        return IO::Async::Loop->new;
+    },
+);
+
+has _pending_requests => (
+    is => 'rw',
+    default => sub { +{} },
+);
+
+sub _remove_pending_request($self,$tag,@) {
+    delete $self->_pending_requests->{$tag};
+}
+
+sub _future_for_tag($self,$tag) {
+    my $f = $self->_loop->new_future;
+    $f->on_done($self->curry::weak::_remove_pending_request($tag));
+    $self->_pending_requests->{$tag} = $f;
+    return $f;
+}
+
+has _socket_f => ( is => 'lazy' );
+
+sub _build__socket_f($self) {
+    my $socket = IO::Async::Socket->new(
+        on_recv => $self->curry::weak::_receive,
+    );
+    $self->_loop->add($socket);
+    return $socket->connect(
+        host => $self->host,
+        service => $self->port,
+        socktype => 'dgram',
+    );
+}
+
+has _last_send_time => ( is => 'rw', default => 0 );
+has _session_key => ( is => 'rw' );
+
+sub _escape($str) { $str =~ s{&}{&amp;}gr =~ s{\n}{<br />}gr }
+
+sub _send($self,$command,$command_args={},$delay=4,$want_reply=1) {
+    if (!$self->_session_key && $command ne 'AUTH' && $command ne 'PING') {
+        return $self->login
+            ->then(
+                $self->curry::weak::_send(
+                    $command,
+                    $command_args,
+                    $want_reply,
+                    $delay,
+                ),
+            );
+    }
+
+    if ($delay && int(time - $self->_last_send_time) < $delay) {
+        return $self->_loop->delay_future(after=>$delay)
+            ->then(
+                $self->curry::weak::_send(
+                    $command,
+                    $command_args,
+                    $want_reply,
+                    $delay,
+                ),
+            );
+    }
+
+    $command_args->{tag} = my $tag =
+        'dr-'.(1+keys $self->_pending_requests->%*);
+    $command_args->{s} = $self->_session_key if $self->_session_key;
+
+    $log->debugf('command %s - %s',$command,$command_args);
+
+    my $packet = "$command " . join '&',
+        map {
+            "$_=" .
+                _escape(
+                    Encode::encode(
+                        'UTF-8',
+                        $command_args->{$_},
+                        Encode::FB_HTMLCREF | Encode::LEAVE_SRC,
+                    )
+                  )
+            }
+        sort keys $command_args->%*;
+
+    my $reply_f = $want_reply ? $self->_future_for_tag($tag) : undef;
+
+    my $send_f;
+    $send_f = $self->_socket_f->then(
+        sub($s) {
+            $s->send($packet);
+            $self->_last_send_time(time);
+            undef $send_f;
+            return Future->done;
+        }
+    );
+
+    return $reply_f;
+}
+
+sub _receive($self,$socket,$data,@) {
+    $log->debugf('received packet <%s>',$data);
+
+    my $tag;
+    $data =~ s{\A(dr-\d+)\s+}{} and $tag=$1;
+    unless ($tag) {
+        $log->errorf('received untagged response <%s>',$data);
+        return;
+    }
+
+    my $future = $self->_pending_requests->{$tag};
+    unless ($future) {
+        $log->errorf(
+            'received unexpected response <%s> for tag <%s>',
+            $data,$tag,
+        );
+        return;
+    }
+
+    my ($code,$payload) = $data =~ m{\A(\d+)\s+(.*)\z}sm;
+    $log->debugf(
+        'response for tag=%s code=%s payload=<%s>',
+        $tag, $code, $payload,
+    );
+
+    if ($code =~ /^2/) {
+        $future->done($code,$payload);
+    }
+    else {
+        $future->fail($code,$payload);
+    }
+
+    return;
+}
+
+1;