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

Comparing AnyEvent-FCP/FCP.pm (file contents):
Revision 1.5 by root, Tue Dec 1 13:49:56 2009 UTC vs.
Revision 1.16 by root, Sat Sep 5 13:26:47 2015 UTC

…		…
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	# transactions return condvars	11	# transactions return condvars
12	my $lp_cv = $fcp->list_peers;	12	my $lp_cv = $fcp->list_peers;
13	my $pr_cv = $fcp->list_persistent_requests;	13	my $pr_cv = $fcp->list_persistent_requests;
14		14
15	my $peers = $lp_cv->recv;	15	my $peers = $lp_cv->recv;
16	my $reqs = $pr_cv->recv;	16	my $reqs = $pr_cv->recv;
…		…
26	The module uses L<AnyEvent> to find a suitable event module.	26	The module uses L<AnyEvent> to find a suitable event module.
27		27
28	Only very little is implemented, ask if you need more, and look at the	28	Only very little is implemented, ask if you need more, and look at the
29	example program later in this section.	29	example program later in this section.
30		30
		31	=head2 EXAMPLE
		32
		33	This example fetches the download list and sets the priority of all files
		34	with "a" in their name to "emergency":
		35
		36	use AnyEvent::FCP;
		37
		38	my $fcp = new AnyEvent::FCP;
		39
		40	$fcp->watch_global (1, 0);
		41	my $req = $fcp->list_persistent_requests;
		42
		43	TODO
		44	for my $req (values %$req) {
		45	if ($req->{filename} =~ /a/) {
		46	$fcp->modify_persistent_request (1, $req->{identifier}, undef, 0);
		47	}
		48	}
		49
31	=head2 IMPORT TAGS	50	=head2 IMPORT TAGS
32		51
33	Nothing much can be "imported" from this module right now.	52	Nothing much can be "imported" from this module right now.
34		53
35	=head2 THE AnyEvent::FCP CLASS	54	=head1 THE AnyEvent::FCP CLASS
36		55
37	=over 4	56	=over 4
38		57
39	=cut	58	=cut
40		59
…		…
42		61
43	use common::sense;	62	use common::sense;
44		63
45	use Carp;	64	use Carp;
46		65
47	our $VERSION = '0.21';	66	our $VERSION = '0.3';
48		67
49	use Scalar::Util ();	68	use Scalar::Util ();
50		69
51	use AnyEvent;	70	use AnyEvent;
52	use AnyEvent::Handle;	71	use AnyEvent::Handle;
		72	use AnyEvent::Util ();
		73
		74	our %TOLC; # tolc cache
53		75
54	sub touc($) {	76	sub touc($) {
55	local $_ = shift;	77	local $_ = shift;
56	1 while s/((?:^\|_)(?:svk\|chk\|uri\|fcp\|ds\|mime)(?:_\|$))/\U$1/;	78	1 while s/((?:^\|_)(?:svk\|chk\|uri\|fcp\|ds\|mime\|dda)(?:_\|$))/\U$1/;
57	s/(?:^\|_)(.)/\U$1/g;	79	s/(?:^\|_)(.)/\U$1/g;
58	$_	80	$_
59	}	81	}
60		82
61	sub tolc($) {	83	sub tolc($) {
62	local $_ = shift;	84	local $_ = shift;
63	1 while s/(SVK\|CHK\|URI\|FCP\|DS\|MIME)([^_])/$1\_$2/i;	85	1 while s/(SVK\|CHK\|URI\|FCP\|DS\|MIME\|DDA)([^_])/$1\_$2/;
64	1 while s/([^_])(SVK\|CHK\|URI\|FCP\|DS\|MIME)/$1\_$2/i;	86	1 while s/([^_])(SVK\|CHK\|URI\|FCP\|DS\|MIME\|DDA)/$1\_$2/;
65	s/(?<=[a-z])(?=[A-Z])/_/g;	87	s/(?<=[a-z])(?=[A-Z])/_/g;
66	lc	88	lc
67	}	89	}
68		90
69	=item $fcp = new AnyEvent::FCP [host => $host][, port => $port][, progress => \&cb][, name => $name]	91	=item $fcp = new AnyEvent::FCP [host => $host][, port => $port][, name => $name]
70		92
71	Create a new FCP connection to the given host and port (default	93	Create a new FCP connection to the given host and port (default
72	127.0.0.1:9481, or the environment variables C<FREDHOST> and C<FREDPORT>).	94	127.0.0.1:9481, or the environment variables C<FREDHOST> and C<FREDPORT>).
73		95
74	If no C<name> was specified, then AnyEvent::FCP will generate a	96	If no C<name> was specified, then AnyEvent::FCP will generate a
75	(hopefully) unique client name for you.	97	(hopefully) unique client name for you.
76		98
77	=cut	99	=cut
78		100
79	#TODO
80	#You can install a progress callback that is being called with the AnyEvent::FCP
81	#object, a txn object, the type of the transaction and the attributes. Use
82	#it like this:
83	#
84	# sub progress_cb {
85	# my ($self, $txn, $type, $attr) = @_;
86	#
87	# warn "progress<$txn,$type," . (join ":", %$attr) . ">\n";
88	# }
89
90	sub new {	101	sub new {
91	my $class = shift;	102	my $class = shift;
		103
		104	my $rand = join "", map chr 0x21 + rand 94, 1..40; # ~ 262 bits entropy
		105
92	my $self = bless { @_ }, $class;	106	my $self = bless {
93		107	host => $ENV{FREDHOST} \|\| "127.0.0.1",
94	$self->{host} \|\|= $ENV{FREDHOST} \|\| "127.0.0.1";	108	port => $ENV{FREDPORT} \|\| 9481,
95	$self->{port} \|\|= $ENV{FREDPORT} \|\| 9481;	109	timeout => 3600 * 2,
96	$self->{name} \|\|= time.rand.rand.rand; # lame	110	name => time.rand.rand.rand, # lame
97	$self->{timeout} \|\|= 600;	111	@_,
98		112	queue => [],
99	$self->{id} = "a0";	113	req => {},
		114	prefix => "..:aefcpid:$rand:",
		115	idseq => "a0",
		116	}, $class;
100		117
101	{	118	{
102	Scalar::Util::weaken (my $self = $self);	119	Scalar::Util::weaken (my $self = $self);
103		120
104	$self->{hdl} = new AnyEvent::Handle	121	$self->{hdl} = new AnyEvent::Handle
105	connect => [$self->{host} => $self->{port}],	122	connect => [$self->{host} => $self->{port}],
106	timeout => $self->{timeout},	123	timeout => $self->{timeout},
107	on_error => sub {	124	on_error => sub {
108	warn "<@_>\n";	125	warn "@_\n";#d#
109	exit 1;	126	exit 1;
110	},	127	},
111	on_read => sub { $self->on_read (@_) },	128	on_read => sub { $self->on_read (@_) },
112	on_eof => $self->{on_eof} \|\| sub { };	129	on_eof => $self->{on_eof} \|\| sub { };
113		130
114	Scalar::Util::weaken ($self->{hdl}{fcp} = $self);	131	Scalar::Util::weaken ($self->{hdl}{fcp} = $self);
115	}	132	}
116		133
117	$self->send_msg (	134	$self->send_msg (client_hello =>
118	client_hello =>
119	name => $self->{name},	135	name => $self->{name},
120	expected_version => "2.0",	136	expected_version => "2.0",
121	);	137	);
122		138
123	$self	139	$self
124	}	140	}
125		141
126	#sub progress {	142	sub identifier {
127	# my ($self, $txn, $type, $attr) = @_;	143	$_[0]{prefix} . ++$_[0]{idseq}
128	#	144	}
129	# $self->{progress}->($self, $txn, $type, $attr)
130	# if $self->{progress};
131	#}
132		145
133	sub send_msg {	146	sub send_msg {
134	my ($self, $type, %kv) = @_;	147	my ($self, $type, %kv) = @_;
135		148
136	my $data = delete $kv{data};	149	my $data = delete $kv{data};
137		150
138	if (exists $kv{id_cb}) {	151	if (exists $kv{id_cb}) {
139	my $id = $kv{identifier} \|\| ++$self->{id};	152	my $id = $kv{identifier} \|\|= $self->identifier;
140	$self->{id}{$id} = delete $kv{id_cb};	153	$self->{id}{$id} = delete $kv{id_cb};
141	$kv{identifier} = $id;
142	}	154	}
143		155
144	my $msg = (touc $type) . "\012"	156	my $msg = (touc $type) . "\012"
145	. join "", map +(touc $_) . "=$kv{$_}\012", keys %kv;	157	. join "", map +(touc $_) . "=$kv{$_}\012", keys %kv;
146		158
…		…
158	}	170	}
159		171
160	$self->{hdl}->push_write ($msg);	172	$self->{hdl}->push_write ($msg);
161	}	173	}
162		174
		175	sub on {
		176	my ($self, $cb) = @_;
		177
		178	# cb return undef - message eaten, remove cb
		179	# cb return 0 - message eaten
		180	# cb return 1 - pass to next
		181
		182	push @{ $self->{on} }, $cb;
		183	}
		184
		185	sub _push_queue {
		186	my ($self, $queue) = @_;
		187
		188	shift @$queue;
		189	$queue->[0]($self, AnyEvent::Util::guard { $self->_push_queue ($queue) })
		190	if @$queue;
		191	}
		192
		193	# lock so only one $type (arbitrary string) is in flight,
		194	# to work around horribly misdesigned protocol.
		195	sub serialise {
		196	my ($self, $type, $cb) = @_;
		197
		198	my $queue = $self->{serialise}{$type} \|\|= [];
		199	push @$queue, $cb;
		200	$cb->($self, AnyEvent::Util::guard { $self->_push_queue ($queue) })
		201	unless $#$queue;
		202	}
		203
		204	# how to merge these types into $self->{persistent}
		205	our %PERSISTENT_TYPE = (
		206	persistent_get => sub { %{ $_[1] } = (type => "persistent_get" , %{ $_[2] }) },
		207	persistent_put => sub { %{ $_[1] } = (type => "persistent_put" , %{ $_[2] }) },
		208	persistent_put_dir => sub { %{ $_[1] } = (type => "persistent_put_dir", %{ $_[2] }) },
		209	persistent_request_modified => sub { %{ $_[1] } = (%{ $_[1] }, %{ $_[2] }) },
		210	persistent_request_removed => sub { delete $_[0]{req}{$_[2]{identifier}} },
		211
		212	simple_progress => sub { $_[1]{simple_progress} = $_[2] }, # get/put
		213
		214	uri_generated => sub { $_[1]{uri_generated} = $_[2] }, # put
		215	generated_metadata => sub { $_[1]{generated_metadata} = $_[2] }, # put
		216	started_compression => sub { $_[1]{started_compression} = $_[2] }, # put
		217	finished_compression => sub { $_[1]{finished_compression} = $_[2] }, # put
		218	put_fetchable => sub { $_[1]{put_fetchable} = $_[2] }, # put
		219	put_failed => sub { $_[1]{put_failed} = $_[2] }, # put
		220	put_successful => sub { $_[1]{put_successful} = $_[2] }, # put
		221
		222	sending_to_network => sub { $_[1]{sending_to_network} = $_[2] }, # get
		223	compatibility_mode => sub { $_[1]{compatibility_mode} = $_[2] }, # get
		224	expected_hashes => sub { $_[1]{expected_hashes} = $_[2] }, # get
		225	expected_mime => sub { $_[1]{expected_mime} = $_[2] }, # get
		226	expected_data_length => sub { $_[1]{expected_data_length} = $_[2] }, # get
		227	get_failed => sub { $_[1]{get_failed} = $_[2] }, # get
		228	data_found => sub { $_[1]{data_found} = $_[2] }, # get
		229	enter_finite_cooldown => sub { $_[1]{enter_finite_cooldown} = $_[2] }, # get
		230	);
		231
		232	sub recv {
		233	my ($self, $type, $kv, @extra) = @_;
		234
		235	if (my $cb = $PERSISTENT_TYPE{$type}) {
		236	my $id = $kv->{identifier};
		237	my $req = $_[0]{req}{$id} \|\|= {};
		238	$cb->($self, $req, $kv);
		239	$self->recv (request_changed => $kv, $type, @extra);
		240	}
		241
		242	my $on = $self->{on};
		243	for (0 .. $#$on) {
		244	unless (my $res = $on->[$_]($self, $type, $kv, @extra)) {
		245	splice @$on, $_, 1 unless defined $res;
		246	return;
		247	}
		248	}
		249
		250	if (my $cb = $self->{queue}[0]) {
		251	$cb->($self, $type, $kv, @extra)
		252	and shift @{ $self->{queue} };
		253	} else {
		254	$self->default_recv ($type, $kv, @extra);
		255	}
		256	}
		257
163	sub on_read {	258	sub on_read {
164	my ($self) = @_;	259	my ($self) = @_;
165		260
166	my $type;	261	my ($k, $v, $type);
167	my %kv;	262	my %kv;
168	my $rdata;	263	my $rdata;
169		264
170	my $done_cb = sub {
171	$kv{pkt_type} = $type;
172
173	if (my $cb = $self->{queue}[0]) {
174	$cb->($self, $type, \%kv, $rdata)
175	and shift @{ $self->{queue} };
176	} else {
177	$self->default_recv ($type, \%kv, $rdata);
178	}
179	};
180
181	my $hdr_cb; $hdr_cb = sub {	265	my $hdr_cb; $hdr_cb = sub {
182	if ($_[1] =~ /^([^=]+)=(.*)$/) {	266	if (($v = index $_[1], "=") >= 0) {
183	my ($k, $v) = ($1, $2);	267	$k = substr $_[1], 0, $v;
		268	$v = substr $_[1], $v + 1;
		269	$k = ($TOLC{$k} \|\|= tolc $k);
		270
		271	if ($k !~ /\./) {
		272	# special case common case, for performance only
		273	$kv{$k} = $v;
		274	} else {
184	my @k = split /\./, tolc $k;	275	my @k = split /\./, $k;
185	my $ro = \\%kv;	276	my $ro = \\%kv;
186		277
187	while (@k) {	278	while (@k) {
188	my $k = shift @k;	279	$k = shift @k;
189	if ($k =~ /^\d+$/) {	280	if ($k =~ /^\d+$/) {
190	$ro = \$$ro->[$k];	281	$ro = \$$ro->[$k];
191	} else {	282	} else {
192	$ro = \$$ro->{$k};	283	$ro = \$$ro->{$k};
		284	}
193	}	285	}
		286
		287	$$ro = $v;
194	}	288	}
195
196	$$ro = $v;
197		289
198	$_[0]->push_read (line => $hdr_cb);	290	$_[0]->push_read (line => $hdr_cb);
199	} elsif ($_[1] eq "Data") {	291	} elsif ($_[1] eq "Data") {
200	$_[0]->push_read (chunk => delete $kv{data_length}, sub {	292	$_[0]->push_read (chunk => delete $kv{data_length}, sub {
201	$rdata = \$_[1];	293	$rdata = \$_[1];
202	$done_cb->();	294	$self->recv ($type, \%kv, $rdata);
203	});	295	});
204	} elsif ($_[1] eq "EndMessage") {	296	} elsif ($_[1] eq "EndMessage") {
205	$done_cb->();	297	$self->recv ($type, \%kv);
206	} else {	298	} else {
207	die "protocol error, expected message end, got $_[1]\n";#d#	299	die "protocol error, expected message end, got $_[1]\n";#d#
208	}	300	}
209	};	301	};
210		302
211	$self->{hdl}->push_read (line => sub {	303	$self->{hdl}->push_read (line => sub {
212	$type = tolc $_[1];	304	$type = ($TOLC{$_[1]} \|\|= tolc $_[1]);
213	$_[0]->push_read (line => $hdr_cb);	305	$_[0]->push_read (line => $hdr_cb);
214	});	306	});
215	}	307	}
216		308
217	sub default_recv {	309	sub default_recv {
…		…
220	if ($type eq "node_hello") {	312	if ($type eq "node_hello") {
221	$self->{node_hello} = $kv;	313	$self->{node_hello} = $kv;
222	} elsif (exists $self->{id}{$kv->{identifier}}) {	314	} elsif (exists $self->{id}{$kv->{identifier}}) {
223	$self->{id}{$kv->{identifier}}($self, $type, $kv, $rdata)	315	$self->{id}{$kv->{identifier}}($self, $type, $kv, $rdata)
224	and delete $self->{id}{$kv->{identifier}};	316	and delete $self->{id}{$kv->{identifier}};
225	} else {
226	# on_warn
227	#warn "protocol warning (unexpected $type message)\n";
228	}	317	}
229	}	318	}
		319
		320	=back
		321
		322	=head2 FCP REQUESTS
		323
		324	The following methods implement various requests. Most of them map
		325	directory to the FCP message of the same name. The added benefit of
		326	these over sending requests yourself is that they handle the necessary
		327	serialisation, protocol quirks, and replies.
		328
		329	All of them exist in two versions, the variant shown in this manpage, and
		330	a variant with an extra C<_> at the end, and an extra C<$cb> argument. The
		331	version as shown is I<synchronous> - it will wait for any replies, and
		332	either return the reply, or croak with an error. The underscore variant
		333	returns immediately and invokes one or more callbacks or condvars later.
		334
		335	For example, the call
		336
		337	$info = $fcp->get_plugin_info ($name, $detailed);
		338
		339	Also comes in this underscore variant:
		340
		341	$fcp->get_plugin_info_ ($name, $detailed, $cb);
		342
		343	You can thinbk of the underscore as a kind of continuation indicator - the
		344	normal function waits and returns with the data, the C<_> indicates that
		345	you pass the continuation yourself, and the continuation will be invoked
		346	with the results.
		347
		348	This callback/continuation argument (C<$cb>) can come in three forms itself:
		349
		350	=over 4
		351
		352	=item A code reference (or rather anything not matching some other alternative)
		353
		354	This code reference will be invoked with the result on success. On an
		355	error, it will die (in the event loop) with a backtrace of the call site.
		356
		357	This is a popular choice, but it makes handling errors hard - make sure
		358	you never generate protocol errors!
		359
		360	=item A condvar (as returned by e.g. C<< AnyEvent->condvar >>)
		361
		362	When a condvar is passed, it is sent (C<< $cv->send ($results) >>) the
		363	results when the request has finished. Should an error occur, the error
		364	will instead result in C<< $cv->croak ($error) >>.
		365
		366	This is also a popular choice.
		367
		368	=item An array with two callbacks C<[$success, $failure]>
		369
		370	The C<$success> callback will be invoked with the results, while the
		371	C<$failure> callback will be invoked on any errors.
		372
		373	=item C<undef>
		374
		375	This is the same thing as specifying C<sub { }> as callback, i.e. on
		376	success, the results are ignored, while on failure, you the module dies
		377	with a backtrace.
		378
		379	This is good for quick scripts, or when you really aren't interested in
		380	the results.
		381
		382	=back
		383
		384	=cut
		385
		386	our $NOP_CB = sub { };
230		387
231	sub _txn {	388	sub _txn {
232	my ($name, $sub) = @_;	389	my ($name, $sub) = @_;
233		390
234	*{$name} = sub {	391	*{$name} = sub {
235	splice @_, 1, 0, (my $cv = AnyEvent->condvar);	392	my $cv = AE::cv;
236	&$sub;
237	$cv
238	};
239		393
240	*{"$name\_sync"} = sub {	394	splice @_, 1, 0, $cv, sub { $cv->croak ($_[0]{extra_description}) };
241	splice @_, 1, 0, (my $cv = AnyEvent->condvar);
242	&$sub;	395	&$sub;
243	$cv->recv	396	$cv->recv
244	};	397	};
245	}
246		398
247	=item $cv = $fcp->list_peers ([$with_metdata[, $with_volatile]])	399	*{"$name\_"} = sub {
		400	my ($ok, $err) = pop;
248		401
		402	if (ARRAY:: eq ref $ok) {
		403	($ok, $err) = @$ok;
		404	} elsif (UNIVERSAL::isa $ok, AnyEvent::CondVar::) {
		405	$err = sub { $ok->croak ($_[0]{extra_description}) };
		406	} else {
		407	my $bt = Carp::longmess "";
		408	$err = sub {
		409	die "$_[0]{code_description} ($_[0]{extra_description})$bt";
		410	};
		411	}
		412
		413	$ok \|\|= $NOP_CB;
		414
		415	splice @_, 1, 0, $ok, $err;
		416	&$sub;
		417	};
		418	}
		419
		420	=over 4
		421
249	=item $peers = $fcp->list_peers_sync ([$with_metdata[, $with_volatile]])	422	=item $peers = $fcp->list_peers ([$with_metdata[, $with_volatile]])
250		423
251	=cut	424	=cut
252		425
253	_txn list_peers => sub {	426	_txn list_peers => sub {
254	my ($self, $cv, $with_metadata, $with_volatile) = @_;	427	my ($self, $ok, undef, $with_metadata, $with_volatile) = @_;
255		428
256	my @res;	429	my @res;
257		430
258	$self->send_msg (list_peers =>	431	$self->send_msg (list_peers =>
259	with_metadata => $with_metadata ? "true" : "false",	432	with_metadata => $with_metadata ? "true" : "false",
260	with_volatile => $with_volatile ? "true" : "false",	433	with_volatile => $with_volatile ? "true" : "false",
261	id_cb => sub {	434	id_cb => sub {
262	my ($self, $type, $kv, $rdata) = @_;	435	my ($self, $type, $kv, $rdata) = @_;
263		436
264	if ($type eq "end_list_peers") {	437	if ($type eq "end_list_peers") {
265	$cv->(\@res);	438	$ok->(\@res);
266	1	439	1
267	} else {	440	} else {
268	push @res, $kv;	441	push @res, $kv;
269	0	442	0
270	}	443	}
271	},	444	},
272	);	445	);
273	};	446	};
274		447
275	=item $cv = $fcp->list_peer_notes ($node_identifier)
276
277	=item $notes = $fcp->list_peer_notes_sync ($node_identifier)	448	=item $notes = $fcp->list_peer_notes ($node_identifier)
278		449
279	=cut	450	=cut
280		451
281	_txn list_peer_notes => sub {	452	_txn list_peer_notes => sub {
282	my ($self, $cv, $node_identifier) = @_;	453	my ($self, $ok, undef, $node_identifier) = @_;
283		454
284	$self->send_msg (list_peer_notes =>	455	$self->send_msg (list_peer_notes =>
285	node_identifier => $node_identifier,	456	node_identifier => $node_identifier,
286	id_cb => sub {	457	id_cb => sub {
287	my ($self, $type, $kv, $rdata) = @_;	458	my ($self, $type, $kv, $rdata) = @_;
288		459
289	$cv->($kv);	460	$ok->($kv);
290	1	461	1
291	},	462	},
292	);	463	);
293	};	464	};
294		465
295	=item $cv = $fcp->watch_global ($enabled[, $verbosity_mask])
296
297	=item $fcp->watch_global_sync ($enabled[, $verbosity_mask])	466	=item $fcp->watch_global ($enabled[, $verbosity_mask])
298		467
299	=cut	468	=cut
300		469
301	_txn watch_global => sub {	470	_txn watch_global => sub {
302	my ($self, $cv, $enabled, $verbosity_mask) = @_;	471	my ($self, $ok, $err, $enabled, $verbosity_mask) = @_;
303		472
304	$self->send_msg (watch_global =>	473	$self->send_msg (watch_global =>
305	enabled => $enabled ? "true" : "false",	474	enabled => $enabled ? "true" : "false",
306	defined $verbosity_mask ? (verbosity_mask => $verbosity_mask+0) : (),	475	defined $verbosity_mask ? (verbosity_mask => $verbosity_mask+0) : (),
307	);	476	);
308		477
309	$cv->();	478	$ok->();
310	};	479	};
311		480
312	=item $cv = $fcp->list_persistent_requests
313
314	=item $reqs = $fcp->list_persistent_requests_sync	481	=item $reqs = $fcp->list_persistent_requests
315		482
316	=cut	483	=cut
317		484
318	_txn list_persistent_requests => sub {	485	_txn list_persistent_requests => sub {
319	my ($self, $cv) = @_;	486	my ($self, $ok, $err) = @_;
320		487
		488	$self->serialise (list_persistent_requests => sub {
		489	my ($self, $guard) = @_;
		490
321	my %res;	491	my @res;
322		492
323	$self->send_msg ("list_persistent_requests");	493	$self->send_msg ("list_persistent_requests");
324		494
325	push @{ $self->{queue} }, sub {	495	$self->on (sub {
326	my ($self, $type, $kv, $rdata) = @_;	496	my ($self, $type, $kv, $rdata) = @_;
327		497
		498	$guard if 0;
		499
328	if ($type eq "end_list_persistent_requests") {	500	if ($type eq "end_list_persistent_requests") {
329	$cv->(\%res);	501	$ok->(\@res);
		502	return;
		503	} else {
		504	my $id = $kv->{identifier};
		505
		506	if ($type =~ /^persistent_(get\|put\|put_dir)$/) {
		507	push @res, [$type, $kv];
		508	}
		509	}
		510
330	1	511	1
331	} else {	512	});
332	my $id = $kv->{identifier};	513	});
		514	};
333		515
334	if ($type =~ /^persistent_(get\|put\|put_dir)$/) {	516	=item $sync = $fcp->modify_persistent_request ($global, $identifier[, $client_token[, $priority_class]])
335	$res{$id} = {	517
336	type => $1,	518	Update either the C<client_token> or C<priority_class> of a request
337	%{ $res{$id} },	519	identified by C<$global> and C<$identifier>, depending on which of
		520	C<$client_token> and C<$priority_class> are not C<undef>.
		521
		522	=cut
		523
		524	_txn modify_persistent_request => sub {
		525	my ($self, $ok, $err, $global, $identifier, $client_token, $priority_class) = @_;
		526
		527	$self->serialise ($identifier => sub {
		528	my ($self, $guard) = @_;
		529
		530	$self->send_msg (modify_persistent_request =>
		531	global => $global ? "true" : "false",
		532	identifier => $identifier,
		533	defined $client_token ? (client_token => $client_token ) : (),
		534	defined $priority_class ? (priority_class => $priority_class) : (),
		535	);
		536
		537	$self->on (sub {
		538	my ($self, $type, $kv, @extra) = @_;
		539
		540	$guard if 0;
		541
		542	if ($kv->{identifier} eq $identifier) {
		543	if ($type eq "persistent_request_modified") {
338	%$kv,	544	$ok->($kv);
		545	return;
		546	} elsif ($type eq "protocol_error") {
		547	$err->($kv);
		548	return;
339	};	549	}
340	} elsif ($type eq "simple_progress") {
341	delete $kv->{pkt_type}; # save memory
342	push @{ $res{delete $kv->{identifier}}{simple_progress} }, $kv;
343	} else {
344	$res{delete $kv->{identifier}}{delete $kv->{pkt_type}} = $kv;
345	}	550	}
		551
346	0	552	1
347	}	553	});
348	};	554	});
349	};	555	};
350		556
351	=item $cv = $fcp->remove_request ($global, $identifier)
352
353	=item $status = $fcp->remove_request_sync ($global, $identifier)
354
355	=cut
356
357	_txn remove_request => sub {
358	my ($self, $cv, $global, $identifier) = @_;
359
360	$self->send_msg (remove_request =>
361	global => $global ? "true" : "false",
362	identifier => $identifier,
363	id_cb => sub {
364	my ($self, $type, $kv, $rdata) = @_;
365
366	$cv->($kv);
367	1
368	},
369	);
370	};
371
372	=item $cv = $fcp->modify_persistent_request ($global, $identifier[, $client_token[, $priority_class]])
373
374	=item $sync = $fcp->modify_persistent_request_sync ($global, $identifier[, $client_token[, $priority_class]])
375
376	=cut
377
378	_txn modify_persistent_request => sub {
379	my ($self, $cv, $global, $identifier, $client_token, $priority_class) = @_;
380
381	$self->send_msg (modify_persistent_request =>
382	global => $global ? "true" : "false",
383	defined $client_token ? (client_token => $client_token ) : (),
384	defined $priority_class ? (priority_class => $priority_class) : (),
385	identifier => $identifier,
386	id_cb => sub {
387	my ($self, $type, $kv, $rdata) = @_;
388
389	$cv->($kv);
390	1
391	},
392	);
393	};
394
395	=item $cv = $fcp->get_plugin_info ($name, $detailed)
396
397	=item $info = $fcp->get_plugin_info_sync ($name, $detailed)	557	=item $info = $fcp->get_plugin_info ($name, $detailed)
398		558
399	=cut	559	=cut
400		560
401	_txn get_plugin_info => sub {	561	_txn get_plugin_info => sub {
402	my ($self, $cv, $name, $detailed) = @_;	562	my ($self, $ok, $err, $name, $detailed) = @_;
		563
		564	my $id = $self->identifier;
403		565
404	$self->send_msg (get_plugin_info =>	566	$self->send_msg (get_plugin_info =>
		567	identifier => $id,
405	plugin_name => $name,	568	plugin_name => $name,
406	detailed => $detailed ? "true" : "false",	569	detailed => $detailed ? "true" : "false",
407	id_cb => sub {
408	my ($self, $type, $kv, $rdata) = @_;
409
410	$cv->($kv);
411	1
412	},
413	);	570	);
		571	$self->on (sub {
		572	my ($self, $type, $kv) = @_;
		573
		574	if ($kv->{identifier} eq $id) {
		575	if ($type eq "get_plugin_info") {
		576	$ok->($kv);
		577	} else {
		578	$err->($kv, $type);
		579	}
		580	return;
		581	}
		582
		583	1
		584	});
414	};	585	};
415		586
416	=item $cv = $fcp->client_get ($uri, $identifier, %kv)
417
418	=item $status = $fcp->client_get_sync ($uri, $identifier, %kv)	587	=item $status = $fcp->client_get ($uri, $identifier, %kv)
419		588
420	%kv can contain (L<http://wiki.freenetproject.org/FCP2p0ClientGet>).	589	%kv can contain (L<http://wiki.freenetproject.org/FCP2p0ClientGet>).
421		590
422	ignore_ds, ds_only, verbosity, max_size, max_temp_size, max_retries,	591	ignore_ds, ds_only, verbosity, max_size, max_temp_size, max_retries,
423	priority_class, persistence, client_token, global, return_type,	592	priority_class, persistence, client_token, global, return_type,
424	binary_blob, allowed_mime_types, filename, temp_filename	593	binary_blob, allowed_mime_types, filename, temp_filename
425		594
426	=cut	595	=cut
427		596
428	_txn client_get => sub {	597	_txn client_get => sub {
429	my ($self, $cv, $uri, $identifier, %kv) = @_;	598	my ($self, $ok, $err, $uri, $identifier, %kv) = @_;
430		599
		600	$self->serialise ($identifier => sub {
		601	my ($self, $guard) = @_;
		602
431	$self->send_msg (client_get =>	603	$self->send_msg (client_get =>
432	%kv,	604	%kv,
433	uri => $uri,	605	uri => $uri,
434	identifier => $identifier,	606	identifier => $identifier,
435	id_cb => sub {	607	);
		608
		609	$self->on (sub {
436	my ($self, $type, $kv, $rdata) = @_;	610	my ($self, $type, $kv, @extra) = @_;
437		611
		612	$guard if 0;
		613
		614	if ($kv->{identifier} eq $identifier) {
		615	if ($type eq "persistent_get") {
438	$cv->($kv);	616	$ok->($kv);
		617	return;
		618	} elsif ($type eq "protocol_error") {
		619	$err->($kv);
		620	return;
		621	}
		622	}
		623
439	1	624	1
		625	});
		626	});
		627	};
		628
		629	=item $status = $fcp->remove_request ($identifier[, $global])
		630
		631	Remove the request with the given isdentifier. Returns true if successful,
		632	false on error.
		633
		634	=cut
		635
		636	_txn remove_request => sub {
		637	my ($self, $ok, $err, $identifier, $global) = @_;
		638
		639	$self->serialise ($identifier => sub {
		640	my ($self, $guard) = @_;
		641
		642	$self->send_msg (remove_request =>
		643	identifier => $identifier,
		644	global => $global ? "true" : "false",
		645	);
		646	$self->on (sub {
		647	my ($self, $type, $kv, @extra) = @_;
		648
		649	$guard if 0;
		650
		651	if ($kv->{identifier} eq $identifier) {
		652	if ($type eq "persistent_request_removed") {
		653	$ok->(1);
		654	return;
		655	} elsif ($type eq "protocol_error") {
		656	$err->($kv);
		657	return;
		658	}
		659	}
		660
		661	1
		662	});
		663	});
		664	};
		665
		666	=item ($can_read, $can_write) = $fcp->test_dda ($local_directory, $remote_directory, $want_read, $want_write))
		667
		668	The DDA test in FCP is probably the single most broken protocol - only
		669	one directory test can be outstanding at any time, and some guessing and
		670	heuristics are involved in mangling the paths.
		671
		672	This function combines C<TestDDARequest> and C<TestDDAResponse> in one
		673	request, handling file reading and writing as well, and tries very hard to
		674	do the right thing.
		675
		676	Both C<$local_directory> and C<$remote_directory> must specify the same
		677	directory - C<$local_directory> is the directory path on the client (where
		678	L<AnyEvent::FCP> runs) and C<$remote_directory> is the directory path on
		679	the server (where the freenet node runs). When both are running on the
		680	same node, the paths are generally identical.
		681
		682	C<$want_read> and C<$want_write> should be set to a true value when you
		683	want to read (get) files or write (put) files, respectively.
		684
		685	On error, an exception is thrown. Otherwise, C<$can_read> and
		686	C<$can_write> indicate whether you can reaqd or write to freenet via the
		687	directory.
		688
		689	=cut
		690
		691	_txn test_dda => sub {
		692	my ($self, $ok, $err, $local, $remote, $want_read, $want_write) = @_;
		693
		694	$self->serialise (test_dda => sub {
		695	my ($self, $guard) = @_;
		696
		697	$self->send_msg (test_dda_request =>
		698	directory => $remote,
		699	want_read_directory => $want_read ? "true" : "false",
		700	want_write_directory => $want_write ? "true" : "false",
		701	);
		702	$self->on (sub {
		703	my ($self, $type, $kv) = @_;
		704
		705	if ($type eq "test_dda_reply") {
		706	# the filenames are all relative to the server-side directory,
		707	# which might or might not match $remote anymore, so we
		708	# need to rewrite the paths to be relative to $local
		709	for my $k (qw(read_filename write_filename)) {
		710	my $f = $kv->{$k};
		711	for my $dir ($kv->{directory}, $remote) {
		712	if ($dir eq substr $f, 0, length $dir) {
		713	substr $f, 0, 1 + length $dir, "";
		714	$kv->{$k} = $f;
		715	last;
		716	}
		717	}
		718	}
		719
		720	my %response = (directory => $remote);
		721
		722	if (length $kv->{read_filename}) {
		723	if (open my $fh, "<:raw", "$local/$kv->{read_filename}") {
		724	sysread $fh, my $buf, -s $fh;
		725	$response{read_content} = $buf;
		726	}
		727	}
		728
		729	if (length $kv->{write_filename}) {
		730	if (open my $fh, ">:raw", "$local/$kv->{write_filename}") {
		731	syswrite $fh, $kv->{content_to_write};
		732	}
		733	}
		734
		735	$self->send_msg (test_dda_response => %response);
		736
		737	$self->on (sub {
		738	my ($self, $type, $kv) = @_;
		739
		740	$guard if 0; # reference
		741
		742	if ($type eq "test_dda_complete") {
		743	$ok->(
		744	$kv->{read_directory_allowed} eq "true",
		745	$kv->{write_directory_allowed} eq "true",
		746	);
		747	} elsif ($type eq "protocol_error" && $kv->{identifier} eq $remote) {
		748	$err->($kv->{extra_description});
		749	return;
		750	}
		751
		752	1
		753	});
		754
		755	return;
		756	} elsif ($type eq "protocol_error" && $kv->{identifier} eq $remote) {
		757	$err->($kv);
		758	return;
		759	}
		760
		761	1
		762	});
		763	});
		764	};
		765
		766	=back
		767
		768	=head2 REQUEST CACHE
		769
		770	The C<AnyEvent::FCP> class keeps a request cache, where it caches all
		771	information from requests.
		772
		773	For these messages, it will store a copy of the key-value pairs, together with a C<type> slot,
		774	in C<< $fcp->{req}{$identifier} >>:
		775
		776	persistent_get
		777	persistent_put
		778	persistent_put_dir
		779
		780	This message updates the stored data:
		781
		782	persistent_request_modified
		783
		784	This message will remove this entry:
		785
		786	persistent_request_removed
		787
		788	These messages get merged into the cache entry, under their
		789	type, i.e. a C<simple_progress> message will be stored in C<<
		790	$fcp->{req}{$identifier}{simple_progress} >>:
		791
		792	simple_progress # get/put
		793
		794	uri_generated # put
		795	generated_metadata # put
		796	started_compression # put
		797	finished_compression # put
		798	put_failed # put
		799	put_fetchable # put
		800	put_successful # put
		801
		802	sending_to_network # get
		803	compatibility_mode # get
		804	expected_hashes # get
		805	expected_mime # get
		806	expected_data_length # get
		807	get_failed # get
		808	data_found # get
		809	enter_finite_cooldown # get
		810
		811	In addition, an event (basically a fake message) of type C<request_changed> is generated
		812	on every change, which will be called as C<< $cb->($fcp, $kv, $type) >>, where C<$type>
		813	is the type of the original message triggering the change,
		814
		815	To fill this cache with the global queue and keep it updated,
		816	call C<watch_global> to subscribe to updates, followed by
		817	C<list_persistent_requests_sync>.
		818
		819	$fcp->watch_global_sync_; # do not wait
		820	$fcp->list_persistent_requests; # wait
		821
		822	To get a better idea of what is stored in the cache, here is an example of
		823	what might be stored in C<< $fcp->{req}{"Frost-gpl.txt"} >>:
		824
		825	{
		826	identifier => "Frost-gpl.txt",
		827	uri => 'CHK@Fnx5kzdrfE,EImdzaVyEWl,AAIC--8/gpl.txt',
		828	binary_blob => "false",
		829	global => "true",
		830	max_retries => -1,
		831	max_size => 9223372036854775807,
		832	persistence => "forever",
		833	priority_class => 3,
		834	real_time => "false",
		835	return_type => "direct",
		836	started => "true",
		837	type => "persistent_get",
		838	verbosity => 2147483647,
		839	sending_to_network => {
		840	identifier => "Frost-gpl.txt",
		841	global => "true",
440	},	842	},
441	);	843	compatibility_mode => {
442	};	844	identifier => "Frost-gpl.txt",
443		845	definitive => "true",
444	=back	846	dont_compress => "false",
		847	global => "true",
		848	max => "COMPAT_1255",
		849	min => "COMPAT_1255",
		850	},
		851	expected_hashes => {
		852	identifier => "Frost-gpl.txt",
		853	global => "true",
		854	hashes => {
		855	ed2k => "d83596f5ee3b7...",
		856	md5 => "e0894e4a2a6...",
		857	sha1 => "...",
		858	sha256 => "...",
		859	sha512 => "...",
		860	tth => "...",
		861	},
		862	},
		863	expected_mime => {
		864	identifier => "Frost-gpl.txt",
		865	global => "true",
		866	metadata => { content_type => "application/rar" },
		867	},
		868	expected_data_length => {
		869	identifier => "Frost-gpl.txt",
		870	data_length => 37576,
		871	global => "true",
		872	},
		873	simple_progress => {
		874	identifier => "Frost-gpl.txt",
		875	failed => 0,
		876	fatally_failed => 0,
		877	finalized_total => "true",
		878	global => "true",
		879	last_progress => 1438639282628,
		880	required => 372,
		881	succeeded => 102,
		882	total => 747,
		883	},
		884	data_found => {
		885	identifier => "Frost-gpl.txt",
		886	completion_time => 1438663354026,
		887	data_length => 37576,
		888	global => "true",
		889	metadata => { content_type => "image/jpeg" },
		890	startup_time => 1438657196167,
		891	},
		892	}
445		893
446	=head1 EXAMPLE PROGRAM	894	=head1 EXAMPLE PROGRAM
447		895
448	use AnyEvent::FCP;	896	use AnyEvent::FCP;
449		897
450	my $fcp = new AnyEvent::FCP;	898	my $fcp = new AnyEvent::FCP;
451		899
452	# let us look at the global request list	900	# let us look at the global request list
453	$fcp->watch_global (1, 0);	901	$fcp->watch_global_ (1);
454		902
455	# list them, synchronously	903	# list them, synchronously
456	my $req = $fcp->list_persistent_requests_sync;	904	my $req = $fcp->list_persistent_requests;
457		905
458	# go through all requests	906	# go through all requests
		907	TODO
459	for my $req (values %$req) {	908	for my $req (values %$req) {
460	# skip jobs not directly-to-disk	909	# skip jobs not directly-to-disk
461	next unless $req->{return_type} eq "disk";	910	next unless $req->{return_type} eq "disk";
462	# skip jobs not issued by FProxy	911	# skip jobs not issued by FProxy
463	next unless $req->{identifier} =~ /^FProxy:/;	912	next unless $req->{identifier} =~ /^FProxy:/;

Diff Legend

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

Comparing AnyEvent-FCP/FCP.pm (file contents): Revision 1.5 by root, Tue Dec 1 13:49:56 2009 UTC vs. Revision 1.16 by root, Sat Sep 5 13:26:47 2015 UTC

Diff Legend

Comparing AnyEvent-FCP/FCP.pm (file contents):
Revision 1.5 by root, Tue Dec 1 13:49:56 2009 UTC vs.
Revision 1.16 by root, Sat Sep 5 13:26:47 2015 UTC