ViewVC Help
View File | Revision Log | Show Annotations | Download File
/cvs/AnyEvent-FCP/FCP.pm
(Generate patch)

Comparing AnyEvent-FCP/FCP.pm (file contents):
Revision 1.2 by root, Sat Jul 25 06:28:49 2009 UTC vs.
Revision 1.9 by root, Tue Aug 4 00:35:16 2015 UTC

… …
2 2
3AnyEvent::FCP - freenet client protocol 2.0 3AnyEvent::FCP - freenet client protocol 2.0
4 4
5=head1 SYNOPSIS 5=head1 SYNOPSIS
6 6
7 use AnyEvent::FCP; 7 use AnyEvent::FCP;
8 8
9 my $fcp = new AnyEvent::FCP; 9 my $fcp = new AnyEvent::FCP;
10 10
11 my $ni = $fcp->txn_node_info->result; 11 # transactions return condvars
12 my $ni = $fcp->node_info; 12 my $lp_cv = $fcp->list_peers;
13 my $pr_cv = $fcp->list_persistent_requests;
14
15 my $peers = $lp_cv->recv;
16 my $reqs = $pr_cv->recv;
13 17
14=head1 DESCRIPTION 18=head1 DESCRIPTION
15 19
16This module implements the freenet client protocol version 2.0, as used by 20This module implements the freenet client protocol version 2.0, as used by
17freenet 0.7. See L<Net::FCP> for the earlier freenet 0.5 version. 21freenet 0.7. See L<Net::FCP> for the earlier freenet 0.5 version.
18 22
19See L<http://wiki.freenetproject.org/FreenetFCPSpec2Point0> for a description 23See L<http://wiki.freenetproject.org/FreenetFCPSpec2Point0> for a
20of what the messages do. 24description of what the messages do.
21 25
22The module uses L<AnyEvent> to find a suitable event module. 26The module uses L<AnyEvent> to find a suitable event module.
23 27
28Only very little is implemented, ask if you need more, and look at the
29example program later in this section.
30
31=head2 EXAMPLE
32
33This example fetches the download list and sets the priority of all files
34with "a" in their name to "emergency":
35
36 use AnyEvent::FCP;
37
38 my $fcp = new AnyEvent::FCP;
39
40 $fcp->watch_global_sync (1, 0);
41 my $req = $fcp->list_persistent_requests_sync;
42
43 for my $req (values %$req) {
44 if ($req->{filename} =~ /a/) {
45 $fcp->modify_persistent_request_sync (1, $req->{identifier}, undef, 0);
46 }
47 }
48
24=head2 IMPORT TAGS 49=head2 IMPORT TAGS
25 50
26Nothing much can be "imported" from this module right now. 51Nothing much can be "imported" from this module right now.
27 52
28=head2 FREENET BASICS 53=head2 THE AnyEvent::FCP CLASS
29
30Ok, this section will not explain any freenet basics to you, just some
31problems I found that you might want to avoid:
32 54
33=over 4 55=over 4
34 56
35=item freenet URIs are _NOT_ URIs
36
37Whenever a "uri" is required by the protocol, freenet expects a kind of
38URI prefixed with the "freenet:" scheme, e.g. "freenet:CHK...". However,
39these are not URIs, as freeent fails to parse them correctly, that is, you
40must unescape an escaped characters ("%2c" => ",") yourself. Maybe in the
41future this library will do it for you, so watch out for this incompatible
42change.
43
44=back
45
46=head2 THE AnyEvent::FCP CLASS
47
48=over 4
49
50=cut 57=cut
51 58
52package AnyEvent::FCP; 59package AnyEvent::FCP;
53 60
54use common::sense; 61use common::sense;
55 62
56use Carp; 63use Carp;
57 64
58our $VERSION = '0.1'; 65our $VERSION = '0.3';
59 66
60use Scalar::Util (); 67use Scalar::Util ();
61 68
62use AnyEvent; 69use AnyEvent;
63use AnyEvent::Handle; 70use AnyEvent::Handle;
71use AnyEvent::Util ();
64 72
65sub touc($) { 73sub touc($) {
66 local $_ = shift; 74 local $_ = shift;
67 1 while s/((?:^|_)(?:svk|chk|uri|fcp)(?:_|$))/\U$1/; 75 1 while s/((?:^|_)(?:svk|chk|uri|fcp|ds|mime|dda)(?:_|$))/\U$1/;
68 s/(?:^|_)(.)/\U$1/g; 76 s/(?:^|_)(.)/\U$1/g;
69 $_ 77 $_
70} 78}
71 79
72sub tolc($) { 80sub tolc($) {
73 local $_ = shift; 81 local $_ = shift;
74 1 while s/(SVK|CHK|URI|FCP)([^_])/$1\_$2/i; 82 1 while s/(SVK|CHK|URI|FCP|DS|MIME|DDA)([^_])/$1\_$2/i;
75 1 while s/([^_])(SVK|CHK|URI|FCP)/$1\_$2/i; 83 1 while s/([^_])(SVK|CHK|URI|FCP|DS|MIME|DDA)/$1\_$2/i;
76 s/(?<=[a-z])(?=[A-Z])/_/g; 84 s/(?<=[a-z])(?=[A-Z])/_/g;
77 lc 85 lc
78} 86}
79 87
80=item $fcp = new AnyEvent::FCP [host => $host][, port => $port][, progress => \&cb][, name => $name] 88=item $fcp = new AnyEvent::FCP [host => $host][, port => $port][, progress => \&cb][, name => $name]
81 89
82Create a new FCP connection to the given host and port (default 90Create a new FCP connection to the given host and port (default
83127.0.0.1:9481, or the environment variables C<FREDHOST> and C<FREDPORT>). 91127.0.0.1:9481, or the environment variables C<FREDHOST> and C<FREDPORT>).
84 92
85If no C<name> was specified, then AnyEvent::FCP will generate a (hopefully) 93If no C<name> was specified, then AnyEvent::FCP will generate a
86unique client name for you. 94(hopefully) unique client name for you.
87 95
88#TODO
89#You can install a progress callback that is being called with the AnyEvent::FCP 96You can install a progress callback that is being called with the AnyEvent::FCP
90#object, a txn object, the type of the transaction and the attributes. Use 97object, the type, a hashref with key-value pairs and a reference to any received data,
91#it like this: 98for all unsolicited messages.
92# 99
100Example:
101
93# sub progress_cb { 102 sub progress_cb {
94# my ($self, $txn, $type, $attr) = @_; 103 my ($self, $type, $kv, $rdata) = @_;
95# 104
96# warn "progress<$txn,$type," . (join ":", %$attr) . ">\n"; 105 if ($type eq "simple_progress") {
106 warn "$kv->{identifier} $kv->{succeeded}/$kv->{required}\n";
107 }
97# } 108 }
98 109
99=cut 110=cut
100 111
101sub new { 112sub new {
102 my $class = shift; 113 my $class = shift;
103 my $self = bless { @_ }, $class; 114 my $self = bless { @_ }, $class;
104 115
105 $self->{host} ||= $ENV{FREDHOST} || "127.0.0.1"; 116 $self->{host} ||= $ENV{FREDHOST} || "127.0.0.1";
106 $self->{port} ||= $ENV{FREDPORT} || 9481; 117 $self->{port} ||= $ENV{FREDPORT} || 9481;
107 $self->{name} ||= time.rand.rand.rand; # lame 118 $self->{name} ||= time.rand.rand.rand; # lame
108 $self->{timeout} ||= 600; 119 $self->{timeout} ||= 3600*2;
120 $self->{progress} ||= sub { };
109 121
110 $self->{id} = "a0"; 122 $self->{id} = "a0";
111 123
112 { 124 {
113 Scalar::Util::weaken (my $self = $self); 125 Scalar::Util::weaken (my $self = $self);
114 126
115 $self->{hdl} = new AnyEvent::Handle 127 $self->{hdl} = new AnyEvent::Handle
116 connect => [$self->{host} => $self->{port}], 128 connect => [$self->{host} => $self->{port}],
117 timeout => $self->{timeout}, 129 timeout => $self->{timeout},
118 on_error => sub { 130 on_error => sub {
119 warn "<@_>\n"; 131 warn "@_\n";#d#
120 exit 1; 132 exit 1;
121 }, 133 },
122 on_read => sub { $self->on_read (@_) }, 134 on_read => sub { $self->on_read (@_) },
123 on_eof => $self->{on_eof} || sub { }; 135 on_eof => $self->{on_eof} || sub { };
124 136
… …
132 ); 144 );
133 145
134 $self 146 $self
135} 147}
136 148
137sub progress {
138 my ($self, $txn, $type, $attr) = @_;
139
140 $self->{progress}->($self, $txn, $type, $attr)
141 if $self->{progress};
142}
143
144sub send_msg { 149sub send_msg {
145 my ($self, $type, %kv) = @_; 150 my ($self, $type, %kv) = @_;
146 151
147 my $data = delete $kv{data}; 152 my $data = delete $kv{data};
148 153
149 if (exists $kv{id_cb}) { 154 if (exists $kv{id_cb}) {
150 my $id = $kv{identifier} || ++$self->{id}; 155 my $id = $kv{identifier} ||= ++$self->{id};
151 $self->{id}{$id} = delete $kv{id_cb}; 156 $self->{id}{$id} = delete $kv{id_cb};
152 $kv{identifier} = $id;
153 } 157 }
154 158
155 my $msg = (touc $type) . "\012" 159 my $msg = (touc $type) . "\012"
156 . join "", map +(touc $_) . "=$kv{$_}\012", keys %kv; 160 . join "", map +(touc $_) . "=$kv{$_}\012", keys %kv;
157 161
… …
169 } 173 }
170 174
171 $self->{hdl}->push_write ($msg); 175 $self->{hdl}->push_write ($msg);
172} 176}
173 177
178sub on {
179 my ($self, $cb) = @_;
180
181 # cb return undef - message eaten, remove cb
182 # cb return 0 - message eaten
183 # cb return 1 - pass to next
184
185 push @{ $self->{on} }, $cb;
186}
187
188sub _push_queue {
189 my ($self, $queue) = @_;
190
191 warn "oush @$queue\n";#d#
192 shift @$queue;
193 $queue->[0]($self, AnyEvent::Util::guard { $self->_push_queue ($queue) })
194 if @$queue;
195}
196
197# lock so only one $type (arbitrary string) is in flight,
198# to work around horribly misdesigned protocol.
199sub serialise {
200 my ($self, $type, $cb) = @_;
201
202 my $queue = $self->{serialise}{$type} ||= [];
203 push @$queue, $cb;
204 $cb->($self, AnyEvent::Util::guard { $self->_push_queue ($queue) })
205 unless $#$queue;
206}
207
174sub on_read { 208sub on_read {
175 my ($self) = @_; 209 my ($self) = @_;
176 210
177 my $type; 211 my $type;
178 my %kv; 212 my %kv;
179 my $rdata; 213 my $rdata;
180 214
181 my $done_cb = sub { 215 my $done_cb = sub {
182 $kv{pkt_type} = $type; 216 $kv{pkt_type} = $type;
217
218 my $on = $self->{on};
219 for (0 .. $#$on) {
220 unless (my $res = $on->[$_]($type, \%kv, $rdata)) {
221 splice @$on, $_, 1 unless defined $res;
222 return;
223 }
224 }
183 225
184 if (my $cb = $self->{queue}[0]) { 226 if (my $cb = $self->{queue}[0]) {
185 $cb->($self, $type, \%kv, $rdata) 227 $cb->($self, $type, \%kv, $rdata)
186 and shift @{ $self->{queue} }; 228 and shift @{ $self->{queue} };
187 } else { 229 } else {
… …
232 $self->{node_hello} = $kv; 274 $self->{node_hello} = $kv;
233 } elsif (exists $self->{id}{$kv->{identifier}}) { 275 } elsif (exists $self->{id}{$kv->{identifier}}) {
234 $self->{id}{$kv->{identifier}}($self, $type, $kv, $rdata) 276 $self->{id}{$kv->{identifier}}($self, $type, $kv, $rdata)
235 and delete $self->{id}{$kv->{identifier}}; 277 and delete $self->{id}{$kv->{identifier}};
236 } else { 278 } else {
237 # on_warn 279 &{ $self->{progress} };
238 #warn "protocol warning (unexpected $type message)\n";
239 } 280 }
240} 281}
241 282
242sub _txn { 283sub _txn {
243 my ($name, $sub) = @_; 284 my ($name, $sub) = @_;
… …
252 splice @_, 1, 0, (my $cv = AnyEvent->condvar); 293 splice @_, 1, 0, (my $cv = AnyEvent->condvar);
253 &$sub; 294 &$sub;
254 $cv->recv 295 $cv->recv
255 }; 296 };
256} 297}
298
299=item $cv = $fcp->list_peers ([$with_metdata[, $with_volatile]])
300
301=item $peers = $fcp->list_peers_sync ([$with_metdata[, $with_volatile]])
302
303=cut
257 304
258_txn list_peers => sub { 305_txn list_peers => sub {
259 my ($self, $cv, $with_metadata, $with_volatile) = @_; 306 my ($self, $cv, $with_metadata, $with_volatile) = @_;
260 307
261 my @res; 308 my @res;
… …
275 } 322 }
276 }, 323 },
277 ); 324 );
278}; 325};
279 326
327=item $cv = $fcp->list_peer_notes ($node_identifier)
328
329=item $notes = $fcp->list_peer_notes_sync ($node_identifier)
330
331=cut
332
280_txn list_peer_notes => sub { 333_txn list_peer_notes => sub {
281 my ($self, $cv, $node_identifier) = @_; 334 my ($self, $cv, $node_identifier) = @_;
282 335
283 $self->send_msg (list_peer_notes => 336 $self->send_msg (list_peer_notes =>
284 node_identifier => $node_identifier, 337 node_identifier => $node_identifier,
… …
289 1 342 1
290 }, 343 },
291 ); 344 );
292}; 345};
293 346
347=item $cv = $fcp->watch_global ($enabled[, $verbosity_mask])
348
349=item $fcp->watch_global_sync ($enabled[, $verbosity_mask])
350
351=cut
352
294_txn watch_global => sub { 353_txn watch_global => sub {
295 my ($self, $cv, $enabled, $verbosity_mask) = @_; 354 my ($self, $cv, $enabled, $verbosity_mask) = @_;
296 355
297 $self->send_msg (watch_global => 356 $self->send_msg (watch_global =>
298 enabled => $enabled ? "true" : "false", 357 enabled => $enabled ? "true" : "false",
299 defined $verbosity_mask ? (verbosity_mask => $verbosity_mask+0) : (), 358 defined $verbosity_mask ? (verbosity_mask => $verbosity_mask+0) : (),
300 ); 359 );
301 360
302 $cv->(); 361 $cv->();
303}; 362};
363
364=item $cv = $fcp->list_persistent_requests
365
366=item $reqs = $fcp->list_persistent_requests_sync
367
368=cut
304 369
305_txn list_persistent_requests => sub { 370_txn list_persistent_requests => sub {
306 my ($self, $cv) = @_; 371 my ($self, $cv) = @_;
307 372
308 my %res; 373 my %res;
… …
333 0 398 0
334 } 399 }
335 }; 400 };
336}; 401};
337 402
403=item $cv = $fcp->remove_request ($global, $identifier)
404
405=item $status = $fcp->remove_request_sync ($global, $identifier)
406
407=cut
408
338_txn remove_request => sub { 409_txn remove_request => sub {
339 my ($self, $cv, $global, $identifier) = @_; 410 my ($self, $cv, $global, $identifier) = @_;
340 411
341 $self->send_msg (remove_request => 412 $self->send_msg (remove_request =>
342 global => $global ? "true" : "false", 413 global => $global ? "true" : "false",
… …
346 417
347 $cv->($kv); 418 $cv->($kv);
348 1 419 1
349 }, 420 },
350 ); 421 );
351
352 $cv->();
353}; 422};
423
424=item $cv = $fcp->modify_persistent_request ($global, $identifier[, $client_token[, $priority_class]])
425
426=item $sync = $fcp->modify_persistent_request_sync ($global, $identifier[, $client_token[, $priority_class]])
427
428=cut
354 429
355_txn modify_persistent_request => sub { 430_txn modify_persistent_request => sub {
356 my ($self, $cv, $global, $identifier, $client_token, $priority_class) = @_; 431 my ($self, $cv, $global, $identifier, $client_token, $priority_class) = @_;
357 432
358 $self->send_msg (modify_persistent_request => 433 $self->send_msg (modify_persistent_request =>
359 global => $global ? "true" : "false", 434 global => $global ? "true" : "false",
360 identifier => $identifier,
361 defined $client_token ? (client_token => $client_token ) : (), 435 defined $client_token ? (client_token => $client_token ) : (),
362 defined $priority_class ? (priority_class => $priority_class) : (), 436 defined $priority_class ? (priority_class => $priority_class) : (),
437 identifier => $identifier,
363 id_cb => sub { 438 id_cb => sub {
364 my ($self, $type, $kv, $rdata) = @_; 439 my ($self, $type, $kv, $rdata) = @_;
365 440
366 $cv->($kv); 441 $cv->($kv);
367 1 442 1
368 }, 443 },
369 ); 444 );
370
371 $cv->();
372}; 445};
446
447=item $cv = $fcp->get_plugin_info ($name, $detailed)
448
449=item $info = $fcp->get_plugin_info_sync ($name, $detailed)
450
451=cut
373 452
374_txn get_plugin_info => sub { 453_txn get_plugin_info => sub {
375 my ($self, $cv, $name, $detailed) = @_; 454 my ($self, $cv, $name, $detailed) = @_;
376 455
377 $self->send_msg (get_plugin_info => 456 $self->send_msg (get_plugin_info =>
… …
382 461
383 $cv->($kv); 462 $cv->($kv);
384 1 463 1
385 }, 464 },
386 ); 465 );
466};
387 467
388 $cv->(); 468=item $cv = $fcp->client_get ($uri, $identifier, %kv)
469
470=item $status = $fcp->client_get_sync ($uri, $identifier, %kv)
471
472%kv can contain (L<http://wiki.freenetproject.org/FCP2p0ClientGet>).
473
474ignore_ds, ds_only, verbosity, max_size, max_temp_size, max_retries,
475priority_class, persistence, client_token, global, return_type,
476binary_blob, allowed_mime_types, filename, temp_filename
477
478=cut
479
480_txn client_get => sub {
481 my ($self, $cv, $uri, $identifier, %kv) = @_;
482
483 $self->send_msg (client_get =>
484 %kv,
485 uri => $uri,
486 identifier => $identifier,
487 id_cb => sub {
488 my ($self, $type, $kv, $rdata) = @_;
489
490 $cv->($kv);
491 1
492 },
493 );
494};
495
496=item $cv = $fcp->test_dda_sync ($local_directory, $remote_directory, $want_read, $want_write)
497
498=item ($can_read, $can_write) = $fcp->test_dda_sync ($local_directory, $remote_directory, $want_read, $want_write))
499
500The DDA test in FCP is probably the single most broken protocol - only
501one directory test can be outstanding at any time, and some guessing and
502heuristics are involved in mangling the paths.
503
504This function combines C<TestDDARequest> and C<TestDDAResponse> in one
505request, handling file reading and writing as well.
506
507=cut
508
509_txn test_dda => sub {
510 my ($self, $cv, $local, $remote, $want_read, $want_write) = @_;
511
512 $self->serialise (test_dda => sub {
513 my ($self, $guard) = @_;
514
515 $self->send_msg (test_dda_request =>
516 directory => $remote,
517 want_read_directory => $want_read ? "true" : "false",
518 want_write_directory => $want_write ? "true" : "false",
519 );
520 $self->on (sub {
521 my ($type, $kv) = @_;
522
523 if ($type eq "test_dda_reply") {
524 # the filenames are all relative to the server-side directory,
525 # which might or might not match $remote anymore, so we
526 # need to rewrite the paths to be relative to $local
527 for my $k (qw(read_filename write_filename)) {
528 my $f = $kv->{$k};
529 for my $dir ($kv->{directory}, $remote) {
530 if ($dir eq substr $f, 0, length $dir) {
531 substr $f, 0, 1 + length $dir, "";
532 $kv->{$k} = $f;
533 last;
534 }
535 }
536 }
537
538 my %response = (directory => $remote);
539
540 if (length $kv->{read_filename}) {
541 warn "$local/$kv->{read_filename}";#d#
542 if (open my $fh, "<:raw", "$local/$kv->{read_filename}") {
543 sysread $fh, my $buf, -s $fh;
544 $response{read_content} = $buf;
545 }
546 }
547
548 if (length $kv->{write_filename}) {
549 if (open my $fh, ">:raw", "$local/$kv->{write_filename}") {
550 syswrite $fh, $kv->{content_to_write};
551 }
552 }
553
554 $self->send_msg (test_dda_response => %response);
555
556 $self->on (sub {
557 my ($type, $kv) = @_;
558
559 $guard if 0; # reference
560
561 if ($type eq "test_dda_complete") {
562 $cv->(
563 $kv->{read_directory_allowed} eq "true",
564 $kv->{write_directory_allowed} eq "true",
565 );
566 } elsif ($type eq "protocol_error" && $kv->{identifier} eq $remote) {
567 $cv->croak ($kv->{extra_description});
568 return;
569 }
570
571 1
572 });
573
574 return;
575 } elsif ($type eq "protocol_error" && $kv->{identifier} eq $remote) {
576 $cv->croak ($kv->{extra_description});
577 return;
578 }
579
580 1
581 });
582 });
389}; 583};
390 584
391=back 585=back
586
587=head1 EXAMPLE PROGRAM
588
589 use AnyEvent::FCP;
590
591 my $fcp = new AnyEvent::FCP;
592
593 # let us look at the global request list
594 $fcp->watch_global (1, 0);
595
596 # list them, synchronously
597 my $req = $fcp->list_persistent_requests_sync;
598
599 # go through all requests
600 for my $req (values %$req) {
601 # skip jobs not directly-to-disk
602 next unless $req->{return_type} eq "disk";
603 # skip jobs not issued by FProxy
604 next unless $req->{identifier} =~ /^FProxy:/;
605
606 if ($req->{data_found}) {
607 # file has been successfully downloaded
608
609 ... move the file away
610 (left as exercise)
611
612 # remove the request
613
614 $fcp->remove_request (1, $req->{identifier});
615 } elsif ($req->{get_failed}) {
616 # request has failed
617 if ($req->{get_failed}{code} == 11) {
618 # too many path components, should restart
619 } else {
620 # other failure
621 }
622 } else {
623 # modify priorities randomly, to improve download rates
624 $fcp->modify_persistent_request (1, $req->{identifier}, undef, int 6 - 5 * (rand) ** 1.7)
625 if 0.1 > rand;
626 }
627 }
628
629 # see if the dummy plugin is loaded, to ensure all previous requests have finished.
630 $fcp->get_plugin_info_sync ("dummy");
392 631
393=head1 SEE ALSO 632=head1 SEE ALSO
394 633
395L<http://wiki.freenetproject.org/FreenetFCPSpec2Point0>, L<Net::FCP>. 634L<http://wiki.freenetproject.org/FreenetFCPSpec2Point0>, L<Net::FCP>.
396 635

Diff Legend

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