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

Comparing AnyEvent-FCP/FCP.pm (file contents):
Revision 1.8 by root, Fri Jun 18 16:59:13 2010 UTC vs.
Revision 1.16 by root, Sat Sep 5 13:26:47 2015 UTC

…		…
35		35
36	use AnyEvent::FCP;	36	use AnyEvent::FCP;
37		37
38	my $fcp = new AnyEvent::FCP;	38	my $fcp = new AnyEvent::FCP;
39		39
40	$fcp->watch_global_sync (1, 0);	40	$fcp->watch_global (1, 0);
41	my $req = $fcp->list_persistent_requests_sync;	41	my $req = $fcp->list_persistent_requests;
42		42
		43	TODO
43	for my $req (values %$req) {	44	for my $req (values %$req) {
44	if ($req->{filename} =~ /a/) {	45	if ($req->{filename} =~ /a/) {
45	$fcp->modify_persistent_request_sync (1, $req->{identifier}, undef, 0);	46	$fcp->modify_persistent_request (1, $req->{identifier}, undef, 0);
46	}	47	}
47	}	48	}
48		49
49	=head2 IMPORT TAGS	50	=head2 IMPORT TAGS
50		51
51	Nothing much can be "imported" from this module right now.	52	Nothing much can be "imported" from this module right now.
52		53
53	=head2 THE AnyEvent::FCP CLASS	54	=head1 THE AnyEvent::FCP CLASS
54		55
55	=over 4	56	=over 4
56		57
57	=cut	58	=cut
58		59
…		…
66		67
67	use Scalar::Util ();	68	use Scalar::Util ();
68		69
69	use AnyEvent;	70	use AnyEvent;
70	use AnyEvent::Handle;	71	use AnyEvent::Handle;
		72	use AnyEvent::Util ();
		73
		74	our %TOLC; # tolc cache
71		75
72	sub touc($) {	76	sub touc($) {
73	local $_ = shift;	77	local $_ = shift;
74	1 while s/((?:^\|_)(?:svk\|chk\|uri\|fcp\|ds\|mime)(?:_\|$))/\U$1/;	78	1 while s/((?:^\|_)(?:svk\|chk\|uri\|fcp\|ds\|mime\|dda)(?:_\|$))/\U$1/;
75	s/(?:^\|_)(.)/\U$1/g;	79	s/(?:^\|_)(.)/\U$1/g;
76	$_	80	$_
77	}	81	}
78		82
79	sub tolc($) {	83	sub tolc($) {
80	local $_ = shift;	84	local $_ = shift;
81	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/;
82	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/;
83	s/(?<=[a-z])(?=[A-Z])/_/g;	87	s/(?<=[a-z])(?=[A-Z])/_/g;
84	lc	88	lc
85	}	89	}
86		90
87	=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]
88		92
89	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
90	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>).
91		95
92	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
93	(hopefully) unique client name for you.	97	(hopefully) unique client name for you.
94		98
95	You can install a progress callback that is being called with the AnyEvent::FCP
96	object, the type, a hashref with key-value pairs and a reference to any received data,
97	for all unsolicited messages.
98
99	Example:
100
101	sub progress_cb {
102	my ($self, $type, $kv, $rdata) = @_;
103
104	if ($type eq "simple_progress") {
105	warn "$kv->{identifier} $kv->{succeeded}/$kv->{required}\n";
106	}
107	}
108
109	=cut	99	=cut
110		100
111	sub new {	101	sub new {
112	my $class = shift;	102	my $class = shift;
		103
		104	my $rand = join "", map chr 0x21 + rand 94, 1..40; # ~ 262 bits entropy
		105
113	my $self = bless { @_ }, $class;	106	my $self = bless {
114		107	host => $ENV{FREDHOST} \|\| "127.0.0.1",
115	$self->{host} \|\|= $ENV{FREDHOST} \|\| "127.0.0.1";	108	port => $ENV{FREDPORT} \|\| 9481,
116	$self->{port} \|\|= $ENV{FREDPORT} \|\| 9481;	109	timeout => 3600 * 2,
117	$self->{name} \|\|= time.rand.rand.rand; # lame	110	name => time.rand.rand.rand, # lame
118	$self->{timeout} \|\|= 3600*2;	111	@_,
119	$self->{progress} \|\|= sub { };	112	queue => [],
120		113	req => {},
121	$self->{id} = "a0";	114	prefix => "..:aefcpid:$rand:",
		115	idseq => "a0",
		116	}, $class;
122		117
123	{	118	{
124	Scalar::Util::weaken (my $self = $self);	119	Scalar::Util::weaken (my $self = $self);
125		120
126	$self->{hdl} = new AnyEvent::Handle	121	$self->{hdl} = new AnyEvent::Handle
…		…
134	on_eof => $self->{on_eof} \|\| sub { };	129	on_eof => $self->{on_eof} \|\| sub { };
135		130
136	Scalar::Util::weaken ($self->{hdl}{fcp} = $self);	131	Scalar::Util::weaken ($self->{hdl}{fcp} = $self);
137	}	132	}
138		133
139	$self->send_msg (	134	$self->send_msg (client_hello =>
140	client_hello =>
141	name => $self->{name},	135	name => $self->{name},
142	expected_version => "2.0",	136	expected_version => "2.0",
143	);	137	);
144		138
145	$self	139	$self
146	}	140	}
147		141
		142	sub identifier {
		143	$_[0]{prefix} . ++$_[0]{idseq}
		144	}
		145
148	sub send_msg {	146	sub send_msg {
149	my ($self, $type, %kv) = @_;	147	my ($self, $type, %kv) = @_;
150		148
151	my $data = delete $kv{data};	149	my $data = delete $kv{data};
152		150
153	if (exists $kv{id_cb}) {	151	if (exists $kv{id_cb}) {
154	my $id = $kv{identifier} \|\| ++$self->{id};	152	my $id = $kv{identifier} \|\|= $self->identifier;
155	$self->{id}{$id} = delete $kv{id_cb};	153	$self->{id}{$id} = delete $kv{id_cb};
156	$kv{identifier} = $id;
157	}	154	}
158		155
159	my $msg = (touc $type) . "\012"	156	my $msg = (touc $type) . "\012"
160	. join "", map +(touc $_) . "=$kv{$_}\012", keys %kv;	157	. join "", map +(touc $_) . "=$kv{$_}\012", keys %kv;
161		158
…		…
173	}	170	}
174		171
175	$self->{hdl}->push_write ($msg);	172	$self->{hdl}->push_write ($msg);
176	}	173	}
177		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
178	sub on_read {	258	sub on_read {
179	my ($self) = @_;	259	my ($self) = @_;
180		260
181	my $type;	261	my ($k, $v, $type);
182	my %kv;	262	my %kv;
183	my $rdata;	263	my $rdata;
184		264
185	my $done_cb = sub {
186	$kv{pkt_type} = $type;
187
188	if (my $cb = $self->{queue}[0]) {
189	$cb->($self, $type, \%kv, $rdata)
190	and shift @{ $self->{queue} };
191	} else {
192	$self->default_recv ($type, \%kv, $rdata);
193	}
194	};
195
196	my $hdr_cb; $hdr_cb = sub {	265	my $hdr_cb; $hdr_cb = sub {
197	if ($_[1] =~ /^([^=]+)=(.*)$/) {	266	if (($v = index $_[1], "=") >= 0) {
198	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 {
199	my @k = split /\./, tolc $k;	275	my @k = split /\./, $k;
200	my $ro = \\%kv;	276	my $ro = \\%kv;
201		277
202	while (@k) {	278	while (@k) {
203	my $k = shift @k;	279	$k = shift @k;
204	if ($k =~ /^\d+$/) {	280	if ($k =~ /^\d+$/) {
205	$ro = \$$ro->[$k];	281	$ro = \$$ro->[$k];
206	} else {	282	} else {
207	$ro = \$$ro->{$k};	283	$ro = \$$ro->{$k};
		284	}
208	}	285	}
		286
		287	$$ro = $v;
209	}	288	}
210
211	$$ro = $v;
212		289
213	$_[0]->push_read (line => $hdr_cb);	290	$_[0]->push_read (line => $hdr_cb);
214	} elsif ($_[1] eq "Data") {	291	} elsif ($_[1] eq "Data") {
215	$_[0]->push_read (chunk => delete $kv{data_length}, sub {	292	$_[0]->push_read (chunk => delete $kv{data_length}, sub {
216	$rdata = \$_[1];	293	$rdata = \$_[1];
217	$done_cb->();	294	$self->recv ($type, \%kv, $rdata);
218	});	295	});
219	} elsif ($_[1] eq "EndMessage") {	296	} elsif ($_[1] eq "EndMessage") {
220	$done_cb->();	297	$self->recv ($type, \%kv);
221	} else {	298	} else {
222	die "protocol error, expected message end, got $_[1]\n";#d#	299	die "protocol error, expected message end, got $_[1]\n";#d#
223	}	300	}
224	};	301	};
225		302
226	$self->{hdl}->push_read (line => sub {	303	$self->{hdl}->push_read (line => sub {
227	$type = tolc $_[1];	304	$type = ($TOLC{$_[1]} \|\|= tolc $_[1]);
228	$_[0]->push_read (line => $hdr_cb);	305	$_[0]->push_read (line => $hdr_cb);
229	});	306	});
230	}	307	}
231		308
232	sub default_recv {	309	sub default_recv {
…		…
235	if ($type eq "node_hello") {	312	if ($type eq "node_hello") {
236	$self->{node_hello} = $kv;	313	$self->{node_hello} = $kv;
237	} elsif (exists $self->{id}{$kv->{identifier}}) {	314	} elsif (exists $self->{id}{$kv->{identifier}}) {
238	$self->{id}{$kv->{identifier}}($self, $type, $kv, $rdata)	315	$self->{id}{$kv->{identifier}}($self, $type, $kv, $rdata)
239	and delete $self->{id}{$kv->{identifier}};	316	and delete $self->{id}{$kv->{identifier}};
240	} else {
241	&{ $self->{progress} };
242	}	317	}
243	}	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 { };
244		387
245	sub _txn {	388	sub _txn {
246	my ($name, $sub) = @_;	389	my ($name, $sub) = @_;
247		390
248	*{$name} = sub {	391	*{$name} = sub {
249	splice @_, 1, 0, (my $cv = AnyEvent->condvar);	392	my $cv = AE::cv;
250	&$sub;
251	$cv
252	};
253		393
254	*{"$name\_sync"} = sub {	394	splice @_, 1, 0, $cv, sub { $cv->croak ($_[0]{extra_description}) };
255	splice @_, 1, 0, (my $cv = AnyEvent->condvar);
256	&$sub;	395	&$sub;
257	$cv->recv	396	$cv->recv
258	};	397	};
259	}
260		398
261	=item $cv = $fcp->list_peers ([$with_metdata[, $with_volatile]])	399	*{"$name\_"} = sub {
		400	my ($ok, $err) = pop;
262		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
263	=item $peers = $fcp->list_peers_sync ([$with_metdata[, $with_volatile]])	422	=item $peers = $fcp->list_peers ([$with_metdata[, $with_volatile]])
264		423
265	=cut	424	=cut
266		425
267	_txn list_peers => sub {	426	_txn list_peers => sub {
268	my ($self, $cv, $with_metadata, $with_volatile) = @_;	427	my ($self, $ok, undef, $with_metadata, $with_volatile) = @_;
269		428
270	my @res;	429	my @res;
271		430
272	$self->send_msg (list_peers =>	431	$self->send_msg (list_peers =>
273	with_metadata => $with_metadata ? "true" : "false",	432	with_metadata => $with_metadata ? "true" : "false",
274	with_volatile => $with_volatile ? "true" : "false",	433	with_volatile => $with_volatile ? "true" : "false",
275	id_cb => sub {	434	id_cb => sub {
276	my ($self, $type, $kv, $rdata) = @_;	435	my ($self, $type, $kv, $rdata) = @_;
277		436
278	if ($type eq "end_list_peers") {	437	if ($type eq "end_list_peers") {
279	$cv->(\@res);	438	$ok->(\@res);
280	1	439	1
281	} else {	440	} else {
282	push @res, $kv;	441	push @res, $kv;
283	0	442	0
284	}	443	}
285	},	444	},
286	);	445	);
287	};	446	};
288		447
289	=item $cv = $fcp->list_peer_notes ($node_identifier)
290
291	=item $notes = $fcp->list_peer_notes_sync ($node_identifier)	448	=item $notes = $fcp->list_peer_notes ($node_identifier)
292		449
293	=cut	450	=cut
294		451
295	_txn list_peer_notes => sub {	452	_txn list_peer_notes => sub {
296	my ($self, $cv, $node_identifier) = @_;	453	my ($self, $ok, undef, $node_identifier) = @_;
297		454
298	$self->send_msg (list_peer_notes =>	455	$self->send_msg (list_peer_notes =>
299	node_identifier => $node_identifier,	456	node_identifier => $node_identifier,
300	id_cb => sub {	457	id_cb => sub {
301	my ($self, $type, $kv, $rdata) = @_;	458	my ($self, $type, $kv, $rdata) = @_;
302		459
303	$cv->($kv);	460	$ok->($kv);
304	1	461	1
305	},	462	},
306	);	463	);
307	};	464	};
308		465
309	=item $cv = $fcp->watch_global ($enabled[, $verbosity_mask])
310
311	=item $fcp->watch_global_sync ($enabled[, $verbosity_mask])	466	=item $fcp->watch_global ($enabled[, $verbosity_mask])
312		467
313	=cut	468	=cut
314		469
315	_txn watch_global => sub {	470	_txn watch_global => sub {
316	my ($self, $cv, $enabled, $verbosity_mask) = @_;	471	my ($self, $ok, $err, $enabled, $verbosity_mask) = @_;
317		472
318	$self->send_msg (watch_global =>	473	$self->send_msg (watch_global =>
319	enabled => $enabled ? "true" : "false",	474	enabled => $enabled ? "true" : "false",
320	defined $verbosity_mask ? (verbosity_mask => $verbosity_mask+0) : (),	475	defined $verbosity_mask ? (verbosity_mask => $verbosity_mask+0) : (),
321	);	476	);
322		477
323	$cv->();	478	$ok->();
324	};	479	};
325		480
326	=item $cv = $fcp->list_persistent_requests
327
328	=item $reqs = $fcp->list_persistent_requests_sync	481	=item $reqs = $fcp->list_persistent_requests
329		482
330	=cut	483	=cut
331		484
332	_txn list_persistent_requests => sub {	485	_txn list_persistent_requests => sub {
333	my ($self, $cv) = @_;	486	my ($self, $ok, $err) = @_;
334		487
		488	$self->serialise (list_persistent_requests => sub {
		489	my ($self, $guard) = @_;
		490
335	my %res;	491	my @res;
336		492
337	$self->send_msg ("list_persistent_requests");	493	$self->send_msg ("list_persistent_requests");
338		494
339	push @{ $self->{queue} }, sub {	495	$self->on (sub {
340	my ($self, $type, $kv, $rdata) = @_;	496	my ($self, $type, $kv, $rdata) = @_;
341		497
		498	$guard if 0;
		499
342	if ($type eq "end_list_persistent_requests") {	500	if ($type eq "end_list_persistent_requests") {
343	$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
344	1	511	1
345	} else {	512	});
346	my $id = $kv->{identifier};	513	});
		514	};
347		515
348	if ($type =~ /^persistent_(get\|put\|put_dir)$/) {	516	=item $sync = $fcp->modify_persistent_request ($global, $identifier[, $client_token[, $priority_class]])
349	$res{$id} = {	517
350	type => $1,	518	Update either the C<client_token> or C<priority_class> of a request
351	%{ $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") {
352	%$kv,	544	$ok->($kv);
		545	return;
		546	} elsif ($type eq "protocol_error") {
		547	$err->($kv);
		548	return;
353	};	549	}
354	} elsif ($type eq "simple_progress") {
355	delete $kv->{pkt_type}; # save memory
356	push @{ $res{delete $kv->{identifier}}{simple_progress} }, $kv;
357	} else {
358	$res{delete $kv->{identifier}}{delete $kv->{pkt_type}} = $kv;
359	}	550	}
		551
360	0	552	1
361	}	553	});
362	};	554	});
363	};	555	};
364		556
365	=item $cv = $fcp->remove_request ($global, $identifier)
366
367	=item $status = $fcp->remove_request_sync ($global, $identifier)
368
369	=cut
370
371	_txn remove_request => sub {
372	my ($self, $cv, $global, $identifier) = @_;
373
374	$self->send_msg (remove_request =>
375	global => $global ? "true" : "false",
376	identifier => $identifier,
377	id_cb => sub {
378	my ($self, $type, $kv, $rdata) = @_;
379
380	$cv->($kv);
381	1
382	},
383	);
384	};
385
386	=item $cv = $fcp->modify_persistent_request ($global, $identifier[, $client_token[, $priority_class]])
387
388	=item $sync = $fcp->modify_persistent_request_sync ($global, $identifier[, $client_token[, $priority_class]])
389
390	=cut
391
392	_txn modify_persistent_request => sub {
393	my ($self, $cv, $global, $identifier, $client_token, $priority_class) = @_;
394
395	$self->send_msg (modify_persistent_request =>
396	global => $global ? "true" : "false",
397	defined $client_token ? (client_token => $client_token ) : (),
398	defined $priority_class ? (priority_class => $priority_class) : (),
399	identifier => $identifier,
400	id_cb => sub {
401	my ($self, $type, $kv, $rdata) = @_;
402
403	$cv->($kv);
404	1
405	},
406	);
407	};
408
409	=item $cv = $fcp->get_plugin_info ($name, $detailed)
410
411	=item $info = $fcp->get_plugin_info_sync ($name, $detailed)	557	=item $info = $fcp->get_plugin_info ($name, $detailed)
412		558
413	=cut	559	=cut
414		560
415	_txn get_plugin_info => sub {	561	_txn get_plugin_info => sub {
416	my ($self, $cv, $name, $detailed) = @_;	562	my ($self, $ok, $err, $name, $detailed) = @_;
		563
		564	my $id = $self->identifier;
417		565
418	$self->send_msg (get_plugin_info =>	566	$self->send_msg (get_plugin_info =>
		567	identifier => $id,
419	plugin_name => $name,	568	plugin_name => $name,
420	detailed => $detailed ? "true" : "false",	569	detailed => $detailed ? "true" : "false",
421	id_cb => sub {
422	my ($self, $type, $kv, $rdata) = @_;
423
424	$cv->($kv);
425	1
426	},
427	);	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	});
428	};	585	};
429		586
430	=item $cv = $fcp->client_get ($uri, $identifier, %kv)
431
432	=item $status = $fcp->client_get_sync ($uri, $identifier, %kv)	587	=item $status = $fcp->client_get ($uri, $identifier, %kv)
433		588
434	%kv can contain (L<http://wiki.freenetproject.org/FCP2p0ClientGet>).	589	%kv can contain (L<http://wiki.freenetproject.org/FCP2p0ClientGet>).
435		590
436	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,
437	priority_class, persistence, client_token, global, return_type,	592	priority_class, persistence, client_token, global, return_type,
438	binary_blob, allowed_mime_types, filename, temp_filename	593	binary_blob, allowed_mime_types, filename, temp_filename
439		594
440	=cut	595	=cut
441		596
442	_txn client_get => sub {	597	_txn client_get => sub {
443	my ($self, $cv, $uri, $identifier, %kv) = @_;	598	my ($self, $ok, $err, $uri, $identifier, %kv) = @_;
444		599
		600	$self->serialise ($identifier => sub {
		601	my ($self, $guard) = @_;
		602
445	$self->send_msg (client_get =>	603	$self->send_msg (client_get =>
446	%kv,	604	%kv,
447	uri => $uri,	605	uri => $uri,
448	identifier => $identifier,	606	identifier => $identifier,
449	id_cb => sub {	607	);
		608
		609	$self->on (sub {
450	my ($self, $type, $kv, $rdata) = @_;	610	my ($self, $type, $kv, @extra) = @_;
451		611
		612	$guard if 0;
		613
		614	if ($kv->{identifier} eq $identifier) {
		615	if ($type eq "persistent_get") {
452	$cv->($kv);	616	$ok->($kv);
		617	return;
		618	} elsif ($type eq "protocol_error") {
		619	$err->($kv);
		620	return;
		621	}
		622	}
		623
453	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",
454	},	842	},
455	);	843	compatibility_mode => {
456	};	844	identifier => "Frost-gpl.txt",
457		845	definitive => "true",
458	=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	}
459		893
460	=head1 EXAMPLE PROGRAM	894	=head1 EXAMPLE PROGRAM
461		895
462	use AnyEvent::FCP;	896	use AnyEvent::FCP;
463		897
464	my $fcp = new AnyEvent::FCP;	898	my $fcp = new AnyEvent::FCP;
465		899
466	# let us look at the global request list	900	# let us look at the global request list
467	$fcp->watch_global (1, 0);	901	$fcp->watch_global_ (1);
468		902
469	# list them, synchronously	903	# list them, synchronously
470	my $req = $fcp->list_persistent_requests_sync;	904	my $req = $fcp->list_persistent_requests;
471		905
472	# go through all requests	906	# go through all requests
		907	TODO
473	for my $req (values %$req) {	908	for my $req (values %$req) {
474	# skip jobs not directly-to-disk	909	# skip jobs not directly-to-disk
475	next unless $req->{return_type} eq "disk";	910	next unless $req->{return_type} eq "disk";
476	# skip jobs not issued by FProxy	911	# skip jobs not issued by FProxy
477	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.8 by root, Fri Jun 18 16:59:13 2010 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.8 by root, Fri Jun 18 16:59:13 2010 UTC vs.
Revision 1.16 by root, Sat Sep 5 13:26:47 2015 UTC