package POE::Component::BitTorrent;
use warnings;
use strict;

use Carp;
use POE qw(Component::Client::HTTP Wheel::ListenAccept);
use IO::Socket::INET;
use Digest::SHA1 qw(sha1);
use Errno;
use HTTP::Request::Common qw(GET);

# For debug _exit
use POSIX;

use Data::Dumper;

our $VERSION = 0.01;

use constant CLEAN => 1;

use fields qw(alias http_alias id min_port max_port port listener me
              downloaders connections crashed_clients to_check to_write
              max_downloads nr_downloads heartbeat_alarm);

my $BUF_SIZE = 2**14;
my $TORRENT_ANNOUNCE = chr(19) . "BitTorrent protocol";
my $TORRENT_EXTENSION = "\x00" x 8;
# Maximum number of running downloads
my $MAX_DOWNLOADS = CLEAN ? 4 : 2;
# Number of requests in transit
my $MAX_REQUESTS = 5;
# Kill crashed trackers. Don't turn this on in production, too evil
my $KILL_CRASHERS = 1;
# Never send updates sooner than this, whatever the tracker requests
# Defends against malicious trackers
my $MIN_TRACKER_INTERVAL = 30;
# Fragment size we tend to request
my $PIECE_REQUEST_SIZE	= CLEAN ? 2**15 : 16;
# Close the connection if someone tries to send a message bigger than this
my $MAX_MESSAGE		= 2**17+1;
my $KEEPALIVE_PERIOD	= 120;
$KEEPALIVE_PERIOD *= 2/3;
# After this much of nothing received at all, close the connection
my $IDLE_TIMEOUT	= 300;
# After this much of no real message, close the connection
my $LOGICAL_IDLE_TIMEOUT = 3000;
my $RANDOM_PIECES	= 5;
# Protocol values (later maybe replace with a real enum)
my $CHOKE	= 0;
my $UNCHOKE	= 1;
my $INTERESTED	= 2;
my $NOT_INTERESTED = 3;
my $HAVE	= 4;
my $BIT_FIELD	= 5;
my $REQUEST	= 6;
my $PIECE	= 7;
my $CANCEL	= 8;
my $KEEPALIVE	= 9;	# Some invalid message number

# destroys the argument string !
sub destructive_bdecode {
    # $_[0] =~ /^(\w*)/; print STDERR "Considering '$1'\n";
    return substr($_[0], 0, $1, "") if $_[0] =~ s/^(\d+)://;
    return 0+$1 if $_[0] =~ s/^i(-?\d+)e//;
    return if $_[0] =~ s/^e//;
    if ($_[0] =~ s/^(l|d)//) {
        my $type = $1;
        my @list;
        do {
            push(@list, scalar(&destructive_bdecode));
        } while defined($list[-1]);
        pop(@list);
        return \@list if $type eq "l";
        die "Odd number of elements in bencoded dictionary" if @list%2;
        my %hash = @list;
        die "Duplicate keys in bencoded dictionary" unless @list==2*keys %hash;
        return \%hash;
    }
    die "Empty bencoded string is invalid" if $_[0] eq "";
    my ($type) = $_[0] =~ /^(\w+)/;
    die "Uknown bencoding type '$type'";
}

sub bdecode {
    my $string = shift;
    return destructive_bdecode($string);
}

# Appends to the argument string
sub constructive_bencode {
    if (ref($_[1])) {
        if (ref($_[1]) eq "ARRAY") {
            $_[0] .= "l";
            constructive_bencode($_[0], $_) for @{$_[1]};
            $_[0] .= "e";
            return;
        }
        if (ref($_[1]) eq "HASH") {
            $_[0] .= "d";
            for (sort keys %{$_[1]}) {
                constructive_bencode($_[0], $_);
                constructive_bencode($_[0], $_[1]{$_});
            }
            $_[0] .= "e";
            return;
        }
        die "Cannot bencode a ", ref($_[1]), " reference";
    }
    # String or integer
    # Extreme hack. Checks if it looks like an integer, and then if
    # the representation is numeric at perl level
    if ($_[1] =~ /^-?[1-9]\d*\z|^0\z/ && ($_[1] & (my $str= "+0")) eq "0") {
        $_[0] .= "i$_[1]e";
        return;
    }
    $_[0] .= length($_[1]) . ":";
    $_[0] .= $_[1];
    return;
}

sub bencode {
    constructive_bencode(my $result = "", shift);
    return $result;
}

sub spawn {
    my $class = shift;
    croak "$class requires an even number of parameters" if @_ % 2;
    my %params = @_;

    my $alias = delete $params{Alias};
    $alias = "bt" unless defined $alias and length $alias;
    my $http_alias = $alias . "_http";
    "POE::Component::Client::HTTP"->spawn
        (Alias	=> $http_alias,
         MaxSize => 65536, # Should be enough for a few thousand pieces
         %params
         );
    my __PACKAGE__ $client = bless [], $class;
    $client->{http_alias} = $http_alias;
    $client->{alias}      = $alias;
    $client->{id} = sha1("TorrentClient",$0,$<,$^T,time,$client,rand,$$);
    print(STDERR "I am ", 
          POE::Component::BitTorrent::Peer::id_string($client->{id}), "\n");

    $client->{min_port} = 6881;
    $client->{max_port} = 6890;
    $client->{downloaders}  = {};
    $client->{crashed_clients} = {};
    $client->{to_check} = [];
    $client->{max_downloads} = $MAX_DOWNLOADS;
    $client->{nr_downloads}  = 0;
    "POE::Session"->create
        (heap	=> $client,
         inline_states => {
             _start		=> \&start,
             heartbeat		=> \&heartbeat,
             request		=> \&request,
             torrent_response	=> \&torrent_response,
             update_tracker	=> \&update_tracker,
             tracker_response	=> \&tracker_response,
             get_port		=> \&get_port,
             bad_peer		=> \&bad_peer,
             new_peer		=> \&new_peer,
             peer_send  	=> \&peer_send,
             check_empty	=> \&check_empty,
             get_handshake	=> \&get_handshake,
             get_message	=> \&get_message,
             start_check_hash	=> \&start_check_hash,
             check_hash		=> \&check_hash,
             write_fragment	=> \&write_fragment,
             _default	=> sub {
                 die("calling non existant state ", $_[ARG0]) unless
                     substr($_[ARG0], 0, 1) eq "_";
                 return;
             },
         });
}

sub start {
    my __PACKAGE__ $client = $_[HEAP];
    $poe_kernel->alias_set($client->{alias});
    $poe_kernel->yield("get_port");
}

sub get_port {
    (my __PACKAGE__ $client, my $p, my $try) = @_[HEAP, ARG0, ARG1];
    my $port = defined($p) ? $p : defined($try) ? $try : $client->{min_port};
    print STDERR "Trying port $port\n";
    if (my $socket = IO::Socket::INET->new
        (LocalPort	=> $port,
         Listen		=> SOMAXCONN)) {
        $client->{port} = $port;
        print STDERR "Got port $port, using it as listener\n";
        $client->{listener} = POE::Wheel::ListenAccept->new
            (Handle      => $socket,
             AcceptEvent => "new_peer",
             ErrorEvent  => "bad_peer") ||
             die "Could not create POE::Wheel::ListenAccept";
        $client->start_downloaders;
        return;
    }
    die "Could not bind to port $port: $!" if
        defined($p) || !$!{EADDRINUSE} || $port >= $client->{max_port};
    $poe_kernel->yield(get_port => $p, ++$port);
}

sub bad_peer {
    my ($operation, $errstr) = @_[ARG0, ARG2];
    print STDERR "bad peer: $operation: $errstr\n";
}

sub new_peer {
    (my __PACKAGE__ $client, my $socket, my $remote) = @_[HEAP, ARG0, ARG1];
    my ($port, $packed_ip) = sockaddr_in($remote);
    my $dotted_quad = inet_ntoa($packed_ip);
    print STDERR "Connection from $dotted_quad:$port\n";
    die "Already have a connection for socket $socket"
        if $client->{connections}{$socket};
    if (!%{$client->{connections}}) {
        die "Heartbeat already set" if $client->{heartbeat_alarm};
        $client->{heartbeat_alarm} = $poe_kernel->delay_set
            ("heartbeat", $KEEPALIVE_PERIOD-rand($KEEPALIVE_PERIOD)/4);
    }
    $client->{connections}{$socket} =
        POE::Component::BitTorrent::Connection->new
        ($client, $socket, $dotted_quad, $port);
}

# A peer socket is getting closed. Clean up all references
sub lost_connection {
    (my __PACKAGE__ $client, my $socket) = @_;
    die "No connection" unless delete $client->{connections}{$socket};
    if (!%{$client->{connections}}) {
        die "No heartbeat set" unless $client->{heartbeat_alarm};
        $poe_kernel->alarm_remove($client->{heartbeat_alarm});
        $client->{heartbeat_alarm} = undef;
    }
    $client->print;
}

sub heartbeat {
    my __PACKAGE__ $client = $_[HEAP];
    die "Heartbeat but no connections" unless %{$client->{connections}};
    die "No heartbeat set" unless $client->{heartbeat_alarm};
    $client->{heartbeat_alarm} = $poe_kernel->delay_set
        ("heartbeat", $KEEPALIVE_PERIOD-rand($KEEPALIVE_PERIOD)/4);
    my $now = time;
    $_->heartbeat($now) for values %{$client->{connections}};
}

# Get the global client id
sub id {
    my __PACKAGE__ $client = shift;
    return $client->{id};
}

# We have buffered data and the socket selected writable
sub peer_send {
    (my __PACKAGE__ $client, my $socket) = @_[HEAP, ARG0];
    $client->{connections}{$socket}->send;
}

# A connection need to be checked for empty
sub check_empty {
    $_[ARG0]->check_empty;
}

# We received some more bytes from the handshake
sub get_handshake {
    (my __PACKAGE__ $client, my $socket) = @_[HEAP, ARG0];
    $client->{connections}{$socket}->get_handshake;
}

# We get a .torrent URL. Retrieve it to get the circus going.
sub request {
    my ($sender, $response_event, $http_request, $user_info) =
        @_[SENDER, ARG0, ARG1, ARG2];
    my __PACKAGE__ $client = $_[HEAP];
    $poe_kernel->post
        ($client->{http_alias},
         "request",		# posts to ua's 'request' state
         "torrent_response",	# parse metafile on answer
         $http_request,
         $sender->postback($response_event, $http_request, $user_info));
}

# Got a .torrent from a webserver
sub torrent_response {
    my __PACKAGE__ $client = $_[HEAP];
    my ($request, $response) = @_[ARG0, ARG1];
    $response = $response->[0];
    if (!$response->is_success) {
        my $string = $request->[0]->uri->as_string() . " failed: " .
            $response->message;
        print STDERR "$string\n";
        $request->[1]->($string);
        return;
    }
    my $type = $response->content_type;
    if ($type ne "application/x-bittorrent") {
        my $string = "Unexpected content type '$type'";
        print STDERR "$string\n";
        $request->[1]->($string);
        return;
    }

    eval { $client->metainfo($response->content, $request->[1]) };
    if ($@) {
        $@ =~ s/\n\z//;
        print STDERR "$@\n";
        $request->[1]->($@);
        return;
    }
}

# Parse a torrent file and set up a download
sub metainfo {
    (my __PACKAGE__ $client, my $torrent, my $callback) = @_;
    my $metainfo = bdecode($torrent);
    die "metainfo is not a dictionary\n" if ref($metainfo) ne "HASH";
    exists $metainfo->{$_} || die "dictionary has no $_ entry\n" for
        qw(announce info);
    my $info = $metainfo->{info};
    my $file_id = sha1(bencode($info));
    my $downloader = $client->{downloaders}{$file_id} ||=
        POE::Component::BitTorrent::Downloader->new
        ($client, $file_id, $info);
    $downloader->add_callback($callback);
    $downloader->announce_tracker($metainfo,
                                  http_alias	=> $client->{http_alias},
                                  peer_id	=> $client->{id},
                                  port 		=> $client->{port});
}

# We finally got our port. Now start all pending downloaders
sub start_downloaders {
    my __PACKAGE__ $client = shift;
    $_->start_trackers(http_alias => $client->{http_alias},
                       peer_id	=> $client->{id},
                       port 	=> $client->{port}
                       ) for values %{$client->{downloaders}};
}

# Start one of the periodic announcements to the tracker
sub update_tracker {
    my ($downloader, $tracker) = @_[ARG0, ARG1];
    my __PACKAGE__ $client = $_[HEAP];
    $downloader->update_tracker($tracker,
                                http_alias => $client->{http_alias},
                                peer_id	=> $client->{id},
                                port 	=> $client->{port});
}

# Trackers tells us about (potential) peers
sub tracker_response {
    my ($request_packet, $response_packet) = @_[ARG0, ARG1];
    my __PACKAGE__ $client = $_[HEAP];

    my $request  = $request_packet->[0];	# HTTP::Request
    my %query = $request->uri->query_form;
    defined(my $file_id = $query{info_hash}) ||
        die("Tracker response from request without info_hash (",
            $request->uri->as_string, ")");
    defined(my $tracker    = $request_packet->[1]) ||
        die "Response not associated with tracker ???";
    defined(my $downloader = $client->{downloaders}{$file_id}) ||
        die "Response for file_id i never asked for ???";

    my $response = $response_packet->[0];	# HTTP::Response
    if (!$response->is_success) {
        # Should (possibly) drop tracker here
        print(STDERR $request->uri->as_string, " failed: ",
              $response->message,"\n");
        $downloader->all_callbacks($downloader);
        return;
    }

    eval {
        my $response = bdecode($response->content);
        # print STDERR "response=", Dumper($response);
        ref($response) eq "HASH" || die "not a dictionary";
        die $response->{"failure reason"} if exists
            $response->{"failure reason"};
        exists $response->{$_} || die "has no $_ entry\n" for
            qw(peers interval);
        my $peers = $response->{peers};
        ref($peers) eq "ARRAY" || die "peers is not a list";
        my %peers;
        for my $peer (@$peers) {
            ref($peer) eq "HASH" || die "Some peer is not a dictionary";
            exists $peer->{$_} || die "Some peer has no $_\n" for
                "peer id", "ip", "port";
            print(STDERR "tracker: $peer->{ip}:$peer->{port}:",
                  POE::Component::BitTorrent::Peer::id_string($peer->{"peer id"}), "\n");
            if ($peer->{"peer id"} eq $client->{id}) {
                $client->{me}{"$peer->{ip} $peer->{port}"} = 1;
                next;
            }
            $downloader->announce_peer($peer->{"peer id"},
                                       $peer->{ip}, $peer->{port});
            push(@{$peers{"$peer->{ip} $peer->{port}"}}, $peer->{"peer id"});
        }
        $tracker->interval($downloader, $response->{interval});
        $downloader->print;
        if (!CLEAN && $query{event} && $query{event} eq "completed") {
            print STDERR "Test done\n";
            exit 0;
        }

        if ($KILL_CRASHERS) {
            # These heuristics are buggy
            # - crashed clients should know on which tracker/file_id they sit
            # Only turn on for development and when not talking to the real net
            for my $me (keys %{$client->{me}}) {
                for my $peer_id (@{delete($peers{$me}) || []}) {
                    # Something with not my id runs on my ip:port
                    # Most likely an old crashed client
                    $client->{crashed_clients}{"$file_id:$tracker:$peer_id:$me"}=
                        [$file_id, $tracker, $request->uri, $peer_id,
                         split " ", $me];
                }
            }
            # Should remove spontanously died clients here
            if ($query{event} && $query{event} eq "stopped") {
                my $id = "$file_id:$tracker:$query{peer_id}:$query{ip} $query{port}";
                delete($client->{crashed_clients}{$id}) ||
                    die "Could not get rid of crashed client $id";
                $downloader->delete_peer_address(@query{qw(peer_id ip port)});
            }
            $client->kill_crasher if %{$client->{crashed_clients}};
        }
    };
    if ($@) {
        print(STDERR $request->uri->as_string,
              " returned a bad tracker response: $@");
        $downloader->all_callbacks();
        return;
    }
}

# A downloader finished its file
sub complete_trackers {
    (my __PACKAGE__ $client, my $downloader) = @_;
    $downloader->complete_trackers(http_alias	=> $client->{http_alias},
                                   peer_id	=> $client->{id},
                                   port 	=> $client->{port});
}

# Send a fake "stopped" update to a tracker
sub kill_crasher {
    my __PACKAGE__ $client = shift;
    # reset iterator
    die "No crashers" unless keys %{$client->{crashed_clients}};
    my (undef, $crasher) = each %{$client->{crashed_clients}};
    my ($file_id, $tracker, $uri, $peer_id, $ip, $port) = @$crasher;
    $uri->query_form(info_hash	=> $file_id,
                     peer_id	=> $peer_id,
                     ip		=> $ip,
                     port	=> $port,
                     uploaded	=> 0,
                     downloaded	=> 0,
                     left	=> 0,
                     event	=> "stopped");
    print(STDERR "Sending stopped for crashed client ",
          POE::Component::BitTorrent::Peer::id_string($peer_id),
          " at $ip:$port\n");
    $poe_kernel->post
        ($client->{http_alias},
         "request",		# posts to ua's 'request' state
         "tracker_response",	# state getting response
         GET($uri),
         $tracker);
}

# Given a file id, returns the corresponding downloader.
# Returns undef if the file id is unknown
sub file_id2downloader {
    my __PACKAGE__ $client = shift;
    return $client->{downloaders}{shift()};
}

# We received a message on a peer connection
sub get_message {
    (my __PACKAGE__ $client, my $socket) = @_[HEAP, ARG0];
    eval {
        $client->{connections}{$socket}->get_message;
    };
    if ($@) {
        print STDERR "->$@";
        $client->{connections}{$socket}->close;
        die "-----";
    }
}

# A good receive file has been created. Now start checking hashes
sub start_check_hash {
    print STDERR "start_check_hash\n";
    my __PACKAGE__ $client = $_[HEAP];
    my ($file, $from, $to_check) = @_[ARG0, ARG1, ARG2];
    my $old_to_check = @{$client->{to_check}};
    push @{$client->{to_check}}, [$file, $_] for $from..$from+$to_check-1;
    $client->start_one_check if $old_to_check == 0 && @{$client->{to_check}};
}

# Start checking a file piece.
# Should only be called if there is at least one piece to be checked
sub start_one_check {
    my __PACKAGE__ $client = shift;

    # Randomly select some piece to be checked
    my $i = int rand(@{$client->{to_check}} || die "No pieces to be checked");
    my $check = $client->{to_check}[$i];
    if ($i != 0) {
        $client->{to_check}[$i] = $client->{to_check}[0];
        $client->{to_check}[0]  = $check;
    }
    $check->[0]->check_one_hash($check->[1]);
}

# A block to check becomes readable
sub check_hash {
    (my __PACKAGE__ $client, my $fh) = @_[HEAP, ARG0];
    die "No checks pending and still got a read" unless @{$client->{to_check}};
    if ($client->{to_check}[0][0]->check_hash($fh)) {
        # piece finished
        shift @{$client->{to_check}};
        $client->start_one_check if @{$client->{to_check}};
    }
}

# We received a complete fragment. Queue it for write
# Call as $client->downloader($file, $last, $piece_nr, $offset, $length, $data)
sub start_write_fragment {
    my __PACKAGE__ $client = shift;
    $client->{to_write}[0][0]->write_one_fragment($client->{to_write}[0]) if
        1 == push @{$client->{to_write}}, [@_];
}

sub write_fragment {
    (my __PACKAGE__ $client, my $fh) = @_[HEAP, ARG0];
    die "No writejob pending" unless @{$client->{to_write}};
    return unless $client->{to_write}[0][0]->write_fragment($fh);
    # This was the last subfragment
    shift @{$client->{to_write}};
    $client->{to_write}[0][0]->write_one_fragment($client->{to_write}[0]) if
        @{$client->{to_write}};
}

sub free_download_slots {
    my __PACKAGE__ $client = shift;
    my $free = $client->{max_downloads} - $client->{nr_downloads};
    return $free < 0 ? 0 : $free;
}

# A new downloader has been activated
sub start_download {
    my __PACKAGE__ $client = shift;
    die "Too many downloads" if
        ++$client->{nr_downloads} > $client->{max_downloads};
    print STDERR "Starting download. There are $client->{nr_downloads} now\n";
    if ($client->{nr_downloads} == $client->{max_downloads}) {
        # Download list is full. Drop interest
        $_->drop_interested for values %{$client->{downloaders}};
    }
}

# A downloader finished his task
# Annoying but real scenario: 
#  - full downloaders, one chokes
#  - local logic still sees full slots -> loose interest
#  - stop download, no ne, still interest -> get interested again
sub stop_download {
    my __PACKAGE__ $client = shift;
    die "Negative downloads" if --$client->{nr_downloads} < 0;
    print STDERR "Stopping download. There are $client->{nr_downloads} now\n";
    if ($client->{nr_downloads} == $client->{max_downloads}-1) {
        my @peers;
        push @peers, $_->check_interested for 
            values %{$client->{downloaders}};
        # check if any peer connections that want to talk are unchoked
        if (my @good_peers = grep $_->remote_unchoked, @peers) {
            # Temporarily increase max to avoid a "not interested" storm
            $client->{max_downloads}++;
            # Make this better than random !
            $good_peers[rand(@good_peers)]->interested;
            die "Mmm, didn't pick up a new download" unless
                --$client->{max_downloads} == $client->{nr_downloads}
        } else {
            $_->interested for @peers;
        }
    }
}

# Debugging print
sub print : method {
    print STDERR shift->string;
}

sub string {
    my __PACKAGE__ $client = shift;
    my $str = "Client\n";
    $str .= $_->string for values %{$client->{downloaders}};
    $str =~ s/\n^/\n    /mg;
    return $str;
}

package POE::Component::BitTorrent::Connection;
use Socket;
use Errno;
use POE;

use fields qw(socket want_octets in_buffer out_buffer assigned
              client downloader peer peer_string pending_fragments messages
              connecter local_interested local_choked pending_requests
              remote_choked remote_interested remote_ip remote_port
              close_on_empty last_out_empty last_in last_logical_in);

sub new {
    my ($class, $client, $socket, $remote_ip, $remote_port, $handshake) = @_;
    my __PACKAGE__ $connection = bless [], $class;

    # In the long run get rid of lost connections
    # setsockopt($socket, SOL_SOCKET, SO_KEEPALIVE, 1) ||
    #    die "Could not set keepalive on socket: $!";

    $connection->{socket} = $socket;
    $connection->{client} = $client;
    $connection->{in_buffer} = "";
    $connection->{peer_string} = "unknown";
    $connection->{local_choked}     = $connection->{remote_choked} = 1;
    $connection->{local_interested} = $connection->{remote_interested} = 0;
    $connection->{remote_port} = $remote_port;
    $connection->{remote_ip}	 = $remote_ip;
    $connection->{pending_requests} = [];
    $connection->{pending_fragments} = [];
    $connection->{messages} = "";
    $connection->{last_out_empty} = 0;
    $connection->{last_in} = $connection->{last_logical_in} = 0;
    $poe_kernel->select_read($socket,  "get_handshake");
    if ($handshake) {
        # We are the connecting side
        $connection->{connecter} = 1;
        $connection->{out_buffer} = $handshake;
        $poe_kernel->select_write($socket, "peer_send");
    } else {
        $connection->{connecter} = 1;
    }
    return $connection;
}

sub heartbeat {
    (my __PACKAGE__ $connection, my $now) = @_;
    if ($connection->{last_in} + $IDLE_TIMEOUT <= $now ||
        $connection->{last_logical_in} + $LOGICAL_IDLE_TIMEOUT <= $now) {
        $connection->close;
        return;
    }
    $connection->send_message($KEEPALIVE) if $connection->{out_buffer} eq "" &&
        $connection->{last_out_empty} + $KEEPALIVE_PERIOD <= $now &&
        $connection->{assigned};
}

sub send : method {
    my __PACKAGE__ $connection = shift;
    my $rc = syswrite($connection->{socket}, $connection->{out_buffer});
    if ($rc) {
        print STDERR "Sent $rc bytes to peer $connection->{peer_string}\n";
        substr($connection->{out_buffer}, 0, $rc) = "";
        return if $connection->{out_buffer} ne "";
        $connection->{last_out_empty} = time;
        $poe_kernel->select_write($connection->{socket});
        $connection->close if $connection->{close_on_empty};
    } else {
        if (defined($rc)) {
            return if $!{EAGAIN} || $!{EINTR};
            print STDERR "Error $! on peer socket. Closing\n";
        } else {
            # Zero write on a socket. Weird stuff
            print STDERR "Zero peer socket write. Giving up\n";
        }
        $connection->close;
        # Should also blacklist this peer !
    }
}

sub get_handshake {
    my __PACKAGE__ $connection = shift;
    my $rc = sysread($connection->{socket}, $connection->{in_buffer},
                     $BUF_SIZE, length($connection->{in_buffer}));
    if ($rc) {
        $connection->{last_in} = time;
        if (length($connection->{in_buffer}) >= 48) {
            # We have enough to identify the wanted file
            if (!$connection->{assigned}) {
                $connection->{last_logical_in} = $connection->{last_in};
                if (substr($connection->{in_buffer}, 0, 20) ne 
                    $TORRENT_ANNOUNCE) {
                    print STDERR "Non torrent connect\n";
                    $connection->close;
                }
                if (substr($connection->{in_buffer}, 20, 8) ne 
                    $TORRENT_EXTENSION) {
                    print(STDERR "Warn: Peer sends unknown extension ",
                          unpack("H*", substr($connection->{in_buffer},20,8)),
                          "\n");
                }
                if ($connection->{downloader}) {
                    my $file_id = $connection->{downloader}->file_id;
                    if ($file_id ne substr($connection->{in_buffer}, 28, 20)) {
                        # Unexpected file_id from connectee. Drop connection
                        print STDERR "Unexpected file id\n";
                        $connection->close;
                        # Should also blacklist this peer !
                    }
                } else {
                    # We are the listening side
                    my $file_id = substr($connection->{in_buffer}, 28, 20);
                    if ($connection->{downloader} =
                        $connection->{client}->file_id2downloader($file_id)) {
                        $connection->{out_buffer} =
                            $TORRENT_ANNOUNCE . $TORRENT_EXTENSION .
                            $file_id . $connection->{client}->id();
                        $poe_kernel->select_write($connection->{socket},
                                                  "peer_send");
                        my $have_bitmap=$connection->{downloader}->have_bitmap;
                        $connection->send_message($BIT_FIELD, $have_bitmap) if
                            $have_bitmap =~ /[^\x00]/;
                    } else {
                        # Unknown file_id from connector. Drop connection
                        print STDERR "Unknown file id\n";
                        $connection->close;
                        # Should also blacklist this peer !
                    }
                }

                $connection->{assigned} = 1;
            }
            if (length($connection->{in_buffer}) >= 68) {
                $connection->{last_logical_in} = $connection->{last_in};
                # We also have the peer ID. Handshake complete
                my $peer_id = substr($connection->{in_buffer}, 48, 20);
                my $peer = $connection->{downloader}->announce_peer($peer_id);
                $connection->{peer} = $peer;
                $connection->{peer_string} =
                    "$connection->{remote_ip}:$connection->{remote_port} " .
                    $peer->id_string;
                $peer->add_connection($connection);
                print STDERR "Peer is called $connection->{peer_string}\n";
                substr($connection->{in_buffer}, 0, 68) = "";
                $poe_kernel->select_read($connection->{socket});
                $connection->{want_octets} = 4;
                $poe_kernel->select_read($connection->{socket}, "get_message");
            }
        };
    } else {
        if (defined($rc)) {
            # EOF
            print STDERR "EOF from peer $connection->{peer_string}\n";
        } else {
            # Some sort of error
            return if $!{EAGAIN} || $!{EINTR};
            print STDERR "Error $! on peer socket. Closing\n";
        }
        $connection->close;
    }
}

my @type2name;
$type2name[$CHOKE]	= "CHOKE";
$type2name[$UNCHOKE]	= "UNCHOKE";
$type2name[$INTERESTED]	= "INTERESTED";
$type2name[$NOT_INTERESTED] = "NOT INTERESTED";
$type2name[$HAVE]	= "HAVE";
$type2name[$BIT_FIELD]	= "BIT_FIELD";
$type2name[$REQUEST]	= "REQUEST";
$type2name[$PIECE]	= "PIECE";
$type2name[$CANCEL]	= "CANCEL";

sub get_message {
    my __PACKAGE__ $connection = shift;
    my $rc = sysread($connection->{socket}, $connection->{in_buffer},
                     $BUF_SIZE, length($connection->{in_buffer}));
    if (!$rc) {
        if (defined($rc)) {
            # EOF
            print STDERR "EOF from peer $connection->{peer_string}\n";
            $connection->close;
            return;
        }
        # Some sort of error
        return if $!{EAGAIN} || $!{EINTR};
        die "Error $! on peer socket. Closing\n";
    }
    $connection->{last_in} = time;
    while (length($connection->{in_buffer}) >= $connection->{want_octets}) {
        if ($connection->{want_octets} == 4) {
            my $more = unpack("N", $connection->{in_buffer});
            if ($more == 0) {
                # keepalive
                print STDERR "Keepalive from $connection->{peer_string}\n";
                substr($connection->{in_buffer}, 0, 4) = "";
                next;
            }
            die "Insanely big message ($more bytes)\n" if $more > $MAX_MESSAGE;
            $connection->{want_octets} += $more;
            $connection->{last_logical_in} = $connection->{last_in};
            next;
        }
        $connection->{last_logical_in} = $connection->{last_in};
        my $type = ord substr($connection->{in_buffer}, 4, 1);
        print(STDERR $type2name[$type] || "Unknown type $type",
              " message from $connection->{peer_string}\n");
        if ($type < 6) {
            if ($type < 4) {
                if ($type < 2) {
                    if ($type == 0) {
                        # type 0, choke
                        $connection->remote_choke if 
                            !$connection->{remote_choked};
                    } elsif ($connection->{remote_choked}) {
                        # type 1, unchoke
                        $connection->{remote_choked} = 0;
                        $connection->{peer}->remote_unchoke
                            ($connection, $connection->{local_interested});
                    }
                } elsif ($type == 2) {
                    # type 2, interested
                    if (!$connection->{remote_interested}) {
                        $connection->{remote_interested} = 1;
                    }
                } elsif ($connection->{remote_interested}) {
                    # type = 3, not interested
                    $connection->{remote_interested} = 0;
                }
            } elsif ($type == 4) {
                # type 4, have
                my $have = substr($connection->{in_buffer},
                                  0, $connection->{want_octets}, "");
                $connection->{want_octets} = 4;
                $connection->close_on_empty unless
                    $connection->{peer}->got_have(unpack("x5N", $have));
                next;
            } else {
                # type = 5, bitfield
                my $bitmap = substr($connection->{in_buffer},
                                    0, $connection->{want_octets}, "");
                $connection->{want_octets} = 4;
                substr($bitmap, 0, 5) = "";
                $connection->close_on_empty unless
                    $connection->{peer}->got_bitmap($bitmap);
                next;
            }
        } elsif ($type > 7) {
            die "Unknown message type $type\n" if $type > 8;
            # type 8, cancel
        } elsif ($type == 6) {
            # type 6, request
        } else {
            # type = 7, piece
            my $fragment=substr($connection->{in_buffer}, 0,
                                $connection->{want_octets},"");
            $connection->got_fragment(unpack("x5NN", 
                                             substr($fragment,0,4+1+8,"")),
                                $connection->{want_octets}-4-1-8, $fragment);
            $connection->{want_octets} = 4;
            next;
        }
        substr($connection->{in_buffer}, 0, $connection->{want_octets}) = "";
        $connection->{want_octets} = 4;
    }
}

# We are loosing the pending fragment requests (choke or close)
sub remote_choke {
    my __PACKAGE__ $connection = shift;
    die "remote_choke called while remote already choked" if
        $connection->{remote_choked};
    $connection->{remote_choked} = 1;
    # It would be nice to first return the pieces without however
    # causing a new request to start on the current connection
    $connection->{peer}->stop_download($connection) if 
        $connection->{local_interested};
    if (@{$connection->{pending_requests}}) {
        my $requests = $connection->{pending_requests};
        $connection->{pending_requests} = [];
        $connection->{downloader}->drop_requests($requests);
    }
}

my @message_templates;
$message_templates[$CHOKE]	= "";
$message_templates[$UNCHOKE]	= "";
$message_templates[$INTERESTED]	= "";
$message_templates[$NOT_INTERESTED] = "";
$message_templates[$HAVE]	= "N";
$message_templates[$BIT_FIELD]	= "a*";
$message_templates[$REQUEST]	= "NNN";
# $PIECE is special
$message_templates[$CANCEL]	= "NNN";

# Send a message to a peer
sub send_message {
    my __PACKAGE__ $connection = shift;
    my $type = shift;
    print(STDERR "Queueing a ", $type2name[$type] ||
          ($type == $KEEPALIVE ? "keepalive" : "unknown type $type"),
          " message for $connection->{peer_string}\n");
    if (defined($message_templates[$type])) {
        if ($type == $INTERESTED) {
            die "Was already interested" if $connection->{local_interested};
            $connection->{local_interested} = 1;
            $connection->{peer}->start_download($connection) unless 
                $connection->{remote_choked};
        } elsif ($type == $NOT_INTERESTED) {
            # Should never be called while the peer has a download_connection
            die "Wasn't already interested" if 
                !$connection->{local_interested};
            $connection->{local_interested} = 0;
            # We don't check for remote unchoked here. The caller must already
            # have handled that (see Peer::got_fragment)
        } elsif ($type == $REQUEST) {
            push(@{$connection->{pending_requests}}, [@_]);
        } elsif ($type == $HAVE) {
            $connection->close_on_empty if $_[1];
        }

        # Don't actually try to send anything if the socket is already closed
        return if !$connection->{socket};

        my $msg = pack($message_templates[$type], @_);
        if (@{$connection->{pending_fragments}}) {
            $connection->{messages} .= pack("NC", 1+length($msg), $type).$msg;
        } else {
            $poe_kernel->select_write($connection->{socket}, "peer_send") if 
                $connection->{out_buffer} eq "";
            $connection->{out_buffer} .= 
                pack("NC", 1+length($msg), $type) . $msg;
        }
    } elsif ($type == $PIECE) {
        # Don't send anything if the socket is already closed
        return unless $connection->{socket};
        die "Piece is not yet implemented";
    } elsif ($type == $KEEPALIVE) {
        die "Keepalive has no payload" if @_;
        $connection->{out_buffer} eq "" ||
            die "Keepalive requested, but out_buffer non-empty";

        # Don't send anything if the socket is closing
        return if !$connection->{socket} || $connection->{close_on_empty};

        $connection->{out_buffer} = pack("N", 0);
        $poe_kernel->select_write($connection->{socket}, "peer_send");
    } else {
        die "Unknown message type $type";
    }
}

sub download_preference {
    my __PACKAGE__ $connection = shift;
    return
        # Remote unchoked is very good: immediate request possible
        4*!$connection->{remote_choked} +
        # Remote interested means a risk of having our requests buried in bulk
        # (on the other hand it means simple tit-for-tat)
        2*!$connection->{remote_interested} +
        # Otherwise prefer connections we initiated
        $connection->{connecter};
}

# Loose interest on this connection
# Returns true if we actually removed an interest
# May only be called if the peer has no download_connection !
sub not_interested {
    my __PACKAGE__ $connection = shift;
    return unless $connection->{local_interested};
    $connection->send_message($NOT_INTERESTED);
    return 1;
}

# Query the pending requests. Returns the amount in scalar context
sub pending_requests {
    my __PACKAGE__ $connection = shift;
    return @{$connection->{pending_requests}};
}

sub got_fragment {
    my __PACKAGE__ $connection = shift;
    my $piece_nr = shift;
    my $offset   = shift;
    my $length   = shift;
    for (0..$#{$connection->{pending_requests}}) {
        if ($connection->{pending_requests}[$_][0] == $piece_nr &&
            $connection->{pending_requests}[$_][1] == $offset &&
            $connection->{pending_requests}[$_][2] == $length) {
            $connection->{pending_requests}[$_] = 
                $connection->{pending_requests}[-1];
            pop @{$connection->{pending_requests}};
            $connection->{downloader}->got_fragment($piece_nr, $offset,
                                                    $length, shift);
            $connection->{peer}->got_fragment
                ($connection, scalar @{$connection->{pending_requests}}) if
                $connection->{local_interested}&&!$connection->{remote_choked};

            # Debugging: Ever so often just loose the connection
            $connection->close if 
                !POE::Component::BitTorrent::CLEAN && rand(30) < 1;

            return;
        }
    }
    # Make this serendipity do something intelligent at some point
    print STDERR "Unexpected piece $piece_nr from $offset for $length from $connection->{peer_string}\n";
}

# Friendly close. Wait for output buffer to become empty
sub close_on_empty {
    my __PACKAGE__ $connection = shift;
    print STDERR "Going to close $connection->{peer_string}\n";
    # Nothing to be done if already closed or already pending
    return if !$connection->{socket} || $connection->{close_on_empty};
    $connection->{close_on_empty} = 1;
    return if $connection->{out_buffer} ne "";
    # Othwerwise queue (gives pending code a chance to fill the output queue)
    $poe_kernel->yield("check_empty", $connection);
}

# Check if remote_choked
sub remote_choked {
    my __PACKAGE__ $connection = shift;
    return $connection->{remote_choked};
}

sub check_empty {
    my __PACKAGE__ $connection = shift;
    return if !$connection->{socket};
    return if $connection->{out_buffer} ne "";
    $connection->close();
}

sub close : method {
    my __PACKAGE__ $connection = shift;
    my $socket = $connection->{socket} || return;
    $connection->{socket} = undef;
    # Flush any pending get_messages
    $connection->{in_buffer} = "";
    $poe_kernel->select_read($socket);
    $poe_kernel->select_write($socket) if $connection->{out_buffer} ne "";
    print STDERR "Closing connection to $connection->{peer_string}\n";

    close($socket);
    $connection->remote_choke if !$connection->{remote_choked};
    $connection->{peer}->delete_connection
        ($connection, $connection->{local_interested}) if $connection->{peer};
    $connection->{client}->lost_connection($socket);
}

sub string {
    my __PACKAGE__ $connection = shift;
    my $str = "Connection to $connection->{peer_string} ";
    $str.=", ".($connection->{local_interested} ? "" : "!")."LocalInterested";
    $str.=", ".($connection->{local_choked}     ? "" : "!")."LocalChoked";
    $str.=", ".($connection->{remote_interested}? "" : "!")."RemoteInterested";
    $str.=", ".($connection->{remote_choked}    ? "" : "!")."RemoteChoked";
    return "$str\n";
}

sub print : method {
    print STDERR shift->string;
}

package POE::Component::BitTorrent::Peer;
use fields qw(downloader have_bitmap nr_pieces addresses connections
              local_wanted id id_string download_connection nr_needed_pieces);

# local_wanted = global_wanted & peer_have, undef if empty
# local_interested if and only if
#    connections & local_wanted & free_slots | download_connection
# local_interest is asserted on at most one of the peer connections
# The download connection is always interested
# Work is queued as much as allowed and possible on the download_connection

sub new {
    my ($class, $downloader, $peer_id, $nr_pieces) = @_;
    my __PACKAGE__ $peer = bless [], $class;
    $peer->{downloader} = $downloader;
    $peer->{id} = $peer_id;
    $peer->{nr_pieces} = $peer->{nr_needed_pieces} = $nr_pieces;
    $peer->{id_string} = id_string($peer_id);
    $peer->{addresses}   = {};
    $peer->{connections} = {};
    $peer->{have_bitmap} = "";
    $peer->{download_connection} = 0;
    return $peer;
}

sub announce_address {
    (my __PACKAGE__ $peer, my $address) = @_;
    $peer->{addresses}{$address} = 1;
}

# An address is not known anymore
# Typically caused by something else now occupying that address or
# a connection attempt failing or getting the wrong peer_id
sub delete_address {
    (my __PACKAGE__ $peer, my $address) = @_;
    die "Dropping an unknown address" unless
        delete $peer->{addresses}{$address};
    $peer->{downloader}->delete_peer($peer->{id}) if
        !%{$peer->{addresses}} && !%{$peer->{connections}};
}

# We have a connection to this peer
sub add_connection {
    (my __PACKAGE__ $peer, my $connection) = @_;
    if (!%{$peer->{connections}} && defined($peer->{local_wanted}) &&
        $peer->{downloader}->free_download_slots) {
        # This is the first connection, we want something from this peer
        # and there is place for a download
        $peer->{connections}{$connection} = $connection;
        $peer->interested;
    } else {
        $peer->{connections}{$connection} = $connection;
    }
}

# A connection is closed
sub delete_connection {
    (my __PACKAGE__ $peer, my $connection, my $local_interested) = @_;
    die "Deleting download connection (should have been remote_choked)" if
        $peer->{download_connection} == $connection;
    die "Dropping an unknown connection" unless
        delete $peer->{connections}{$connection};
    # If we loose our interested connection, try to transfer interest
    if ($local_interested) {
        die "No record of why we are interested" if
            !defined($peer->{local_wanted});
        $peer->interested if %{$peer->{connections}};
    }
    # Forget about this peer if we don't know how to reach it anymore
    $peer->{downloader}->delete_peer($peer->{id}) if
        !%{$peer->{addresses}} && !%{$peer->{connections}};
}

# A peer gave us his piece bitmap
# Return true if we have anything to exchange with this peer
sub got_bitmap {
    (my __PACKAGE__ $peer, my $bitmap) = @_;
    if (length($bitmap) > $peer->{nr_pieces}/8) {
        die "Bitmap too long" if length($bitmap) > int($peer->{nr_pieces}/8)+1;
        die "Trailing extra ones in bitmap" if
            ord(substr($bitmap, -1)) & (0xff >> $peer->{nr_pieces} % 8);
    } elsif (length($bitmap) < int((7+$peer->{nr_pieces}) / 8)) {
        $bitmap .= "\x00" x (int((7+$peer->{nr_pieces}) / 8)-length($bitmap));
    }
    if ($peer->{have_bitmap} ne $bitmap) {
        $peer->{nr_needed_pieces} =
            $peer->{nr_pieces} - unpack("%32B*", $bitmap);
        die "Duh, need negative pieces ?" if $peer->{nr_needed_pieces} < 0;
        $peer->{have_bitmap} = $bitmap;
        $bitmap &= $peer->{downloader}->want_bitmap;
        if ($bitmap =~ /[^\x00]/) {
            # Peer has stuff
            if (defined($peer->{local_wanted})) {
                # We were already wanted stuff from this peer,
                # only the pieces (maybe) changed
                if ($peer->{local_wanted} ne $bitmap) {
                    if ($peer->{download_connection}) {
                        # We should drop any pending lost stuff here
                        my $new = ~$peer->{local_wanted} & $bitmap;
                        $peer->{local_wanted} = $bitmap;
                        $peer->push if $new =~ /[^\x00]/;
                    } else {
                        $peer->{local_wanted} = $bitmap;
                    }
                }
            } else {
                # Aha, for the first time we want something from this peer
                $peer->{local_wanted} = $bitmap;
                if ($peer->{download_connection}) {
                    $peer->push;
                } elsif ($peer->{downloader}->free_download_slots) {
                    # And there is place for a download
                    $peer->interested;
                }
            }
        } elsif (defined($peer->{local_wanted})) {
            # Peer has nothing we want (anymore)
            $peer->not_wanted;
        }
    }
    return $peer->{nr_needed_pieces} || $peer->{downloader}->nr_needed_pieces;
}

# Received "HAVE" message from a peer
# Return true if we have anything to exchange with this peer
sub got_have {
    (my __PACKAGE__ $peer, my $piece_nr) = @_;
    print STDERR "Have piece $piece_nr\n";
    if (!vec($peer->{have_bitmap}, $piece_nr ^ 7, 1)) {
        # Only needs work if its a new piece
        vec($peer->{have_bitmap}, $piece_nr ^ 7, 1) = 1;
        die "Negative need ???" if --$peer->{nr_needed_pieces} < 0;
        if ($peer->{downloader}->want_piece($piece_nr)) {
            # Only interesting if the downloader wants it
            if (defined($peer->{local_wanted})) {
                # We were already wanted stuff from this peer,
                # just a new piece
                vec($peer->{local_wanted}, $piece_nr ^ 7, 1) = 1;
                $peer->push if $peer->{download_connection};
            } else {
                # Aha, for the first time we want something from this peer
                $peer->{local_wanted} = "\x00" x ((7+$peer->{nr_pieces}) / 8);
                vec($peer->{local_wanted}, $piece_nr ^ 7, 1) = 1;
                if ($peer->{download_connection}) {
                    $peer->push;
                } elsif ($peer->{downloader}->free_download_slots) {
                    # And there is place for a download
                    $peer->interested;
                }
            }
        }
    }
    return $peer->{nr_needed_pieces} || $peer->{downloader}->nr_needed_pieces;
}

# We became interested in this peer while we didn't use to be.
# Announce this on at least one connection. 
# MUST be remote_unchoked if any connection is remote_unchoked.
sub interested {
    my __PACKAGE__ $peer = shift;
    die "Interest without reason" if !defined($peer->{local_wanted});
    die "Interest while already downloading" if $peer->{download_connection};
    # The common case of only one connection is easy
    if (keys %{$peer->{connections}} == 1) {
        (values %{$peer->{connections}})[0]->send_message($INTERESTED);
        return;
    }
    # Multiple connections. Choose one
    my @good_cons;
    my $pref = -1;
    for (values %{$peer->{connections}}) {
        my $p = $_->download_preference;
        if ($p > $pref) {
            @good_cons = ($_);
            $pref = $p;
        } elsif ($p == $pref) {
            push @good_cons, $_;
        }
    }
    die "Called interested without connections" unless @good_cons;
    # Choose one at random
    $good_cons[rand @good_cons]->send_message($INTERESTED);
}

# Local_wanted went from a value to undef. Check if we need to drop interest.
sub not_wanted {
    my __PACKAGE__ $peer = shift;
    die "Calling not_wanted while already not interested" if
        !defined($peer->{local_wanted});
    $peer->{local_wanted} = undef;
    return if $peer->{download_connection} || !%{$peer->{connections}};
    $_->not_interested for values %{$peer->{connections}};
}

# Exactly one download slot just came free. Return peers that would 
# be interested
sub check_interested {
    my __PACKAGE__ $peer = shift;
    return if !%{$peer->{connections}} || $peer->{download_connection} ||
        !defined($peer->{local_wanted});
    return $peer;
}

# Check if any of this peers connections are unchoked
sub remote_unchoked {
    my __PACKAGE__ $peer = shift;
    for (values %{$peer->{connections}}) {
        return 1 unless $_->remote_choked;
    }
    return 0;
}

# All download slots are filled. Drop interest
sub drop_interested {
    my __PACKAGE__ $peer = shift;
    return if !%{$peer->{connections}} || $peer->{download_connection} ||
        !defined($peer->{local_wanted});
    die "Wrong interest count for $peer->{id_string}" unless 
        1 == grep $_->not_interested, values %{$peer->{connections}};
}

# One of our connections got unchoked
sub remote_unchoke {
    (my __PACKAGE__ $peer, my $connection, my $local_interested) = @_;
    die "Why was there interest on a choked connection while a downloader ?" if
        $peer->{download_connection};
    # Were we already interested in this connection ?
    if ($local_interested) {
        $peer->start_download($connection);
        return;
    }
    # Check if one of the other connections maybe was interested
    return unless defined($peer->{local_wanted}) &&
        $peer->{downloader}->free_download_slots;
    # Yup, move the interest quickly before the unchoke goes away
    # Mmm, doesn't this risk being transiently not interested and
    # triggering a choke ?
    for (values %{$peer->{connections}}) {
        next if $_ == $connection || !$_->not_interested;
        # The send_message will trigger the start_download
        $connection->send_message($INTERESTED);
        return;
    }
    die "No interested sockets while local wanted is set";
}

# This connection just became the download_connection. Give it work.
sub start_download {
    (my __PACKAGE__ $peer, my $connection) = @_;
    die "Already a downloader" if $peer->{download_connection};
    die "start_download while not interested" unless
        defined($peer->{local_wanted});
    $peer->{download_connection} = $connection;
    $peer->{downloader}->start_download
        (1, $connection, $peer->{local_wanted});
}

# At least one new piece became wanted on a download connection
# Returns false if no piece was queued, true otherwise 
# (but it can be a partial queue, so maybe the piece is still wanted)
sub push : method {
    my __PACKAGE__ $peer = shift;
    die "No downloader" unless $peer->{download_connection};
    die "Nothing wanted" unless defined($peer->{local_wanted});
    return 0 if 
        $peer->{download_connection}->pending_requests >= $MAX_REQUESTS;
    $peer->{downloader}->start_download
        (0, $peer->{download_connection}, $peer->{local_wanted});
}

# Got a fragment on the download_connection
sub got_fragment {
    (my __PACKAGE__ $peer, my $connection, my $left) = @_;
    die "No current downloader" if !$peer->{download_connection};
    die "Wrong downloader" if $peer->{download_connection} != $connection;
    if (defined($peer->{local_wanted})) {
        die "Why is there so little work left on $peer->{id_string}?" if 
            $left < $MAX_REQUESTS - 1;
        # Still work on this connection
        $peer->{downloader}->start_download
            (0, $connection, $peer->{local_wanted});
    } elsif ($left == 0) {
        # Just finished the last task on the download_connection
        $peer->{download_connection} = 0;
        $connection->send_message($NOT_INTERESTED);
        $peer->{downloader}->stop_download;
    }
}

# We got a remote choke (and we are local_interested)
# (loss of local interest is already handled in got_fragment)
sub stop_download {
    (my __PACKAGE__ $peer, my $connection) = @_;
    die "I wasn't downloading" unless $peer->{download_connection};
    die "Wrong connection stops" unless
        $peer->{download_connection} == $connection;
    $peer->{download_connection} = 0;
    $connection->send_message($NOT_INTERESTED) unless
        defined($peer->{local_wanted}) && 
        $peer->{downloader}->free_download_slots;
    $peer->{downloader}->stop_download;
}

sub have_count {
    my $pieces = shift;
    return map {
        my $count = 0;
        for my __PACKAGE__ $connection (@_) {
            $count++ if vec($connection->{have_bitmap}, $_, 1);
        }
        $count;
    } @$pieces;
}

# Downloader lost interest in a given piece
sub stop_wanting {
    my __PACKAGE__ $peer = shift;
    my $piece_nr = shift;
    # Return if we didn't want that piece in the first place
    return unless defined($peer->{local_wanted}) &&
        vec($peer->{local_wanted}, $piece_nr ^ 7, 1);
    vec($peer->{local_wanted}, $piece_nr ^ 7, 1) = 0;
    # If we still want stuff, we can remain interested
    return if $peer->{local_wanted} =~ /[^\x00]/;
    # Ok, we just lost interest in this peer
    $peer->not_wanted;
}

# Downloader gained interest in a list of pieces in LSB format
# (hash check finished unsuccesfully or HAVE message from peer)
# Remember that calling this can actually make the pieces unwanted if
# all remaining fragments get queued
# Returns true if a piece may have been absorbed
# Returns false if it's certain no piece was absorbed
sub start_wanting {
    my __PACKAGE__ $peer = shift;
    if (my @pieces = grep vec($peer->{have_bitmap}, $_, 1), @_) {
        # Only do work if the remote has it
        if (defined($peer->{local_wanted})) {
            # There was already stuff we wanted. Just update
            vec($peer->{local_wanted}, $_, 1) = 1 for @pieces;
            return $peer->{download_connection} && $peer->push;
        }
        # A completely new interest develops
        $peer->{local_wanted} = "\x00" x ((7+$peer->{nr_pieces}) / 8);
        vec($peer->{local_wanted}, $_, 1) = 1 for @pieces;
        return $peer->push if $peer->{download_connection};
        # And there is place for a download
        $peer->interested if %{$peer->{connections}} &&
            $peer->{downloader}->free_download_slots;
        return $peer->{download_connection};
    }
    return;
}

# A piece was completed. Tell everybody
sub announce_piece {
    (my __PACKAGE__ $peer, my $piece_nr, my $nr_needed) = @_;
    my $done = $nr_needed == 0 && $peer->{nr_needed_pieces} == 0;
    $_->send_message($HAVE,$piece_nr,$done) for values %{$peer->{connections}};
}

sub id_string {
    my __PACKAGE__ $peer = shift;
    return $peer->{id_string} if ref($peer);
    # return unpack("H*", $peer);
    my $str = pack("u", $peer);
    $str =~ s/\n\z//;
    return $str;
}

sub string {
    my __PACKAGE__ $peer = shift;
    my $nr = keys %{$peer->{connections}};
    my $str = sprintf("Peer: %s (%d connection%s, %s downloader)\n",
                      $peer->{id_string}, $nr, $nr == 1 ? "" : "s",
                      $peer->{download_connection} ? "has" : "no");
    $str .= sprintf("Peer bitmap: %s, needs %d piece%s\n",
                    defined($peer->{have_bitmap}) ?
                    unpack("H*", $peer->{have_bitmap}) : "unknown",
                    $peer->{nr_needed_pieces},
                    $peer->{nr_needed_pieces} == 1 ? "" : "s");
    $str .= "Want bitmap: " .
        (defined($peer->{local_wanted}) ? unpack("H*", $peer->{local_wanted}) :
         "none") . "\n";
    $str .= "address $_\n" for keys %{$peer->{addresses}};
    $str .= $_->string for values %{$peer->{connections}};
    $str =~ s/\n^/\n    /mg;
    return $str;
}

sub print : method {
    print STDERR shift->string;
}

package POE::Component::BitTorrent::File;
use Fcntl qw(:DEFAULT :seek);
use Digest::SHA1;
use POE;

use fields qw(handle_cache
              check_handle check_pos check_left check_hash check_piece
              write_handle write_left write_fragment
              downloader name piece_length length dev ino io_block);

sub new {
    my ($class, $downloader, $name, $piece_length) = @_;
    my __PACKAGE__ $file = bless [], $class;
    $file->{name} = $name;
    $file->{downloader} = $downloader;
    $file->{piece_length} = $piece_length;
    return $file;
}

sub open : method {
    my __PACKAGE__ $file = shift;

    if ($file->{handle_cache}) {
        my $fh = $file->{handle_cache};
        $file->{handle_cache} = undef;
        return $fh
    }
    sysopen(my $fh, $file->{name}, O_RDWR | O_CREAT) ||
        die "Could not open $file->{name}: $!";
    binmode($fh);
    my @stat = stat($fh) or die "Could not stat $file->{name}: $!";
    if (defined($file->{ino})) {
        $file->{dev} == $stat[0] || die "Device of $file->{name} changed";
        $file->{ino} == $stat[1] || die "Inode  of $file->{name} changed";
    } else {
        $file->{dev} = $stat[0];
        $file->{ino} = $stat[1];
        my $io_block = $stat[11] || 0;
        $io_block = 4096 if $io_block < 512;
        $file->{io_block} = $io_block;
    }
    return $fh;
}

# Called when you are not interested in a handle anymore
# Maybe should set up a timeout for the handle_cache....
sub drop {
    my __PACKAGE__ $file = shift;
    $file->{handle_cache} ||= shift;
}

# Return how many to check. The rest is guaranteed incomplete
sub set_length {
    (my __PACKAGE__ $file, my $length) = @_;
    $file->{"length"} = $length;
    my $fh = $file->open;
    my $to_check;
    if (-s _ < $length) {
        $to_check = int((-s _) / $file->{piece_length});
    } else {
        if (-s _ > $length) {
            truncate($fh, $length) ||
                die "Could not truncate $file->{name}: $!";
        }
        $to_check= $length ?
            int(($length+$file->{piece_length}-1)/$file->{piece_length}) :
            0;
    }
    $poe_kernel->yield("start_check_hash", $file, 0, $to_check);
    $file->drop($fh);
    return $to_check;
}

# Start checking the given piece
sub check_one_hash {
    (my __PACKAGE__ $file, my $piece_nr) = @_;
    die "Already a check pending" if $file->{check_handle};

    $file->{check_piece} = $piece_nr;
    $file->{check_pos} = $piece_nr * $file->{piece_length};
    my $to_check = $file->{"length"}-$file->{check_pos}<$file->{piece_length} ?
        $file->{"length"}-$file->{check_pos} : $file->{piece_length};
    $file->{check_left} = $to_check || die "Request to check length 0";
    $file->{check_hash} = Digest::SHA1->new;
    $file->{check_handle} = $file->open;
    sysseek($file->{check_handle}, $file->{check_pos}, SEEK_SET) ||
        die "Could not seek to offset $file->{check_pos}: $!";
    $poe_kernel->select_read($file->{check_handle}, "check_hash");
}

# A block to be checked became available for read
# Return 1 if we are done with the current piece
sub check_hash {
    my __PACKAGE__ $file = shift;
    die "No handle to check on" unless $file->{check_handle};
    die "Unexpected handle"	unless $file->{check_handle} == shift;
    # Want up to the next multiple of the prefered I/O block
    my $want = (1+int($file->{check_pos}/$file->{io_block}))*$file->{io_block}-
        $file->{check_pos};
    $want = $file->{check_left} if $want > $file->{check_left};
    my $rc = sysread($file->{check_handle}, my $buffer, $want);
    if ($rc) {
        $file->{check_hash}->add($buffer);
        $file->{check_pos} += $rc;
        $file->{check_left}  -= $rc;
        if ($file->{check_left} == 0) {
            # Finished
            $poe_kernel->select_read($file->{check_handle});
            $file->drop($file->{check_handle});
            $file->{check_handle} = undef;
            $file->{downloader}->checked_piece
                ($file->{check_piece}, $file->{check_hash}->digest);
            return 1;
        }
    } elsif (defined($rc)) {
        die "Unexpected EOF while checking $file->{name}";
    } elsif ($!{EAGAIN} || $!{EINTR}) {
        # Benign errors
    } else {
        die "Error $! while reading from $file->{name}";
    }
    return;
}

sub write_one_fragment {
    my __PACKAGE__ $file = shift;
    die "Already a check pending" if $file->{write_handle};
    $file->{write_fragment} = shift;
    (my $piece_nr, my $offset, $file->{write_left}) =
        @{$file->{write_fragment}}[2, 3, 4];
    my $pos = $offset + $piece_nr * $file->{piece_length};
    die "Trying to write more than the exepected size\n" if
        $pos + $file->{write_left} > $file->{length};
    $file->{write_handle} = $file->open;
    sysseek($file->{write_handle}, $pos, SEEK_SET) ||
        die "Could not seek to offset $pos}: $!";
    $poe_kernel->select_write($file->{write_handle}, "write_fragment");
}

sub write_fragment {
    my __PACKAGE__ $file = shift;
    die "Already a check pending" unless $file->{write_handle};
    die "Write_fragment on the wrong filehandle" unless
        $file->{write_handle} == shift;

    my $rc = syswrite($file->{write_handle}, $file->{write_fragment}[5],
                      $file->{write_left}, -$file->{write_left});
    if ($rc) {
        # print STDERR "Wrote $rc bytes\n";
        $file->{write_left} -= $rc;
        if ($file->{write_left} <= 0) {
            die "What, negative left" if $file->{write_left} < 0;
            # Finished
            $poe_kernel->select_write($file->{write_handle});
            $file->drop($file->{write_handle});
            $file->{write_handle}   = undef;
            # Check hash if this fragment finishes a piece.
            $poe_kernel->yield("start_check_hash",
                               $file, $file->{write_fragment}[2], 1) if
                                   $file->{write_fragment}[1];
            $file->{write_fragment} = undef;
            return 1;
        }
    } elsif (defined($rc)) {
        die "Length 0 write";
    } elsif ($!{EAGAIN} || $!{EINTR}) {
        # Benign errors
    } else {
        die "Error $! while reading from $file->{name}";
    }
    return;
}

package POE::Component::BitTorrent::Piece;
use fields qw(nr end_offset done_offset ranges failed);

sub new {
    my ($class, $nr, $size) = @_;
    my __PACKAGE__ $piece = bless [], $class;
    $piece->{nr} = $nr;
    $piece->{end_offset}  = $size;
    $piece->{done_offset} = 0;
    $piece->{ranges} = {};
    $piece->{failed} = [];
    return $piece;
}

sub get_range {
    my __PACKAGE__ $piece = shift;
    my $from;
    if (@{$piece->{failed}}) {
        $from = $piece->{failed}[-1][0];
        $piece->{ranges}{$from} = pop @{$piece->{failed}};
    } elsif ($piece->{done_offset} >= $piece->{end_offset}) {
        die "Requesting range from empty in piece $piece->{nr}";
    } else {
        $from = $piece->{done_offset};
        $piece->{done_offset} += $PIECE_REQUEST_SIZE;
        $piece->{done_offset} = $piece->{end_offset} if
            $piece->{done_offset} > $piece->{end_offset};
        $piece->{ranges}{$from} = [$from, $piece->{done_offset}-$from];
    }
    return(@{$piece->{failed}} == 0 &&
           $piece->{done_offset} >= $piece->{end_offset},
           @{$piece->{ranges}{$from}});
}

# One of the requested ranges was received. Remove it from the list.
# Return true if this was the last range we needed to complete a piece
sub got_range {
    (my __PACKAGE__ $piece, my $from) = @_;
    defined(my $range = delete $piece->{ranges}{$from}) ||
        die "subfragment $from wasn't pending";
    return !%{$piece->{ranges}} && @{$piece->{failed}} == 0 &&
        $piece->{done_offset} >= $piece->{end_offset};
}

sub drop_range {
    (my __PACKAGE__ $piece, my $from) = @_;
    defined(my $range = delete $piece->{ranges}{$from}) ||
        die "subfragment $from wasn't pending";
    push @{$piece->{failed}}, $range;
}

package POE::Component::BitTorrent::Downloader;
# Downloaders represent attempts to fetch a given file_id
use fields qw(files piece_length piece_hashes have_bitmap want_bitmap
              partial_bitmap requested_pieces nr_needed_pieces
              file_id callbacks trackers id2peer
              left downloaded uploaded length file client);

use Digest::SHA1 qw(sha1);

my $counter = "a";

sub file_filter {
    my $name = shift;
    $name =~ s!.*/!!s;
    $name =~ s/[^\w+=:.,\-]//g;
    $name =~ s/^\.+//;
    $name = "file_" . $counter++ if $name eq "";
    return $name;
}

sub new {
    my ($class, $client, $file_id, $info) = @_;
    my __PACKAGE__ $downloader = bless [], $class;

    die "info is not a dictionary\n" if ref($info) ne "HASH";
    exists $info->{$_} || die "dictionary info has no $_ entry\n" for
        "name", "piece length", "pieces";
    die "piece length is not a natural number, but '$info->{'piece length'}'\n"
        unless $info->{"piece length"} =~ /^\d+\z/;
    $downloader->{piece_length} = $info->{"piece length"};
    die "pieces length is not a multiple of 20" if
        length($info->{pieces}) % 20;
    my @piece_hashes = $info->{pieces} =~ /(.{20})/sg;
    my $length = 0;
    if (exists $info->{length}) {
        die "dictionary info has both length and files\n" if
            exists $info->{files};
        $downloader->{files} = [file_filter($info->{name}), $info->{length}];
        $length = $info->{length};
    } elsif (exists $info->{files}) {
        my $files = $info->{files};
        die "files is not a list\n" if ref($files) ne "ARRAY";
        my $base = file_filter($info->{name});
        $downloader->{files} = [map {
            my $file = $_;
            die "some file is not a dictionary" unless ref($file) eq "HASH";
            exists $file->{$_} || die "file dictionary has no $_ entry\n" for
                qw(length path);
            my $path = $file->{path};
            ref($path) eq "ARRAY" || die "path is not a list";
            @$path || die "path is an empty list";
            $length += $file->{length};
            [join("/", $base, map file_filter($_), @$path), $file->{length}];
        } @$files];
        use Data::Dumper;
        print STDERR Dumper($downloader->{files});
    } else {
        die "dictionary info has neither length nor files\n";
    }
    if ($downloader->{piece_length}) {
        my $pieces = int(($length+$downloader->{piece_length}-1)/
                         $downloader->{piece_length});
        @piece_hashes == $pieces ||
            die("number of hashes=", scalar(@piece_hashes),
                " differs from number of pieces=$pieces (total length=$length",
                ", piece_length=$downloader->{piece_length}");
    } elsif ($length) {
        die "piece_length=0, but total expected length=$length";
    } elsif (@piece_hashes) {
        die "piece_length=0, length=0 but still have piece hashes";
    }
    $downloader->{piece_hashes} = \@piece_hashes;
    $downloader->{nr_needed_pieces} = @piece_hashes;
    $downloader->{callbacks}	= [];
    $downloader->{trackers}	= {};
    $downloader->{requested_pieces} = {};
    $downloader->{id2peer}	= {};
    $downloader->{left}	= $length;
    $downloader->{uploaded}	= $downloader->{downloaded} = 0;
    $downloader->{length}	= $length;
    $downloader->{file_id}	= $file_id;
    my $file_name = unpack("H*", $file_id);
    $downloader->{file} =
        POE::Component::BitTorrent::File->new
        ($downloader, $file_name, $downloader->{piece_length});

    $downloader->{partial_bitmap} = $downloader->{have_bitmap}  =
        "\x00" x ((7+@piece_hashes) / 8);
    my $to_check = $downloader->{file}->set_length($length);
    $downloader->{want_bitmap}  =
        ("\xff" x (@piece_hashes / 8) .
         (@piece_hashes % 8 ? ~chr(0xff >> @piece_hashes % 8) : "")) ^
        ("\xff" x ($to_check     / 8) .
         ($to_check     % 8 ? ~chr(0xff >> $to_check     % 8) : ""));
    $downloader->{client} = $client;
    return $downloader;
}

# We learn about a peer_id. Return the corresponding peer
sub announce_peer {
    my __PACKAGE__ $downloader = shift;
    my ($peer_id, $host, $port) = @_;
    my $peer = $downloader->{id2peer}{$peer_id} ||=
        POE::Component::BitTorrent::Peer->new
        ($downloader, $peer_id, scalar @{$downloader->{piece_hashes}});

    # No address (yet) if we learned about the peer by being connected to
    return $peer unless $host;

    die "Host $host contains a :" if $host =~ /:/;
    die "Port $port contains a :" if $port =~ /:/;
    my $address = "$host:$port";
    $peer->announce_address($address);
    return $peer;
}

# Called when a given address for a peer becomes invalid
sub delete_peer_address {
    (my __PACKAGE__ $downloader, my $peer_id, my $host, my $port) = @_;
    defined(my $peer = $downloader->{id2peer}{$peer_id}) ||
        die "Unknown peer $peer_id";
    my $address = "$host:$port";
    $peer->delete_address($address);
}

# Call when a peer has no addresses or connections anymore
sub delete_peer {
    my __PACKAGE__ $downloader = shift;
    delete $downloader->{id2peer}{shift()} || 
        die "Deleteing non-existent peer";
}

# Start trackers once we've gotten our port
sub start_trackers {
    my __PACKAGE__ $downloader = shift;
    $_->start(@_, 
              file_id => $downloader->{file_id},
              left => $downloader->{left},
              downloaded => $downloader->{downloaded},
              uploaded => $downloader->{uploaded}) for 
                  values %{$downloader->{trackers}};
}

# Do one of the periodic announcements to the tracker
sub update_tracker {
    my __PACKAGE__ $downloader = shift;
    my $tracker = shift;
    $tracker->status_update(@_,
                            file_id => $downloader->{file_id},
                            left => $downloader->{left},
                            downloaded => $downloader->{downloaded},
                            uploaded => $downloader->{uploaded});
}

# This downloader finished its job
sub complete_trackers {
    my __PACKAGE__ $downloader = shift;
    push @_, (file_id	=> $downloader->{file_id},
              left	=> $downloader->{left},
              downloaded => $downloader->{downloaded},
              uploaded	=> $downloader->{uploaded});
    $_->complete(@_) for values %{$downloader->{trackers}};
}

# A (possibly new) tracker became known for this downloader
sub announce_tracker {
    my __PACKAGE__ $downloader = shift;
    my $metainfo = shift;

    defined(my $tracker = $metainfo->{announce}) ||
        die "No announce field in torrent file";
    my $uri = URI->new($metainfo->{announce});
    $downloader->{trackers}{$uri->as_string} ||=
        POE::Component::BitTorrent::Tracker->new
        ($uri, @_,
         file_id => $downloader->{file_id},
         left => $downloader->{left},
         downloaded => $downloader->{downloaded},
         uploaded => $downloader->{uploaded});
}

# Given a connection and what that peer wants, start queueing requests
# peer download_connection must already be set when xalling this since
# it may cause interests to drop
sub start_download {
    (my __PACKAGE__ $downloader, my $new, my $connection, my $local_wanted)=@_;
    # First we need to pick what we want
    die "Connection interest if nothing wanted ?" if !defined($local_wanted);

    $downloader->{client}->start_download if $new;

    my $submitted = 0;
    my $max_submit = $MAX_REQUESTS - $connection->pending_requests;
    die "Connection has no space" if $max_submit <= 0;
    while ($submitted < $max_submit) {
        # First check if the peer has a piece which is already being requested
        my $running = $downloader->{partial_bitmap} & $local_wanted;
        # Otherwise any of the available pieces will do
        $running =unpack("b*", $running =~ /[^\x00]/ ? $running:$local_wanted);
        # Now choose a piece. Mix of random and rarest first
        # Choose $RANDOM_PIECES random pieces
        my $k = 0;
        my @pieces;
        while ($running =~ /1/g) {
            if (++$k <= $RANDOM_PIECES) {
                push @pieces, $-[0];
            } elsif (rand($k) < $RANDOM_PIECES) {
                $pieces[rand($RANDOM_PIECES)] = $-[0];
            }
        }
        # Notice that the piece numbers are vec() indices, not bit MSB offsets
        if (@pieces <= 1) {
            if (!@pieces) {
                die "No wanted pieces" if $submitted == 0;
                last;
            }
            $k = 0;
        } else {
            # Now check how rare the pieces are
            my @count = POE::Component::BitTorrent::Peer::have_count
                (\@pieces, values %{$downloader->{id2peer}});
            die "length got damaged" if @count != @pieces;

            my $min = 1e50;
            $k = undef;
            for (0..$#count) {
                if ($count[$_] < $min) {
                    $min = $count[$_];
                    $k = $_;
                }
            }
            die "No minimum ?" if !defined($k);
            die "Count zero" if $min <= 0;
        }
        # convert little to big endian bitcount
        my $piece_nr = $pieces[$k] ^ 7;
        # print STDERR "Running=", unpack("H*", pack("b*", $running)), ", set=@pieces,chose $piece_nr\n";
        my $piece = $downloader->{requested_pieces}{$piece_nr} ||=
            POE::Component::BitTorrent::Piece->new
            ($piece_nr, $piece_nr == $#{$downloader->{piece_hashes}} ?
             $downloader->{length}-$piece_nr*$downloader->{piece_length} :
             $downloader->{piece_length});
         while ($submitted < $max_submit) {
            my ($last, @range) = $piece->get_range;
            print(STDERR "Request piece $piece_nr, ",
                  "offset $range[0], length $range[1]\n");
            $connection->send_message($REQUEST, $piece_nr, @range);
            ++$submitted;
            if ($last) {
                vec($local_wanted, $piece_nr ^ 7, 1) = 0;
                vec($downloader->{partial_bitmap}, $piece_nr ^ 7, 1) = 0;
                $downloader->stop_wanting($piece_nr);
                last;
            }
            vec($downloader->{partial_bitmap}, $piece_nr ^ 7, 1) = 1;
        }
    }
}

sub stop_download {
    my __PACKAGE__ $downloader = shift;
    $downloader->{client}->stop_download;
}

# Stop wanting a given piece nr (requests for all pieces pending)
sub stop_wanting {
    (my __PACKAGE__ $downloader, my $piece_nr) = @_;
    print STDERR "Downloader $downloader->{file_id} stops wanting piece $piece_nr\n";
    vec($downloader->{want_bitmap}, $piece_nr ^ 7, 1) = 0;
    $_->stop_wanting($piece_nr) for values %{$downloader->{id2peer}};
}

# Start wanting a list of pieces in LSB format
# Currently starts a download on the first non-busy unchoked peer with a
# connection that has the piece. Needs a preference system !
sub start_wanting {
    my __PACKAGE__ $downloader = shift;
    die "No pieces wanted" unless my @pieces = @_;
    for my $piece_nr (@pieces) {
        print(STDERR "Downloader $downloader->{file_id} starts wanting ", 
              $piece_nr ^ 7, "\n");
        die "We already have piece number ", $piece_nr ^ 7 if
            vec($downloader->{have_bitmap}, $piece_nr, 1);
        die "We already wanted piece number ", $piece_nr ^ 7 if
            vec($downloader->{want_bitmap}, $piece_nr, 1);
        vec($downloader->{want_bitmap}, $piece_nr, 1) = 1;
    }
    $_->start_wanting(@pieces) and do {
        # Maybe we stopped wanting some pieces
        return unless 
            @pieces = grep vec($downloader->{want_bitmap}, $_, 1), @pieces;
    } for values %{$downloader->{id2peer}};
}

sub got_fragment {
    my __PACKAGE__ $downloader = shift;
    my $piece_nr = shift;
    my $offset   = shift;
    my $length   = shift;
    die "Piece $piece_nr is not in the request set" unless
        $downloader->{requested_pieces}{$piece_nr};
    my $last = $downloader->{requested_pieces}{$piece_nr}->got_range($offset);
    delete $downloader->{requested_pieces}{$piece_nr} if $last;
    $downloader->{downloaded} += $length;
    $downloader->{client}->start_write_fragment
        ($downloader->{file}, $last, $piece_nr, $offset, $length, shift);
}

sub drop_requests {
    (my __PACKAGE__ $downloader, my $requests) = @_;
    my (%pieces, @pieces);
    push @{$pieces{$_->[0]}}, $_->[1] for @$requests;
    while (my ($piece_nr, $offsets) = each %pieces) {
        die "Piece $piece_nr is not in the request set" unless
            $downloader->{requested_pieces}{$piece_nr};
        $downloader->{requested_pieces}{$piece_nr}->drop_range($_) for 
            @$offsets;
        # Convert to LSB format
        $piece_nr ^= 7;
        vec($downloader->{partial_bitmap}, $piece_nr, 1) = 1;
        push @pieces, $piece_nr if 
            !vec($downloader->{want_bitmap}, $piece_nr, 1);
    }
    $downloader->start_wanting(@pieces) if @pieces;
}

# A piece was hash checked and now reports the result
sub checked_piece {
    (my __PACKAGE__ $downloader, my $piece_nr, my $digest) = @_;
    die "piece number $piece_nr < 0" if $piece_nr < 0;
    die "piece number $piece_nr >= nr of pieces " . @{$downloader->{piece_hashes}} if
        $piece_nr >= @{$downloader->{piece_hashes}};
    die "piece number $piece_nr was wanted" if
        vec($downloader->{want_bitmap}, $piece_nr ^ 7, 1);
    if ($digest eq $downloader->{piece_hashes}[$piece_nr]) {
        print STDERR "Good digest for piece $piece_nr of $downloader->{file_id}\n";
        $downloader->announce_piece($piece_nr);
    } else {
        # We have a bad hash on this piece. Maybe start requesting it...
        print STDERR "Bad digest for piece $piece_nr of $downloader->{file_id}\n";
        $downloader->start_wanting($piece_nr ^ 7);
    }
}

# A piece was completed. Tell everybody
sub announce_piece {
    (my __PACKAGE__ $downloader, my $piece_nr) = @_;
    die "Announcing a piece we already had" if
        vec($downloader->{have_bitmap}, $piece_nr ^ 7, 1);
    vec($downloader->{have_bitmap}, $piece_nr ^ 7, 1) = 1;
    if (--$downloader->{nr_needed_pieces} <= 0) {
        die "Negative needed pieces" if $downloader->{nr_needed_pieces} < 0;
        print STDERR "File $downloader->{file_id} is now complete\n";
        $downloader->{client}->complete_trackers($downloader) if 
            $downloader->{downloaded};
        $downloader->all_callbacks("Completed");
    }
    if ($piece_nr == $#{$downloader->{piece_hashes}}) {
        # Last (possibly truncated) piece
        $downloader->{left} -= 
            $downloader->{length} - $downloader->{piece_length} * $piece_nr;
    } else {
        $downloader->{left} -= $downloader->{piece_length};
    }
    # Could be interesting if we had a list of actually open sockets here...
    $_->announce_piece($piece_nr, $downloader->{nr_needed_pieces}) for
        values %{$downloader->{id2peer}};
}

# Exactly one download slots became free
sub check_interested {
    my __PACKAGE__ $downloader = shift;
    return map $_->check_interested, values %{$downloader->{id2peer}};
}

# All download slots are filled. Drop interest
sub drop_interested {
    my __PACKAGE__ $downloader = shift;
    $_->drop_interested for values %{$downloader->{id2peer}};
}

# Yet another wants to a file
sub add_callback {
    my __PACKAGE__ $downloader = shift;
    push(@{$downloader->{callbacks}}, @_);
}

# Download finished. Tell everybody !
sub all_callbacks {
    my __PACKAGE__ $downloader = shift;
    my $callbacks = $downloader->{callbacks};
    $downloader->{callbacks} = [];
    $_->(@_) for @$callbacks;
}

# Return the file_id for the current downloader
sub file_id {
    my __PACKAGE__ $downloader = shift;
    return $downloader->{file_id};
}

# Given a peer_id, return the peer. Returns undef for unkown peers
sub id2peer {
    my __PACKAGE__ $downloader = shift;
    return $downloader->{shift()};
}

sub have_bitmap {
    my __PACKAGE__ $downloader = shift;
    return $downloader->{have_bitmap};
}

sub want_bitmap {
    my __PACKAGE__ $downloader = shift;
    return $downloader->{want_bitmap};
}

# Check if the downloader wants a given piece
sub want_piece {
    my __PACKAGE__ $downloader = shift;
    return vec($downloader->{want_bitmap}, 7 ^ shift, 1);
}

sub nr_needed_pieces {
    my __PACKAGE__ $downloader = shift;
    return $downloader->{nr_needed_pieces};
}

sub free_download_slots {
    my __PACKAGE__ $downloader = shift;
    return $downloader->{client}->free_download_slots;
}

# Debug. Return a string representing current state
sub string {
    my __PACKAGE__ $downloader = shift;
    my $str = "Downloader for $downloader->{file_id} ($downloader->{nr_needed_pieces} pieces still needed)\n";
    $str.= "Have bitmap:     ".unpack("H*",$downloader->{have_bitmap})	 ."\n";
    $str.= "Want bitmap:     ".unpack("H*",$downloader->{want_bitmap})	 ."\n";
    $str.= "Part bitmap:     ".unpack("H*",$downloader->{partial_bitmap})."\n";
    $str .= $_->string for values %{$downloader->{id2peer}};
    $str =~ s/\n^/\n    /mg;
    return $str;
}

# Debug. Print the current state
sub print : method {
    print STDERR shift->string;
}

package POE::Component::BitTorrent::Tracker;
use HTTP::Request::Common qw(GET);
use fields qw(uri status_alarm);
use POE;

sub new {
    my $class = shift;
    my __PACKAGE__ $tracker = bless [], $class;
    $tracker->{uri} = shift;
    $tracker->start(@_);
    return $tracker;
}

sub start {
    my __PACKAGE__ $tracker = shift;
    $tracker->status_update(@_, event => "started");
}

sub complete {
    my __PACKAGE__ $tracker = shift;
    die "No interval alarm pending" unless $tracker->{status_alarm};
    $poe_kernel->alarm_remove($tracker->{status_alarm});
    $tracker->status_update(@_, event => "completed");
}

sub status_update {
    (my __PACKAGE__ $tracker, my %parameters) = @_;
    return unless $parameters{port};
    $tracker->{uri}->query_form(info_hash	=> $parameters{file_id},
                             peer_id	=> $parameters{peer_id},
                             port	=> $parameters{port},
                             uploaded	=> $parameters{uploaded},
                             downloaded	=> $parameters{downloaded},
                             left	=> $parameters{left},
                             $parameters{event} ?
                             (event	=> $parameters{event}) : ());
    print STDERR gmtime() . ": Sending ", $tracker->{uri}->as_string, "\n";
    $poe_kernel->post($parameters{http_alias},
                      "request",		# posts to ua's 'request' state
                      "tracker_response",	# state getting response
                      GET($tracker->{uri}),
                      $tracker);
}

# A new timeout became known
sub interval {
    (my __PACKAGE__ $tracker, my $downloader, my $interval) = @_;
    if ($interval < $MIN_TRACKER_INTERVAL) {
        print STDERR "Tracker asks for insanely low interval $interval, replaced by $MIN_TRACKER_INTERVAL\n";
        $interval = $MIN_TRACKER_INTERVAL;
    }
    print STDERR "tracker delay_set $interval\n";
    $poe_kernel->alarm_remove($tracker->{status_alarm}) if 
        $tracker->{status_alarm};
    $tracker->{status_alarm} = $poe_kernel->delay_set
        ("update_tracker", $interval, $downloader, $tracker);
}

sub print : method {
    print STDERR shift->string;
}

1;
