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
|
//! In `--recursive` mode we set the `lintcheck` binary as the `RUSTC_WRAPPER` of `cargo check`,
//! this allows [`crate::driver`] to be run for every dependency. The driver connects to
//! [`LintcheckServer`] to ask if it should be skipped, and if not sends the stderr of running
//! clippy on the crate to the server
use crate::ClippyWarning;
use crate::RecursiveOptions;
use std::collections::HashSet;
use std::io::{BufRead, BufReader, Read, Write};
use std::net::{SocketAddr, TcpListener, TcpStream};
use std::sync::{Arc, Mutex};
use std::thread;
use cargo_metadata::diagnostic::Diagnostic;
use crossbeam_channel::{Receiver, Sender};
use serde::de::DeserializeOwned;
use serde::{Deserialize, Serialize};
#[derive(Debug, Eq, Hash, PartialEq, Clone, Serialize, Deserialize)]
pub(crate) struct DriverInfo {
pub package_name: String,
pub crate_name: String,
pub version: String,
}
pub(crate) fn serialize_line<T, W>(value: &T, writer: &mut W)
where
T: Serialize,
W: Write,
{
let mut buf = serde_json::to_vec(&value).expect("failed to serialize");
buf.push(b'\n');
writer.write_all(&buf).expect("write_all failed");
}
pub(crate) fn deserialize_line<T, R>(reader: &mut R) -> T
where
T: DeserializeOwned,
R: BufRead,
{
let mut string = String::new();
reader.read_line(&mut string).expect("read_line failed");
serde_json::from_str(&string).expect("failed to deserialize")
}
fn process_stream(
stream: TcpStream,
sender: &Sender<ClippyWarning>,
options: &RecursiveOptions,
seen: &Mutex<HashSet<DriverInfo>>,
) {
let mut stream = BufReader::new(stream);
let driver_info: DriverInfo = deserialize_line(&mut stream);
let unseen = seen.lock().unwrap().insert(driver_info.clone());
let ignored = options.ignore.contains(&driver_info.package_name);
let should_run = unseen && !ignored;
serialize_line(&should_run, stream.get_mut());
let mut stderr = String::new();
stream.read_to_string(&mut stderr).unwrap();
let messages = stderr
.lines()
.filter_map(|json_msg| serde_json::from_str::<Diagnostic>(json_msg).ok())
.filter_map(|diag| ClippyWarning::new(diag, &driver_info.package_name, &driver_info.version));
for message in messages {
sender.send(message).unwrap();
}
}
pub(crate) struct LintcheckServer {
pub local_addr: SocketAddr,
receiver: Receiver<ClippyWarning>,
sender: Arc<Sender<ClippyWarning>>,
}
impl LintcheckServer {
pub fn spawn(options: RecursiveOptions) -> Self {
let listener = TcpListener::bind("localhost:0").unwrap();
let local_addr = listener.local_addr().unwrap();
let (sender, receiver) = crossbeam_channel::unbounded::<ClippyWarning>();
let sender = Arc::new(sender);
// The spawned threads hold a `Weak<Sender>` so that they don't keep the channel connected
// indefinitely
let sender_weak = Arc::downgrade(&sender);
// Ignore dependencies multiple times, e.g. for when it's both checked and compiled for a
// build dependency
let seen = Mutex::default();
thread::spawn(move || {
thread::scope(|s| {
s.spawn(|| {
while let Ok((stream, _)) = listener.accept() {
let sender = sender_weak.upgrade().expect("received connection after server closed");
let options = &options;
let seen = &seen;
s.spawn(move || process_stream(stream, &sender, options, seen));
}
});
});
});
Self {
local_addr,
receiver,
sender,
}
}
pub fn warnings(self) -> impl Iterator<Item = ClippyWarning> {
// causes the channel to become disconnected so that the receiver iterator ends
drop(self.sender);
self.receiver.into_iter()
}
}
|
