leyoda/pop-launcher
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

improv(service): Use postage with futures traits

Browse source
This commit is contained in:
Michael Aaron Murphy 2021-08-25 19:19:39 +02:00
parent 9c491a4f9f
commit 4abe6a894a
4 changed files with 6 additions and 9 deletions
Download patch file Download diff file Expand all files Collapse all files

1
Cargo.lock generated
View file

@ -1312,6 +1312,7 @@ checksum = "a63d25391d04a097954b76aba742b6b5b74f213dfe3dbaeeb36e8ddc1c657f0b"
dependencies = [ dependencies = [
"atomic", "atomic",
"crossbeam-queue", "crossbeam-queue",
"futures",
"log", "log",
"pin-project 1.0.8", "pin-project 1.0.8",
"pollster", "pollster",

2
plugins/Cargo.toml
View file

@ -16,7 +16,7 @@ human_format = "1.0"
human-sort = "0.2" human-sort = "0.2"
new_mime_guess = "3" new_mime_guess = "3"
pop-launcher = { path = "../" } pop-launcher = { path = "../" }
postage = "0.4" postage = { version = "0.4", features = ["futures-traits"] }
regex = "1" regex = "1"
ron = "0.6" ron = "0.6"
serde = "1" serde = "1"

2
service/Cargo.toml
View file

@ -14,7 +14,7 @@ futures-lite = "1"
gen-z = "0.1" gen-z = "0.1"
num_cpus = "1" num_cpus = "1"
pop-launcher = { path = "../" } pop-launcher = { path = "../" }
postage = "0.4" postage = { version = "0.4", features = ["futures-traits"] }
regex = "1.5" regex = "1.5"
ron = "0.6" ron = "0.6"
serde = { version = "1", features = ["derive"] } serde = { version = "1", features = ["derive"] }

10
service/src/lib.rs
View file

@ -37,19 +37,17 @@ pub async fn main() {
} }
}); });
let (output_tx, mut output_rx) = postage::mpsc::channel(16); let (output_tx, mut output_rx) = mpsc::channel(16);
// Service will operate for as long as it is being awaited // Service will operate for as long as it is being awaited
let service = Service::new(output_tx).exec(input_stream); let service = Service::new(output_tx).exec(input_stream);
// Responses from the service will be streamed to stdout // Responses from the service will be streamed to stdout
let responder = async move { let responder = async move {
use postage::prelude::Stream;
let stdout = io::stdout(); let stdout = io::stdout();
let stdout = &mut stdout.lock(); let stdout = &mut stdout.lock();
while let Some(response) = output_rx.recv().await { while let Some(response) = output_rx.next().await {
serialize_out(stdout, &response); serialize_out(stdout, &response);
} }
}; };
@ -84,7 +82,6 @@ impl Service {
pub async fn exec(mut self, input: impl Stream<Item = Request>) { pub async fn exec(mut self, input: impl Stream<Item = Request>) {
let (service_tx, service_rx) = mpsc::channel(1); let (service_tx, service_rx) = mpsc::channel(1);
let stream = plugins::external::load::from_paths(); let stream = plugins::external::load::from_paths();
futures_lite::pin!(stream); futures_lite::pin!(stream);
@ -121,8 +118,7 @@ impl Service {
} }
async fn response_handler(&mut self, mut service_rx: mpsc::Receiver<Event>) { async fn response_handler(&mut self, mut service_rx: mpsc::Receiver<Event>) {
use postage::prelude::Stream; while let Some(event) = service_rx.next().await {
while let Some(event) = service_rx.recv().await {
match event { match event {
Event::Request(request) => { Event::Request(request) => {
match request { match request {