Return to Das.pm CVS log

Up to [biodas] / Bio-Das2

fixed dsn call according to James Smith

package Bio::Das;
# $Id: Das.pm,v 1.8 2002/02/19 23:10:15 lstein Exp $

# prototype parallel-fetching Das

use strict;
use Bio::Das::HTTP::Fetch;
use Bio::Das::Request::Dsn;    # bring in dsn  parser
use Bio::Das::Request::Types;  # bring in type parser
use Bio::Das::Request::Dnas;
use Bio::Das::Request::Features;
use Bio::Das::Util 'rearrange';
use Carp;
use IO::Socket;
use IO::Select;
use vars '$VERSION';
$VERSION = 0.60;
sub new {
  my $package = shift;
  my $timeout = shift;
  return bless {
		'sockets'  => {},   # map socket to Bio::Das::HTTP::Fetch objects
		'timeout'  => $timeout,
	       },$package;
}
sub proxy {
  my $self = shift;
  my $d    = $self->{proxy};
  $self->{proxy} = shift if @_;
  $d;
}
sub timeout {
  my $self = shift;
  my $d = $self->{timeout};
  $self->{timeout} = shift if @_;
  $d;
}
sub debug {
  my $self = shift;
  my $d = $self->{debug};
  $self->{debug} = shift if @_;
  $d;
}
# call with list of base names
# will return a list of DSN objects
sub dsn {
  my $self = shift;
#  my @requests = map { Bio::Das::Request::Dsn->new($_) } @_;
  my @requests = Bio::Das::Request::Dsn->new(@_);
  $self->run_requests(\@requests);
}
# call with list of DSN objects, and optionally list of segments and categories
sub types {
  my $self = shift;
  my ($dsn,$segments,$categories) = rearrange([['dsn','dsns'],
					       ['segment','segments'],
					       ['category','categories'],
					      ],@_);
  croak "must provide -dsn argument" unless $dsn;
  my @dsn = ref $dsn ? @$dsn : $dsn;
  my @request;
  for my $dsn (@dsn) {
    push @request,Bio::Das::Request::Types->new(-dsn        => $dsn,
						-segment    => $segments,
						-categories => $categories);
  }
  $self->run_requests(\@request);
}
# call with list of DSN objects, and a list of one or more segments
sub dna {
  my $self = shift;
  my ($dsn,$segments) = rearrange([['dsn','dsns'],
				   ['segment','segments']
				  ],@_);
  croak "must provide -dsn argument" unless $dsn;
  my @dsn = ref $dsn ? @$dsn : $dsn;
  my @request;
  for my $dsn (@dsn) {
    push @request,Bio::Das::Request::Dnas->new(-dsn        => $dsn,
					       -segment    => $segments);
  }
  $self->run_requests(\@request);
}
# call with list of DSNs, and optionally list of segments and categories
sub features {
  my $self = shift;
  my ($dsn,$segments,$types,$categories,$callback) = rearrange([['dsn','dsns'],
								['segment','segments'],
								['type','types'],
								['category','categories'],
								'callback',
							       ],@_);
  croak "must provide -dsn argument" unless $dsn;
  my @dsn = ref $dsn ? @$dsn : $dsn;
  my @request;
  for my $dsn (@dsn) {
    push @request,Bio::Das::Request::Features->new(-dsn        => $dsn,
						   -segments   => $segments,
						   -types      => $types,
						   -categories => $categories,
						   -callback   => $callback);
  }
  $self->run_requests(\@request);
}
sub add_pending {
  my $self    = shift;
  my $fetcher = shift;
  $self->{sockets}{$fetcher->socket} = $fetcher;
}
sub run_requests {
  my $self     = shift;
  my $requests = shift;
  for my $request (@$requests) {
    my $fetcher = Bio::Das::HTTP::Fetch->new(-request => $request,
					     -headers => {'Accept-encoding' => 'gzip'},
					     -proxy   => $self->proxy || '',
					    ) or next;
    $fetcher->debug(1) if $self->debug;
    $self->add_pending($fetcher);
  }
  my $timeout = $self->timeout;
  # create two IO::Select objects to handle writing & reading  
  my $readers = IO::Select->new;
  my $writers = IO::Select->new;
  for my $fetcher (values %{$self->{sockets}}) {
    my $socket = $fetcher->socket;
    $writers->add($socket);
  }
  my $timed_out;
  while ($readers->count or $writers->count) {
    my ($readable,$writable) = IO::Select->select($readers,$writers,undef,$timeout);
    ++$timed_out && last unless $readable || $writable;
    foreach (@$writable) {                      # handle is ready for writing
      my $fetcher = $self->{sockets}{$_};       # recover the HTTP fetcher
      my $result = $fetcher->send_request;      # try to send the request
      $readers->add($_) if $result;             # send successful, so monitor for reading
      $fetcher->request->error($fetcher->error) 
	unless $result;                         # copy the error message
      $writers->remove($_);                     # and remove from list monitored for writing
    }
    foreach (@$readable) {        # handle is ready for reading
      my $fetcher = $self->{sockets}{$_};       # recover the HTTP object
      my $result = $fetcher->read;              # read some data
      unless ($result) {                    # remove if some error occurred
	$fetcher->request->error($fetcher->error) unless defined $result;
	$readers->remove($_);
	delete $self->{sockets}{$_};
      }
    }
  }
  # handle timeouts
  if ($timed_out) {
    while (my ($sock,$f) = each %{$self->{sockets}}) { # list of still-pending requests
      $f->request->error('timeout');
      $readers->remove($sock);
      $writers->remove($sock);
      close $sock;
    }
  }
  delete $self->{sockets};
  return wantarray ? @$requests : $requests->[0];
}
1;