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