bip_disk 0.6.0

Bittorrent Infrastructure Project Disk Module
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90

use std::sync::{Arc, RwLock, Mutex};
use std::collections::HashMap;

use disk::ODiskMessage;
use disk::tasks::helpers::piece_checker::PieceCheckerState;

use bip_metainfo::Metainfo;
use bip_util::bt::InfoHash;
use futures::sync::mpsc::Sender;
use futures::sink::Sink;
use futures::sink::Wait;

pub struct DiskManagerContext<F> {
    torrents:    Arc<RwLock<HashMap<InfoHash, Mutex<MetainfoState>>>>,
    out:         Sender<ODiskMessage>,
    fs:          Arc<F>
}

pub struct MetainfoState {
    file:  Metainfo,
    state: PieceCheckerState
}

impl MetainfoState {
    pub fn new(file: Metainfo, state: PieceCheckerState) -> MetainfoState {
        MetainfoState{ file: file, state: state }
    }
}

impl<F> DiskManagerContext<F> {
    pub fn new(out: Sender<ODiskMessage>, fs: F) -> DiskManagerContext<F> {
        DiskManagerContext{ torrents: Arc::new(RwLock::new(HashMap::new())), out: out, fs: Arc::new(fs) }
    }

    pub fn blocking_sender(&self) -> Wait<Sender<ODiskMessage>> {
        self.out.clone().wait()
    }

    pub fn filesystem(&self) -> &F {
        &self.fs
    }

    pub fn insert_torrent(&self, file: Metainfo, state: PieceCheckerState) -> bool {
        let mut write_torrents = self.torrents.write()
            .expect("bip_disk: DiskManagerContext::insert_torrents Failed To Write Torrent");

        let hash = file.info().info_hash();
        let hash_not_exists = !write_torrents.contains_key(&hash);

        if hash_not_exists {
            write_torrents.insert(hash, Mutex::new(MetainfoState::new(file, state)));
        }

        hash_not_exists
    }

    pub fn update_torrent<C>(&self, hash: InfoHash, call: C) -> bool
        where C: FnOnce(&Metainfo, &mut PieceCheckerState) {
        let read_torrents = self.torrents.read()
            .expect("bip_disk: DiskManagerContext::update_torrent Failed To Read Torrent");

        match read_torrents.get(&hash) {
            Some(state) => {
                let mut lock_state = state.lock()
                    .expect("bip_disk: DiskManagerContext::update_torrent Failed To Lock State");
                let deref_state = &mut *lock_state;

                call(&deref_state.file, &mut deref_state.state);

                true
            },
            None => false
        }
    }

    pub fn remove_torrent(&self, hash: InfoHash) -> bool {
        let mut write_torrents = self.torrents.write()
            .expect("bip_disk: DiskManagerContext::remove_torrent Failed To Write Torrent");

        write_torrents.remove(&hash)
            .map(|_| true)
            .unwrap_or(false)
    }
}

impl<F> Clone for DiskManagerContext<F> {
    fn clone(&self) -> DiskManagerContext<F> {
        DiskManagerContext{ torrents: self.torrents.clone(), out: self.out.clone(), fs: self.fs.clone() }
    }
}