garage_util 0.8.2

Utility crate for the Garage object store
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

use bytes::Bytes;
use digest::Digest;

use tokio::sync::mpsc;
use tokio::task::JoinHandle;

use crate::data::*;

/// Compute the sha256 of a slice,
/// spawning on a tokio thread for CPU-intensive processing
/// The argument has to be an owned Bytes, as it is moved out to a new thread.
pub async fn async_sha256sum(data: Bytes) -> Hash {
	tokio::task::spawn_blocking(move || sha256sum(&data))
		.await
		.unwrap()
}

/// Compute the blake2sum of a slice,
/// spawning on a tokio thread for CPU-intensive processing.
/// The argument has to be an owned Bytes, as it is moved out to a new thread.
pub async fn async_blake2sum(data: Bytes) -> Hash {
	tokio::task::spawn_blocking(move || blake2sum(&data))
		.await
		.unwrap()
}

// ----

pub struct AsyncHasher<D: Digest> {
	sendblk: mpsc::Sender<Bytes>,
	task: JoinHandle<digest::Output<D>>,
}

impl<D: Digest> AsyncHasher<D> {
	pub fn new() -> Self {
		let (sendblk, mut recvblk) = mpsc::channel::<Bytes>(1);
		let task = tokio::task::spawn_blocking(move || {
			let mut digest = D::new();
			while let Some(blk) = recvblk.blocking_recv() {
				digest.update(&blk[..]);
			}
			digest.finalize()
		});
		Self { sendblk, task }
	}

	pub async fn update(&self, b: Bytes) {
		self.sendblk.send(b).await.unwrap();
	}

	pub async fn finalize(self) -> digest::Output<D> {
		drop(self.sendblk);
		self.task.await.unwrap()
	}
}

impl<D: Digest> Default for AsyncHasher<D> {
	fn default() -> Self {
		Self::new()
	}
}