<commit_before>use futures::future::{Future,FutureResult};
use futures::future;
use futures::{Stream,Sink};
use std::collections::vec_deque::{VecDeque};
use std::default::Default;
use std::net::{SocketAddr};
use std::path::Path;
use std::time::Duration;
use tokio_core::net::{TcpStream};
use tokio_core::reactor;
use tokio_io::codec::{Framed};
use tokio_io::{AsyncRead};

use datastore::DataStore;
use error::{Error,Result};
use fillable::*;
use manifest::{ManifestWithFile};
use metainfo::{MetaInfo,InfoHash};
use peer_protocol::{PeerID,Message,BitTorrentPeerCodec};
use peer_protocol;
use tracker::{TrackerClient};
use util::{tcp_connect2,BxFuture,FutureEnhanced,VecDequeStream};

type PeerStream = Framed<TcpStream, BitTorrentPeerCodec>;

struct DownloaderState {
    info: MetaInfo,
    datastore: DataStore,
    manifest: ManifestWithFile,
    peer_state: PeerState,
    blah_state: BlahState,
}

type LoopState = (PeerStream, DownloaderState);

pub fn start<P: AsRef<Path>>(
    info: MetaInfo,
    peer_id: PeerID,
    store_path: P,
    manifest_path: P)
    -> Result<()>
{
    let mut core = reactor::Core::new()?;
    let handle = core.handle();

    let datastore = DataStore::create_or_open(&info, store_path)?;
    let manifest = ManifestWithFile::load_or_new(info.clone(), manifest_path)?;
    let mut tc = TrackerClient::new(info.clone(), peer_id.clone())?;

    let tracker_res = tc.easy_start()?;
    println!("tracker res: {:#?}", tracker_res);
    if let Some(reason) = tracker_res.failure_reason {
        return Err(Error::new_str(&format!("tracker failed: {}", reason)));
    }

    if tracker_res.peers.is_empty() {
        return Err(Error::new_str("tracker returned no peers"));
    }

    let peer = tracker_res.peers[0].clone();

    let num_pieces = info.num_pieces();
    let info_hash = info.info_hash.clone();

    let dstate = DownloaderState {
        info: info,
        datastore: datastore,
        manifest: manifest,
        peer_state: PeerState::new(num_pieces),
        blah_state: BlahState::default(),
    };

    let future_root = connect_peer(peer.address, info_hash, peer_id, &handle)
        .and_then(|(stream, remote_peer_id)| {
            let dstate = dstate;

            main_loop(stream, dstate)

                // .map_err(|err, _s| err)
                // .map(|(_, _s)| Ok(()))

            // Below lies ye ol' blocking implementation
            // for a single stream.

            // let mut state = PeerState::new(info.num_pieces());
            // let mut s = BlahState::default();

            // loop {
            //     let m = peer_protocol::read_message(&mut stream)?;
            //     println!("message: {}", m.summarize());
            //     match m {
            //         Message::Choke =>         state.peer_choking = true,
            //         Message::Unchoke =>       state.peer_choking = false,
            //         Message::Interested =>    state.peer_interested = true,
            //         Message::NotInterested => state.peer_interested = false,
            //         Message::Bitfield { bits } => {
            //             if bits.len() < info.num_pieces() {
            //                 println!("{}", Error::new_str(&format!("bitfield has less bits {} than pieces {}",
            //                                                 bits.len(), info.num_pieces())));
            //             }
            //             let mut i_start = 0;
            //             let mut in_interval = false;
            //             for b in 0..info.num_pieces() {
            //                 if bits[b] && !in_interval {
            //                     i_start = b;
            //                     in_interval = true;
            //                 } else if !bits[b] && in_interval {
            //                     state.has.add(i_start as u32, b as u32);
            //                     in_interval = false;
            //                 }
            //             }
            //             if in_interval {
            //                 state.has.add(i_start as u32, info.num_pieces() as u32);
            //             }
            //         },
            //         Message::Have { piece } => {
            //             state.has.add(piece, piece+1);
            //         },
            //         Message::Piece { piece, offset, block } => {
            //             datastore.write_block(piece as usize, offset, &block)?;
            //             manifest.manifest.add_block(piece as usize, offset, block.len() as u32)?;
            //             manifest.store()?;
            //             s.waiting = false;
            //         },
            //         _ => {},
            //     }
            //     s.nreceived += 1;
            //     if s.nreceived >= 1 && state.am_choking {
            //         let out = Message::Unchoke{};
            //         println!("sending message: {:?}", out);
            //         peer_protocol::send_message(&mut stream, &out)?;
            //         state.am_choking = false;
            //     }
            //     if s.nreceived >= 1 && !state.am_interested {
            //         let out = Message::Interested{};
            //         println!("sending message: {:?}", out);
            //         peer_protocol::send_message(&mut stream, &out)?;
            //         state.am_interested = true;
            //     }
            //     if !s.waiting && !state.peer_choking && state.am_interested {
            //         match manifest.manifest.next_desired_block() {
            //             None => {
            //                 return Err(Error::todo())
            //             },
            //             Some(desire) => {
            //                 let out = Message::Request{
            //                     piece: desire.piece,
            //                     offset: desire.offset,
            //                     length: desire.length,
            //                 };
            //                 println!("sending message: {:?}", out);
            //                 peer_protocol::send_message(&mut stream, &out)?;
            //                 s.waiting = true;
            //             }
            //         }
            //     }
            //     println!("state: {:?}", state);
            // }
        });
    core.run(future_root)?;
    Ok(())
}

/// Connect to a remote peer
fn connect_peer(<commit_msg>delete sync code<commit_after>use futures::future::{Future,FutureResult};
use futures::future;
use futures::{Stream,Sink};
use std::collections::vec_deque::{VecDeque};
use std::default::Default;
use std::net::{SocketAddr};
use std::path::Path;
use std::time::Duration;
use tokio_core::net::{TcpStream};
use tokio_core::reactor;
use tokio_io::codec::{Framed};
use tokio_io::{AsyncRead};

use datastore::DataStore;
use error::{Error,Result};
use fillable::*;
use manifest::{ManifestWithFile};
use metainfo::{MetaInfo,InfoHash};
use peer_protocol::{PeerID,Message,BitTorrentPeerCodec};
use peer_protocol;
use tracker::{TrackerClient};
use util::{tcp_connect2,BxFuture,FutureEnhanced,VecDequeStream};

type PeerStream = Framed<TcpStream, BitTorrentPeerCodec>;

struct DownloaderState {
    info: MetaInfo,
    datastore: DataStore,
    manifest: ManifestWithFile,
    peer_state: PeerState,
    blah_state: BlahState,
}

type LoopState = (PeerStream, DownloaderState);

pub fn start<P: AsRef<Path>>(
    info: MetaInfo,
    peer_id: PeerID,
    store_path: P,
    manifest_path: P)
    -> Result<()>
{
    let mut core = reactor::Core::new()?;
    let handle = core.handle();

    let datastore = DataStore::create_or_open(&info, store_path)?;
    let manifest = ManifestWithFile::load_or_new(info.clone(), manifest_path)?;
    let mut tc = TrackerClient::new(info.clone(), peer_id.clone())?;

    let tracker_res = tc.easy_start()?;
    println!("tracker res: {:#?}", tracker_res);
    if let Some(reason) = tracker_res.failure_reason {
        return Err(Error::new_str(&format!("tracker failed: {}", reason)));
    }

    if tracker_res.peers.is_empty() {
        return Err(Error::new_str("tracker returned no peers"));
    }

    let peer = tracker_res.peers[0].clone();

    let num_pieces = info.num_pieces();
    let info_hash = info.info_hash.clone();

    let dstate = DownloaderState {
        info: info,
        datastore: datastore,
        manifest: manifest,
        peer_state: PeerState::new(num_pieces),
        blah_state: BlahState::default(),
    };

    let future_root = connect_peer(peer.address, info_hash, peer_id, &handle)
        .and_then(|(stream, remote_peer_id)| {
            let dstate = dstate;

            main_loop(stream, dstate)
        });
    core.run(future_root)?;
    Ok(())
}

/// Connect to a remote peer
fn connect_peer(
{'fraction_non_alphanumeric': 0.14219299240276675, 'fraction_numerical': 0.002267830819820841, 'mean_word_length': 1.793791574279379, 'pattern_counts': {'":': 0, '<': 12, '<?xml version=': 0, '>': 25, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 18, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2801617', 'n_tokens_mistral': 2606, 'n_tokens_neox': 2565, 'n_words': 503}