[ViewVC] Diff of: cvs/AnyEvent-FCP/FCP.pm

Comparing AnyEvent-FCP/FCP.pm (file contents):
Revision 1.14 by root, Sat Aug 8 14:09:47 2015 UTC vs.
Revision 1.24 by root, Sun Jun 12 04:44:05 2016 UTC

 use common::sense;
 use Carp;
-our $VERSION = '0.3';
+our $VERSION = 0.5;
 use Scalar::Util ();
 use AnyEvent;
 use AnyEvent::Handle;
 use AnyEvent::Util ();
+our %TOLC; # tolc cache
 sub touc($) {
    local $_ = shift;
    1 while s/((?:^|_)(?:svk|chk|uri|fcp|ds|mime|dda)(?:_|$))/\U$1/;
    s/(?:^|_)(.)/\U$1/g;
    1 while s/([^_])(SVK|CHK|URI|FCP|DS|MIME|DDA)/$1\_$2/;
    s/(?<=[a-z])(?=[A-Z])/_/g;
    lc
 }
-=item $fcp = new AnyEvent::FCP [host => $host][, port => $port][, name => $name]
+=item $fcp = new AnyEvent::FCP key => value...;
 Create a new FCP connection to the given host and port (default
 127.0.0.1:9481, or the environment variables C<FREDHOST> and C<FREDPORT>).
 If no C<name> was specified, then AnyEvent::FCP will generate a
 (hopefully) unique client name for you.
+The following keys can be specified (they are all optional):
+=over 4
+=item name => $string
+A unique name to identify this client. If none is specified, a randomly
+generated name will be used.
+=item host => $hostname
+The hostname or IP address of the freenet node. Default is C<$ENV{FREDHOST}>
+or C<127.0.0.1>.
+=item port => $portnumber
+The port number of the FCP port. Default is C<$ENV{FREDPORT}> or C<9481>.
+=item timeout => $seconds
+The timeout, in seconds, after which a connection error is assumed when
+there is no activity. Default is C<7200>, i.e. two hours.
+=item keepalive => $seconds
+The interval, in seconds, at which keepalive messages will be
+sent. Default is C<540>, i.e. nine minutes.
+These keepalive messages are useful both to detect that a connection is
+no longer working and to keep any (home) routers from expiring their
+masquerading entry.
+=item on_eof => $callback->($fcp)
+Invoked when the underlying L<AnyEvent::Handle> signals EOF, currently
+regardless of whether the EOF was expected or not.
+=item on_error => $callback->($fcp, $message)
+Invoked on any (fatal) errors, such as unexpected connection close. The
+callback receives the FCP object and a textual error message.
+=item on_failure => $callback->($fcp, $type, $backtrace, $args, $error)
+Invoked when an FCP request fails that didn't have a failure callback. See
+L<FCP REQUESTS> for details.
+=back
 =cut
 sub new {
    my $class = shift;
    my $self = bless {
       host       => $ENV{FREDHOST} || "127.0.0.1",
       port       => $ENV{FREDPORT} || 9481,
       timeout    => 3600 * 2,
+      keepalive  => 9 * 60,
       name       => time.rand.rand.rand, # lame
       @_,
       queue      => [],
       req        => {},
-      prefix     => "..:aefcpid-$rand:",
+      prefix     => "..:aefcpid:$rand:",
       idseq      => "a0",
    }, $class;
    {
       Scalar::Util::weaken (my $self = $self);
+      $self->{kw} = AE::timer $self->{keepalive}, $self->{keepalive}, sub {
+         $self->{hdl}->push_write ("\n");
+      };
+      our $ENDMESSAGE = qr<\012(EndMessage|Data)\012>;
+      # these are declared here for performance reasons
+      my ($k, $v, $type);
+      my $rdata;
+      my $on_read = sub {
+         my ($hdl) = @_;
+         # we only carve out whole messages here
+         while ($hdl->{rbuf} =~ /\012(EndMessage|Data)\012/) {
+            # remember end marker
+            $rdata = $1 eq "Data"
+               or $1 eq "EndMessage"
+               or return $self->fatal ("protocol error, expected message end, got $1\n");
+            my @lines = split /\012/, substr $hdl->{rbuf}, 0, $-[0];
+            substr $hdl->{rbuf}, 0, $+[0], ""; # remove pkg
+            $type = shift @lines;
+            $type = ($TOLC{$type} ||= tolc $type);
+            my %kv;
+            for (@lines) {
+               ($k, $v) = split /=/, $_, 2;
+               $k = ($TOLC{$k} ||= tolc $k);
+               if ($k =~ /\./) {
+                  # generic, slow case
+                  my @k = split /\./, $k;
+                  my $ro = \\%kv;
+                  while (@k) {
+                     $k = shift @k;
+                     if ($k =~ /^\d+$/) {
+                        $ro = \$$ro->[$k];
+                     } else {
+                        $ro = \$$ro->{$k};
+                     }
+                  }
+                  $$ro = $v;
+                  next;
+               }
+               # special comon case, for performance only
+               $kv{$k} = $v;
+            }
+            if ($rdata) {
+               $_[0]->push_read (chunk => delete $kv{data_length}, sub {
+                  $rdata = \$_[1];
+                  $self->recv ($type, \%kv, $rdata);
+               });
+               last; # do not tgry to parse more messages
+            } else {
+               $self->recv ($type, \%kv);
+            }
+         }
+      };
       $self->{hdl} = new AnyEvent::Handle
          connect  => [$self->{host} => $self->{port}],
          timeout  => $self->{timeout},
+         on_read  => $on_read,
+         on_eof   => sub {
+            if ($self->{on_eof}) {
+               $self->{on_eof}($self);
+            } else {
+               $self->fatal ("EOF");
+            }
+         },
          on_error => sub {
-            warn "@_\n";#d#
+            $self->fatal ($_[2]);
-            exit 1;
          },
-         on_read  => sub { $self->on_read (@_) },
+      ;
-         on_eof   => $self->{on_eof} || sub { };
       Scalar::Util::weaken ($self->{hdl}{fcp} = $self);
    }
    $self->send_msg (client_hello =>
       name             => $self->{name},
       expected_version => "2.0",
    );
    $self
+}
+sub fatal {
+   my ($self, $msg) = @_;
+   $self->{hdl}->shutdown;
+   delete $self->{kw};
+   if ($self->{on_error}) {
+      $self->{on_error}->($self, $msg);
+   } else {
+      die $msg;
+   }
 }
 sub identifier {
    $_[0]{prefix} . ++$_[0]{idseq}
 }
    if (my $cb = $PERSISTENT_TYPE{$type}) {
       my $id  = $kv->{identifier};
       my $req = $_[0]{req}{$id} ||= {};
       $cb->($self, $req, $kv);
-      $self->recv (request_change => $kv, $type, @extra);
+      $self->recv (request_changed => $kv, $type, @extra);
    }
    my $on = $self->{on};
    for (0 .. $#$on) {
       unless (my $res = $on->[$_]($self, $type, $kv, @extra)) {
    } else {
       $self->default_recv ($type, $kv, @extra);
    }
 }
-sub on_read {
-   my ($self) = @_;
-   my $type;
-   my %kv;
-   my $rdata;
-   my $hdr_cb; $hdr_cb = sub {
-      if ($_[1] =~ /^([^=]+)=(.*)$/) {
-         my ($k, $v) = ($1, $2);
-         my @k = split /\./, tolc $k;
-         my $ro = \\%kv;
-         while (@k) {
-            my $k = shift @k;
-            if ($k =~ /^\d+$/) {
-               $ro = \$$ro->[$k];
-            } else {
-               $ro = \$$ro->{$k};
-            }
-         }
-         $$ro = $v;
-         $_[0]->push_read (line => $hdr_cb);
-      } elsif ($_[1] eq "Data") {
-         $_[0]->push_read (chunk => delete $kv{data_length}, sub {
-            $rdata = \$_[1];
-            $self->recv ($type, \%kv, $rdata);
-         });
-      } elsif ($_[1] eq "EndMessage") {
-         $self->recv ($type, \%kv);
-      } else {
-         die "protocol error, expected message end, got $_[1]\n";#d#
-      }
-   };
-   $self->{hdl}->push_read (line => sub {
-      $type = tolc $_[1];
-      $_[0]->push_read (line => $hdr_cb);
-   });
-}
 sub default_recv {
    my ($self, $type, $kv, $rdata) = @_;
    if ($type eq "node_hello") {
       $self->{node_hello} = $kv;
 Also comes in this underscore variant:
    $fcp->get_plugin_info_ ($name, $detailed, $cb);
-You can thinbk of the underscore as a kind of continuation indicator - the
+You can think of the underscore as a kind of continuation indicator - the
 normal function waits and returns with the data, the C<_> indicates that
 you pass the continuation yourself, and the continuation will be invoked
 with the results.
 This callback/continuation argument (C<$cb>) can come in three forms itself:
 =over 4
 =item A code reference (or rather anything not matching some other alternative)
 This code reference will be invoked with the result on success. On an
+error, it will invoke the C<on_failure> callback of the FCP object, or,
-error, it will die (in the event loop) with a backtrace of the call site.
+if none was defined, will die (in the event loop) with a backtrace of the
+call site.
 This is a popular choice, but it makes handling errors hard - make sure
 you never generate protocol errors!
+In the failure case, if an C<on_failure> hook exists, it will be invoked
+with the FCP object, the request type (the name of the method), a
+(textual) backtrace as generated by C<Carp::longmess>, and arrayref
+containing the arguments from the original request invocation and the
+error object from the server, in this order, e.g.:
+   on_failure => sub {
+      my ($fcp, $request_type, $backtrace, $orig_args, $error_object) = @_;
+      warn "FCP failure ($type), $error_object->{code_description} ($error_object->{extra_description})$backtrace";
+      exit 1;
+   },
 =item A condvar (as returned by e.g. C<< AnyEvent->condvar >>)
 When a condvar is passed, it is sent (C<< $cv->send ($results) >>) the
 results when the request has finished. Should an error occur, the error
 =item An array with two callbacks C<[$success, $failure]>
 The C<$success> callback will be invoked with the results, while the
 C<$failure> callback will be invoked on any errors.
+The C<$failure> callback will be invoked with the error object from the
+server.
 =item C<undef>
 This is the same thing as specifying C<sub { }> as callback, i.e. on
-success, the results are ignored, while on failure, you the module dies
+success, the results are ignored, while on failure, the C<on_failure> hook
-with a backtrace.
+is invoked or the module dies with a backtrace.
 This is good for quick scripts, or when you really aren't interested in
 the results.
 =back
       if (ARRAY:: eq ref $ok) {
          ($ok, $err) = @$ok;
       } elsif (UNIVERSAL::isa $ok, AnyEvent::CondVar::) {
          $err = sub { $ok->croak ($_[0]{extra_description}) };
       } else {
-         my $bt = Carp::longmess "";
+         my $bt = Carp::longmess "AnyEvent::FCP request $name";
+         Scalar::Util::weaken (my $self = $_[0]);
+         my $args = [@_]; shift @$args;
          $err = sub {
+            if ($self->{on_failure}) {
+               $self->{on_failure}($self, $name, $args, $bt, $_[0]);
+            } else {
-            die "$_[0]{extra_description}$bt";
+               die "$_[0]{code_description} ($_[0]{extra_description})$bt";
+            }
          };
       }
       $ok ||= $NOP_CB;
 =cut
 _txn get_plugin_info => sub {
    my ($self, $ok, $err, $name, $detailed) = @_;
+   my $id = $self->identifier;
    $self->send_msg (get_plugin_info =>
+      identifier  => $id,
       plugin_name => $name,
       detailed    => $detailed ? "true" : "false",
-      id_cb       => sub {
-         my ($self, $type, $kv, $rdata) = @_;
-         $ok->($kv);
-         1
-      },
    );
+   $self->on (sub {
+      my ($self, $type, $kv) = @_;
+      if ($kv->{identifier} eq $id) {
+         if ($type eq "get_plugin_info") {
+            $ok->($kv);
+         } else {
+            $err->($kv, $type);
+         }
+         return;
+      }
+      1
+   });
 };
 =item $status = $fcp->client_get ($uri, $identifier, %kv)
 %kv can contain (L<http://wiki.freenetproject.org/FCP2p0ClientGet>).
    });
 };
 =item $status = $fcp->remove_request ($identifier[, $global])
-Remove the request with the given isdentifier. Returns true if successful,
+Remove the request with the given identifier. Returns true if successful,
 false on error.
 =cut
 _txn remove_request => sub {
 C<$want_read> and C<$want_write> should be set to a true value when you
 want to read (get) files or write (put) files, respectively.
 On error, an exception is thrown. Otherwise, C<$can_read> and
-C<$can_write> indicate whether you can reaqd or write to freenet via the
+C<$can_write> indicate whether you can read or write to freenet via the
 directory.
 =cut
 _txn test_dda => sub {
 on every change, which will be called as C<< $cb->($fcp, $kv, $type) >>, where C<$type>
 is the type of the original message triggering the change,
 To fill this cache with the global queue and keep it updated,
 call C<watch_global> to subscribe to updates, followed by
-C<list_persistent_requests_sync>.
+C<list_persistent_requests>.
-   $fcp->watch_global_sync_; # do not wait
+   $fcp->watch_global_; # do not wait
    $fcp->list_persistent_requests; # wait
 To get a better idea of what is stored in the cache, here is an example of
 what might be stored in C<< $fcp->{req}{"Frost-gpl.txt"} >>:
             if 0.1 > rand;
       }
    }
    # see if the dummy plugin is loaded, to ensure all previous requests have finished.
-   $fcp->get_plugin_info_sync ("dummy");
+   $fcp->get_plugin_info ("dummy");
 =head1 SEE ALSO
 L<http://wiki.freenetproject.org/FreenetFCPSpec2Point0>, L<Net::FCP>.

Diff Legend

-–
+Removed lines
-+
+Added lines
-<
+Changed lines
->
+Changed lines

Comparing AnyEvent-FCP/FCP.pm (file contents): Revision 1.14 by root, Sat Aug 8 14:09:47 2015 UTC vs. Revision 1.24 by root, Sun Jun 12 04:44:05 2016 UTC

Diff Legend

Comparing AnyEvent-FCP/FCP.pm (file contents):
Revision 1.14 by root, Sat Aug 8 14:09:47 2015 UTC vs.
Revision 1.24 by root, Sun Jun 12 04:44:05 2016 UTC