apt_cmd/
apt_cache.rs

1// Copyright 2021-2022 System76 <info@system76.com>
2// SPDX-License-Identifier: MPL-2.0
3
4use anyhow::Context;
5use as_result::{IntoResult, MapResult};
6use futures::stream::{Stream, StreamExt};
7use std::collections::HashMap;
8use std::io;
9use std::pin::Pin;
10use tokio::io::{AsyncBufReadExt, AsyncReadExt, BufReader};
11use tokio::process::{Child, ChildStdout, Command};
12use tokio_stream::wrappers::LinesStream;
13
14pub type PackageStream = Pin<Box<dyn Stream<Item = String>>>;
15
16#[derive(Debug, Clone)]
17pub struct Policy {
18    pub package: String,
19    pub installed: String,
20    pub candidate: String,
21    pub version_table: HashMap<String, Vec<String>>,
22}
23
24pub type Policies = Pin<Box<dyn Stream<Item = Policy>>>;
25
26pub fn policies(lines: impl Stream<Item = io::Result<String>>) -> impl Stream<Item = Policy> {
27    async_stream::stream! {
28        futures::pin_mut!(lines);
29
30        let mut policy = Policy {
31            package: String::new(),
32            installed: String::new(),
33            candidate: String::new(),
34            version_table: HashMap::new()
35        };
36
37        while let Some(Ok(line)) = lines.next().await {
38            if line.is_empty() {
39                continue
40            }
41
42            if !line.starts_with(' ') {
43                policy.package = line;
44                if policy.package.ends_with(':') {
45                    policy.package.truncate(policy.package.len() - 1);
46                }
47                continue
48            }
49
50            let line = line.trim();
51
52            if line.starts_with('I') {
53                if let Some(v) = line.split_ascii_whitespace().nth(1) {
54                    policy.installed = v.to_owned();
55                }
56            } else if line.starts_with('C') {
57                if let Some(v) = line.split_ascii_whitespace().nth(1) {
58                    policy.candidate = v.to_owned();
59                }
60            } else if line.starts_with('V') {
61                // Start parsing the version table
62                let mut current_version = String::from("unknown");
63                while let Some(Ok(line)) = lines.next().await {
64
65
66                    if let Some(source) = line.strip_prefix("      ") {
67                        policy.version_table.entry(current_version.clone())
68                            .or_insert_with(Vec::new)
69                            .push(source.trim().to_owned());
70                    } else if let Some(version) = line.strip_prefix(" *** ") {
71                        current_version = version.trim().to_owned();
72                    } else if let Some(version) = line.strip_prefix("   ") {
73                        current_version = version.trim().to_owned();
74                    } else {
75                        yield policy.clone();
76                        policy.version_table.clear();
77                        policy.package = line;
78                        if policy.package.ends_with(':') {
79                            policy.package.truncate(policy.package.len() - 1);
80                        }
81                        break
82                    }
83                }
84            }
85        }
86
87        yield policy;
88    }
89}
90
91#[derive(AsMut, Deref, DerefMut)]
92#[as_mut(forward)]
93pub struct AptCache(Command);
94
95impl AptCache {
96    #[allow(clippy::new_without_default)]
97    pub fn new() -> Self {
98        let mut cmd = Command::new("apt-cache");
99        cmd.env("LANG", "C");
100        Self(cmd)
101    }
102
103    pub async fn depends<I, S>(mut self, packages: I) -> io::Result<(Child, ChildStdout)>
104    where
105        I: IntoIterator<Item = S>,
106        S: AsRef<std::ffi::OsStr>,
107    {
108        self.arg("depends");
109        self.args(packages);
110        self.spawn_with_stdout().await
111    }
112
113    pub async fn rdepends<I, S>(mut self, packages: I) -> io::Result<(Child, PackageStream)>
114    where
115        I: IntoIterator<Item = S>,
116        S: AsRef<std::ffi::OsStr>,
117    {
118        self.arg("rdepends");
119        self.args(packages);
120        self.stream_packages().await
121    }
122
123    pub async fn policy<S: AsRef<std::ffi::OsStr>>(
124        mut self,
125        packages: &[S],
126    ) -> anyhow::Result<(Child, Policies)> {
127        self.arg("policy");
128        self.args(packages);
129        self.env("LANG", "C");
130
131        let (child, stdout) = self.spawn_with_stdout().await?;
132
133        let lines = LinesStream::new(BufReader::new(stdout).lines());
134
135        let stream = Box::pin(policies(lines));
136
137        Ok((child, stream))
138    }
139
140    pub async fn predepends_of<'a>(
141        out: &'a mut String,
142        package: &'a str,
143    ) -> anyhow::Result<Vec<&'a str>> {
144        let (mut child, mut packages) = AptCache::new()
145            .rdepends(&[&package])
146            .await
147            .with_context(|| format!("failed to launch `apt-cache rdepends {}`", package))?;
148
149        let mut depends = Vec::new();
150        while let Some(package) = packages.next().await {
151            depends.push(package);
152        }
153
154        child
155            .wait()
156            .await
157            .map_result()
158            .with_context(|| format!("bad status from `apt-cache rdepends {}`", package))?;
159
160        let (mut child, mut stdout) = AptCache::new()
161            .depends(&depends)
162            .await
163            .with_context(|| format!("failed to launch `apt-cache depends {}`", package))?;
164
165        stdout
166            .read_to_string(out)
167            .await
168            .with_context(|| format!("failed to get output of `apt-cache depends {}`", package))?;
169
170        child.wait().await.map_result()?;
171
172        Ok(PreDependsIter::new(out.as_str(), package)?.collect::<Vec<_>>())
173    }
174
175    async fn stream_packages(self) -> io::Result<(Child, PackageStream)> {
176        let (child, stdout) = self.spawn_with_stdout().await?;
177
178        let mut lines = LinesStream::new(BufReader::new(stdout).lines()).skip(2);
179
180        let stream = async_stream::stream! {
181            while let Some(Ok(package)) = lines.next().await {
182                yield package.trim_start().to_owned();
183            }
184        };
185
186        Ok((child, Box::pin(stream)))
187    }
188
189    pub async fn status(mut self) -> io::Result<()> {
190        self.0.status().await?.into_result()
191    }
192
193    pub async fn spawn_with_stdout(self) -> io::Result<(Child, ChildStdout)> {
194        crate::utils::spawn_with_stdout(self.0).await
195    }
196}
197pub struct PreDependsIter<'a> {
198    lines: std::str::Lines<'a>,
199    predepend: &'a str,
200    active: &'a str,
201}
202
203impl<'a> PreDependsIter<'a> {
204    pub fn new(output: &'a str, predepend: &'a str) -> io::Result<Self> {
205        let mut lines = output.lines();
206
207        let active = lines.next().ok_or_else(|| {
208            io::Error::new(
209                io::ErrorKind::NotFound,
210                "expected the first line of the output of apt-cache depends to be a package name",
211            )
212        })?;
213
214        Ok(Self {
215            lines,
216            predepend,
217            active: active.trim(),
218        })
219    }
220}
221
222impl<'a> Iterator for PreDependsIter<'a> {
223    type Item = &'a str;
224
225    fn next(&mut self) -> Option<Self::Item> {
226        let mut found = false;
227        for line in &mut self.lines {
228            if !line.starts_with(' ') {
229                let prev = self.active;
230                self.active = line.trim();
231                if found {
232                    return Some(prev);
233                }
234            } else if !found && line.starts_with("  PreDepends: ") && &line[14..] == self.predepend
235            {
236                found = true;
237            }
238        }
239
240        None
241    }
242}
apt_cmd/apt_cache.rs

apt_cmd/
apt_cache.rs
