uniprot 0.7.0

Rust data structures and parser for the Uniprot database(s).
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
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133

use std::collections::HashSet;
use std::io::BufRead;
use std::num::NonZeroUsize;
use std::str::FromStr;
use std::sync::atomic::AtomicBool;
use std::sync::atomic::Ordering;
use std::sync::Arc;
use std::sync::Mutex;
use std::time::Duration;

use crossbeam_channel::Sender;
use quick_xml::Error as XmlError;

use crate::error::Error;

#[cfg(feature = "threading")]
#[derive(Debug, PartialEq, Eq)]
/// The state of the `Producer`.
enum State {
    Started,
    Reading,
    Finished,
}

#[cfg(feature = "threading")]
pub struct Producer<B> {
    reader: Option<B>,
    threads: usize,
    s_text: Sender<Option<Result<Vec<u8>, Error>>>,
    alive: Arc<AtomicBool>,
    handle: Option<std::thread::JoinHandle<()>>,
}

impl<B: BufRead + Send + 'static> Producer<B> {
    pub(super) fn new(
        reader: B,
        threads: usize,
        s_text: Sender<Option<Result<Vec<u8>, Error>>>,
    ) -> Self {
        Self {
            reader: Some(reader),
            s_text,
            threads,
            handle: None,
            alive: Arc::new(AtomicBool::new(false)),
        }
    }

    pub fn start(&mut self) {
        self.alive.store(true, Ordering::SeqCst);

        let alive = self.alive.clone();
        let threads = self.threads;
        let s_text = self.s_text.clone();
        let mut reader = self.reader.take().unwrap();

        self.handle = Some(std::thread::spawn(move || {
            let mut buffer = Vec::new();
            let mut state = State::Started;
            loop {
                match state {
                    State::Started => match reader.read_until(b'>', &mut buffer) {
                        // we reached EOF, but that's okay, we were not
                        // reading an entry;
                        Ok(0) => {
                            state = State::Finished;
                        }
                        // we found the beginning of an entry, now we
                        // must read the entire entry until the end.
                        Ok(_) => {
                            let i = memchr::memrchr(b'<', &buffer).unwrap();
                            if buffer[i..].starts_with(b"<entry") {
                                state = State::Reading;
                            }
                        }
                        // if an error is encountered, send it and bail out
                        Err(e) => {
                            s_text.send(Some(Err(Error::from(e)))).ok();
                            state = State::Finished;
                        }
                    },
                    State::Reading => {
                        // read until the end of the entry.
                        match reader.read_until(b'>', &mut buffer) {
                            // if a full entry is found, send it
                            Ok(_) if buffer.ends_with(&b"</entry>"[..]) => {
                                s_text.send(Some(Ok(buffer.as_slice().to_vec()))).ok();
                                state = State::Started;
                                buffer.clear();
                            }
                            // if we reach EOF before finding the end of the
                            // entry, that's an issue, we report an error.
                            Ok(0) => {
                                s_text
                                    .send(Some(Err(Error::from(XmlError::UnexpectedEof(
                                        String::from("entry"),
                                    )))))
                                    .ok();
                                state = State::Finished;
                            }
                            // if an error is encountered, send it and bail out
                            Err(e) => {
                                s_text.send(Some(Err(Error::from(e)))).ok();
                                state = State::Finished;
                            }
                            // otherwise just keep iterating.
                            _ => (),
                        }
                    }
                    State::Finished => {
                        for _ in 0..threads {
                            s_text.send(None).ok();
                        }
                        alive.store(false, Ordering::SeqCst);
                        break;
                    }
                }
            }
        }));
    }

    pub fn join(&mut self) -> std::thread::Result<()> {
        if let Some(handle) = self.handle.take() {
            handle.join()
        } else {
            Ok(())
        }
    }

    pub fn is_alive(&self) -> bool {
        self.alive.load(Ordering::SeqCst)
    }
}