| … | |
… | |
| 66 | |
66 | |
| 67 | use Scalar::Util (); |
67 | use Scalar::Util (); |
| 68 | |
68 | |
| 69 | use AnyEvent; |
69 | use AnyEvent; |
| 70 | use AnyEvent::Handle; |
70 | use AnyEvent::Handle; |
|
|
71 | use AnyEvent::Util (); |
| 71 | |
72 | |
| 72 | sub touc($) { |
73 | sub touc($) { |
| 73 | local $_ = shift; |
74 | local $_ = shift; |
| 74 | 1 while s/((?:^|_)(?:svk|chk|uri|fcp|ds|mime)(?:_|$))/\U$1/; |
75 | 1 while s/((?:^|_)(?:svk|chk|uri|fcp|ds|mime|dda)(?:_|$))/\U$1/; |
| 75 | s/(?:^|_)(.)/\U$1/g; |
76 | s/(?:^|_)(.)/\U$1/g; |
| 76 | $_ |
77 | $_ |
| 77 | } |
78 | } |
| 78 | |
79 | |
| 79 | sub tolc($) { |
80 | sub tolc($) { |
| 80 | local $_ = shift; |
81 | local $_ = shift; |
| 81 | 1 while s/(SVK|CHK|URI|FCP|DS|MIME)([^_])/$1\_$2/i; |
82 | 1 while s/(SVK|CHK|URI|FCP|DS|MIME|DDA)([^_])/$1\_$2/i; |
| 82 | 1 while s/([^_])(SVK|CHK|URI|FCP|DS|MIME)/$1\_$2/i; |
83 | 1 while s/([^_])(SVK|CHK|URI|FCP|DS|MIME|DDA)/$1\_$2/i; |
| 83 | s/(?<=[a-z])(?=[A-Z])/_/g; |
84 | s/(?<=[a-z])(?=[A-Z])/_/g; |
| 84 | lc |
85 | lc |
| 85 | } |
86 | } |
| 86 | |
87 | |
| 87 | =item $fcp = new AnyEvent::FCP [host => $host][, port => $port][, progress => \&cb][, name => $name] |
88 | =item $fcp = new AnyEvent::FCP [host => $host][, port => $port][, progress => \&cb][, name => $name] |
| … | |
… | |
| 149 | my ($self, $type, %kv) = @_; |
150 | my ($self, $type, %kv) = @_; |
| 150 | |
151 | |
| 151 | my $data = delete $kv{data}; |
152 | my $data = delete $kv{data}; |
| 152 | |
153 | |
| 153 | if (exists $kv{id_cb}) { |
154 | if (exists $kv{id_cb}) { |
| 154 | my $id = $kv{identifier} || ++$self->{id}; |
155 | my $id = $kv{identifier} ||= ++$self->{id}; |
| 155 | $self->{id}{$id} = delete $kv{id_cb}; |
156 | $self->{id}{$id} = delete $kv{id_cb}; |
| 156 | $kv{identifier} = $id; |
|
|
| 157 | } |
157 | } |
| 158 | |
158 | |
| 159 | my $msg = (touc $type) . "\012" |
159 | my $msg = (touc $type) . "\012" |
| 160 | . join "", map +(touc $_) . "=$kv{$_}\012", keys %kv; |
160 | . join "", map +(touc $_) . "=$kv{$_}\012", keys %kv; |
| 161 | |
161 | |
| … | |
… | |
| 173 | } |
173 | } |
| 174 | |
174 | |
| 175 | $self->{hdl}->push_write ($msg); |
175 | $self->{hdl}->push_write ($msg); |
| 176 | } |
176 | } |
| 177 | |
177 | |
|
|
178 | sub on { |
|
|
179 | my ($self, $cb) = @_; |
|
|
180 | |
|
|
181 | # cb return undef - message eaten, remove cb |
|
|
182 | # cb return 0 - message eaten |
|
|
183 | # cb return 1 - pass to next |
|
|
184 | |
|
|
185 | push @{ $self->{on} }, $cb; |
|
|
186 | } |
|
|
187 | |
|
|
188 | sub _push_queue { |
|
|
189 | my ($self, $queue) = @_; |
|
|
190 | |
|
|
191 | warn "oush @$queue\n";#d# |
|
|
192 | shift @$queue; |
|
|
193 | $queue->[0]($self, AnyEvent::Util::guard { $self->_push_queue ($queue) }) |
|
|
194 | if @$queue; |
|
|
195 | } |
|
|
196 | |
|
|
197 | # lock so only one $type (arbitrary string) is in flight, |
|
|
198 | # to work around horribly misdesigned protocol. |
|
|
199 | sub serialise { |
|
|
200 | my ($self, $type, $cb) = @_; |
|
|
201 | |
|
|
202 | my $queue = $self->{serialise}{$type} ||= []; |
|
|
203 | push @$queue, $cb; |
|
|
204 | $cb->($self, AnyEvent::Util::guard { $self->_push_queue ($queue) }) |
|
|
205 | unless $#$queue; |
|
|
206 | } |
|
|
207 | |
| 178 | sub on_read { |
208 | sub on_read { |
| 179 | my ($self) = @_; |
209 | my ($self) = @_; |
| 180 | |
210 | |
| 181 | my $type; |
211 | my $type; |
| 182 | my %kv; |
212 | my %kv; |
| 183 | my $rdata; |
213 | my $rdata; |
| 184 | |
214 | |
| 185 | my $done_cb = sub { |
215 | my $done_cb = sub { |
| 186 | $kv{pkt_type} = $type; |
216 | $kv{pkt_type} = $type; |
|
|
217 | |
|
|
218 | my $on = $self->{on}; |
|
|
219 | for (0 .. $#$on) { |
|
|
220 | unless (my $res = $on->[$_]($type, \%kv, $rdata)) { |
|
|
221 | splice @$on, $_, 1 unless defined $res; |
|
|
222 | return; |
|
|
223 | } |
|
|
224 | } |
| 187 | |
225 | |
| 188 | if (my $cb = $self->{queue}[0]) { |
226 | if (my $cb = $self->{queue}[0]) { |
| 189 | $cb->($self, $type, \%kv, $rdata) |
227 | $cb->($self, $type, \%kv, $rdata) |
| 190 | and shift @{ $self->{queue} }; |
228 | and shift @{ $self->{queue} }; |
| 191 | } else { |
229 | } else { |
| … | |
… | |
| 453 | 1 |
491 | 1 |
| 454 | }, |
492 | }, |
| 455 | ); |
493 | ); |
| 456 | }; |
494 | }; |
| 457 | |
495 | |
|
|
496 | =item $cv = $fcp->test_dda_sync ($local_directory, $remote_directory, $want_read, $want_write) |
|
|
497 | |
|
|
498 | =item ($can_read, $can_write) = $fcp->test_dda_sync ($local_directory, $remote_directory, $want_read, $want_write)) |
|
|
499 | |
|
|
500 | The DDA test in FCP is probably the single most broken protocol - only |
|
|
501 | one directory test can be outstanding at any time, and some guessing and |
|
|
502 | heuristics are involved in mangling the paths. |
|
|
503 | |
|
|
504 | This function combines C<TestDDARequest> and C<TestDDAResponse> in one |
|
|
505 | request, handling file reading and writing as well. |
|
|
506 | |
|
|
507 | =cut |
|
|
508 | |
|
|
509 | _txn test_dda => sub { |
|
|
510 | my ($self, $cv, $local, $remote, $want_read, $want_write) = @_; |
|
|
511 | |
|
|
512 | $self->serialise (test_dda => sub { |
|
|
513 | my ($self, $guard) = @_; |
|
|
514 | |
|
|
515 | $self->send_msg (test_dda_request => |
|
|
516 | directory => $remote, |
|
|
517 | want_read_directory => $want_read ? "true" : "false", |
|
|
518 | want_write_directory => $want_write ? "true" : "false", |
|
|
519 | ); |
|
|
520 | $self->on (sub { |
|
|
521 | my ($type, $kv) = @_; |
|
|
522 | |
|
|
523 | if ($type eq "test_dda_reply") { |
|
|
524 | # the filenames are all relative to the server-side directory, |
|
|
525 | # which might or might not match $remote anymore, so we |
|
|
526 | # need to rewrite the paths to be relative to $local |
|
|
527 | for my $k (qw(read_filename write_filename)) { |
|
|
528 | my $f = $kv->{$k}; |
|
|
529 | for my $dir ($kv->{directory}, $remote) { |
|
|
530 | if ($dir eq substr $f, 0, length $dir) { |
|
|
531 | substr $f, 0, 1 + length $dir, ""; |
|
|
532 | $kv->{$k} = $f; |
|
|
533 | last; |
|
|
534 | } |
|
|
535 | } |
|
|
536 | } |
|
|
537 | |
|
|
538 | my %response = (directory => $remote); |
|
|
539 | |
|
|
540 | if (length $kv->{read_filename}) { |
|
|
541 | warn "$local/$kv->{read_filename}";#d# |
|
|
542 | if (open my $fh, "<:raw", "$local/$kv->{read_filename}") { |
|
|
543 | sysread $fh, my $buf, -s $fh; |
|
|
544 | $response{read_content} = $buf; |
|
|
545 | } |
|
|
546 | } |
|
|
547 | |
|
|
548 | if (length $kv->{write_filename}) { |
|
|
549 | if (open my $fh, ">:raw", "$local/$kv->{write_filename}") { |
|
|
550 | syswrite $fh, $kv->{content_to_write}; |
|
|
551 | } |
|
|
552 | } |
|
|
553 | |
|
|
554 | $self->send_msg (test_dda_response => %response); |
|
|
555 | |
|
|
556 | $self->on (sub { |
|
|
557 | my ($type, $kv) = @_; |
|
|
558 | |
|
|
559 | $guard if 0; # reference |
|
|
560 | |
|
|
561 | if ($type eq "test_dda_complete") { |
|
|
562 | $cv->( |
|
|
563 | $kv->{read_directory_allowed} eq "true", |
|
|
564 | $kv->{write_directory_allowed} eq "true", |
|
|
565 | ); |
|
|
566 | } elsif ($type eq "protocol_error" && $kv->{identifier} eq $remote) { |
|
|
567 | $cv->croak ($kv->{extra_description}); |
|
|
568 | return; |
|
|
569 | } |
|
|
570 | |
|
|
571 | 1 |
|
|
572 | }); |
|
|
573 | |
|
|
574 | return; |
|
|
575 | } elsif ($type eq "protocol_error" && $kv->{identifier} eq $remote) { |
|
|
576 | $cv->croak ($kv->{extra_description}); |
|
|
577 | return; |
|
|
578 | } |
|
|
579 | |
|
|
580 | 1 |
|
|
581 | }); |
|
|
582 | }); |
|
|
583 | }; |
|
|
584 | |
| 458 | =back |
585 | =back |
| 459 | |
586 | |
| 460 | =head1 EXAMPLE PROGRAM |
587 | =head1 EXAMPLE PROGRAM |
| 461 | |
588 | |
| 462 | use AnyEvent::FCP; |
589 | use AnyEvent::FCP; |
