diffs_rs/
reading.rs

1//! Source-faithful reads, independent of display rows and terminal ownership.
2
3use std::{
4    collections::HashMap,
5    io::{BufReader, Write},
6    sync::{
7        Arc, Weak,
8        atomic::{AtomicBool, Ordering},
9    },
10};
11
12use anyhow::{Result, ensure};
13
14use crate::{
15    engine::{ReviewId, Side},
16    source::{CheckedPair, Snapshot},
17    storage::{FileIndex, copy_checked},
18};
19
20pub(crate) mod domain;
21pub(crate) mod literal;
22pub(crate) mod search;
23mod search_read;
24pub use literal::Direction as SearchDirection;
25pub(crate) use search_read::SearchStart;
26pub use search_read::{SearchMatch, SearchRead};
27
28/// An inclusive range of original lines on one side of one open review.
29#[derive(Clone, Copy, Debug, PartialEq, Eq)]
30pub struct SourceRange {
31    pub review: ReviewId,
32    pub file: usize,
33    pub side: Side,
34    pub first: u32,
35    pub last: u32,
36}
37
38pub(crate) struct LineSelection {
39    pub range: SourceRange,
40    pub anchor: u32,
41    pub extending: bool,
42    pub checked: bool,
43}
44
45enum Source {
46    Patch(Arc<FileIndex>),
47    Full {
48        snapshot: Arc<Snapshot>,
49        valid: Arc<AtomicBool>,
50    },
51}
52
53#[derive(Default)]
54pub(crate) struct SourceLeases {
55    files: HashMap<usize, Weak<AtomicBool>>,
56}
57
58impl SourceLeases {
59    pub(crate) fn lease(&mut self, file: usize) -> Arc<AtomicBool> {
60        self.files.retain(|_, lease| lease.strong_count() != 0);
61        if let Some(lease) = self.files.get(&file).and_then(Weak::upgrade) {
62            return lease;
63        }
64        let lease = Arc::new(AtomicBool::new(true));
65        self.files.insert(file, Arc::downgrade(&lease));
66        lease
67    }
68
69    pub(crate) fn invalidate(&mut self, file: usize) {
70        if let Some(lease) = self.files.remove(&file).and_then(|lease| lease.upgrade()) {
71            lease.store(false, Ordering::Release);
72        }
73    }
74}
75
76/// A source job. Drop it after use to release its source ownership.
77/// Closing the engine revokes reads; it does not drop a caller-owned job.
78pub struct SourceRead {
79    range: SourceRange,
80    source: Source,
81    active: Arc<AtomicBool>,
82}
83
84impl SourceRead {
85    pub(crate) fn new(
86        range: SourceRange,
87        file: Arc<FileIndex>,
88        pair: Option<(&Arc<CheckedPair>, Arc<AtomicBool>)>,
89        active: Arc<AtomicBool>,
90    ) -> Result<Self> {
91        ensure!(
92            range.first != 0 && range.first <= range.last,
93            "Invalid source line range"
94        );
95        ensure!(
96            file.id == range.file,
97            "Source range belongs to another file"
98        );
99        let source = if let Some((pair, valid)) = pair {
100            ensure!(
101                Arc::ptr_eq(pair.file(), &file),
102                "Full source belongs to another file occurrence"
103            );
104            let snapshot = match range.side {
105                Side::Old => pair.old(),
106                Side::New => pair.new_source(),
107            };
108            let snapshot =
109                snapshot.ok_or_else(|| anyhow::anyhow!("The requested side has no source"))?;
110            ensure!(
111                range.last <= snapshot.lines(),
112                "Source range is outside the full file"
113            );
114            Source::Full {
115                snapshot: snapshot.clone(),
116                valid,
117            }
118        } else {
119            Source::Patch(file)
120        };
121        Ok(Self {
122            range,
123            source,
124            active,
125        })
126    }
127
128    pub fn range(&self) -> SourceRange {
129        self.range
130    }
131
132    /// Checks source identity without reading bytes. Use before publishing prepared output.
133    pub fn check(&self) -> Result<()> {
134        ensure!(
135            self.active.load(Ordering::Acquire),
136            "Source read belongs to a closed review"
137        );
138        if let Source::Full { valid, .. } = &self.source {
139            ensure!(
140                valid.load(Ordering::Acquire),
141                "Checked source is no longer available"
142            );
143        }
144        Ok(())
145    }
146
147    /// Streams exact source bytes without flushing the caller's output.
148    /// Run off the input thread. Cancellation is checked between bounded copies.
149    /// The caller must unblock a sink if its `Write` implementation can wait indefinitely.
150    /// A failed sink can contain partial output; stage before visible publication.
151    pub fn write_to(
152        &self,
153        output: &mut impl Write,
154        cancel: &impl Fn() -> Result<()>,
155    ) -> Result<u64> {
156        let check = || {
157            self.check()?;
158            cancel()
159        };
160        check()?;
161        let written = match &self.source {
162            Source::Patch(file) => file.write_source_lines(
163                self.range.side == Side::Old,
164                self.range.first..=self.range.last,
165                output,
166                &check,
167            )?,
168            Source::Full { snapshot, .. } => {
169                let reader = snapshot.uncached_range(self.range.first - 1..self.range.last)?;
170                let len = reader.remaining();
171                copy_checked(
172                    &mut BufReader::with_capacity(64 * 1024, reader),
173                    output,
174                    len,
175                    &check,
176                )?;
177                len
178            }
179        };
180        check()?;
181        Ok(written)
182    }
183}
diffs_rs/reading.rs

diffs_rs/
reading.rs
