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

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