add problem 09 (WIP)
This commit is contained in:
timeshifter
Generated
+344
@@ -0,0 +1,344 @@
|
|||||||
|
# This file is automatically @generated by Cargo.
|
||||||
|
# It is not intended for manual editing.
|
||||||
|
version = 4
|
||||||
|
|
||||||
|
[[package]]
|
||||||
|
name = "anyhow"
|
||||||
|
version = "1.0.103"
|
||||||
|
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||||
|
checksum = "2a4385e2e34eb35d6b3efe798b9eb88096925d87726c0798709bf56d9ed84af3"
|
||||||
|
|
||||||
|
[[package]]
|
||||||
|
name = "bitflags"
|
||||||
|
version = "2.13.1"
|
||||||
|
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||||
|
checksum = "b588b76d00fde79687d7646a9b5bdf3cc0f655e0bbd080335a95d7e96f3587da"
|
||||||
|
|
||||||
|
[[package]]
|
||||||
|
name = "bytes"
|
||||||
|
version = "1.12.1"
|
||||||
|
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||||
|
checksum = "fc652a48c352aef3ea3aed32080501cf3ef6ed5da78602a020c991775b0aff04"
|
||||||
|
|
||||||
|
[[package]]
|
||||||
|
name = "cfg-if"
|
||||||
|
version = "1.0.4"
|
||||||
|
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||||
|
checksum = "9330f8b2ff13f34540b44e946ef35111825727b38d33286ef986142615121801"
|
||||||
|
|
||||||
|
[[package]]
|
||||||
|
name = "chacha20"
|
||||||
|
version = "0.10.1"
|
||||||
|
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||||
|
checksum = "d524456ba66e72eb8b115ff89e01e497f8e6d11d78b70b1aa13c0fbd97540a81"
|
||||||
|
dependencies = [
|
||||||
|
"cfg-if",
|
||||||
|
"cpufeatures",
|
||||||
|
"rand_core",
|
||||||
|
]
|
||||||
|
|
||||||
|
[[package]]
|
||||||
|
name = "cpufeatures"
|
||||||
|
version = "0.3.0"
|
||||||
|
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||||
|
checksum = "8b2a41393f66f16b0823bb79094d54ac5fbd34ab292ddafb9a0456ac9f87d201"
|
||||||
|
dependencies = [
|
||||||
|
"libc",
|
||||||
|
]
|
||||||
|
|
||||||
|
[[package]]
|
||||||
|
name = "errno"
|
||||||
|
version = "0.3.14"
|
||||||
|
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||||
|
checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb"
|
||||||
|
dependencies = [
|
||||||
|
"libc",
|
||||||
|
"windows-sys",
|
||||||
|
]
|
||||||
|
|
||||||
|
[[package]]
|
||||||
|
name = "getrandom"
|
||||||
|
version = "0.4.3"
|
||||||
|
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||||
|
checksum = "300e883d756b2e4ec94e02791f39b04b522276138852cfc41d9fb7e904106099"
|
||||||
|
dependencies = [
|
||||||
|
"cfg-if",
|
||||||
|
"libc",
|
||||||
|
"r-efi",
|
||||||
|
"rand_core",
|
||||||
|
]
|
||||||
|
|
||||||
|
[[package]]
|
||||||
|
name = "itoa"
|
||||||
|
version = "1.0.18"
|
||||||
|
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||||
|
checksum = "8f42a60cbdf9a97f5d2305f08a87dc4e09308d1276d28c869c684d7777685682"
|
||||||
|
|
||||||
|
[[package]]
|
||||||
|
name = "libc"
|
||||||
|
version = "0.2.186"
|
||||||
|
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||||
|
checksum = "68ab91017fe16c622486840e4c83c9a37afeff978bd239b5293d61ece587de66"
|
||||||
|
|
||||||
|
[[package]]
|
||||||
|
name = "lock_api"
|
||||||
|
version = "0.4.14"
|
||||||
|
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||||
|
checksum = "224399e74b87b5f3557511d98dff8b14089b3dadafcab6bb93eab67d3aace965"
|
||||||
|
dependencies = [
|
||||||
|
"scopeguard",
|
||||||
|
]
|
||||||
|
|
||||||
|
[[package]]
|
||||||
|
name = "memchr"
|
||||||
|
version = "2.8.3"
|
||||||
|
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||||
|
checksum = "cf8baf1c55e62ffcace7a9f06f4bd9cd3f0c4beb022d3b367256b91b87513d98"
|
||||||
|
|
||||||
|
[[package]]
|
||||||
|
name = "mio"
|
||||||
|
version = "1.2.2"
|
||||||
|
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||||
|
checksum = "30d65c71f1ce40ab09135ce117d742b9f8a19ff91a41a8b57ed50bc2de59c427"
|
||||||
|
dependencies = [
|
||||||
|
"libc",
|
||||||
|
"wasi",
|
||||||
|
"windows-sys",
|
||||||
|
]
|
||||||
|
|
||||||
|
[[package]]
|
||||||
|
name = "parking_lot"
|
||||||
|
version = "0.12.5"
|
||||||
|
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||||
|
checksum = "93857453250e3077bd71ff98b6a65ea6621a19bb0f559a85248955ac12c45a1a"
|
||||||
|
dependencies = [
|
||||||
|
"lock_api",
|
||||||
|
"parking_lot_core",
|
||||||
|
]
|
||||||
|
|
||||||
|
[[package]]
|
||||||
|
name = "parking_lot_core"
|
||||||
|
version = "0.9.12"
|
||||||
|
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||||
|
checksum = "2621685985a2ebf1c516881c026032ac7deafcda1a2c9b7850dc81e3dfcb64c1"
|
||||||
|
dependencies = [
|
||||||
|
"cfg-if",
|
||||||
|
"libc",
|
||||||
|
"redox_syscall",
|
||||||
|
"smallvec",
|
||||||
|
"windows-link",
|
||||||
|
]
|
||||||
|
|
||||||
|
[[package]]
|
||||||
|
name = "pin-project-lite"
|
||||||
|
version = "0.2.17"
|
||||||
|
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||||
|
checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd"
|
||||||
|
|
||||||
|
[[package]]
|
||||||
|
name = "problem09"
|
||||||
|
version = "0.1.0"
|
||||||
|
dependencies = [
|
||||||
|
"anyhow",
|
||||||
|
"rand",
|
||||||
|
"serde",
|
||||||
|
"serde_json",
|
||||||
|
"tokio",
|
||||||
|
]
|
||||||
|
|
||||||
|
[[package]]
|
||||||
|
name = "proc-macro2"
|
||||||
|
version = "1.0.106"
|
||||||
|
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||||
|
checksum = "8fd00f0bb2e90d81d1044c2b32617f68fcb9fa3bb7640c23e9c748e53fb30934"
|
||||||
|
dependencies = [
|
||||||
|
"unicode-ident",
|
||||||
|
]
|
||||||
|
|
||||||
|
[[package]]
|
||||||
|
name = "quote"
|
||||||
|
version = "1.0.46"
|
||||||
|
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||||
|
checksum = "dfbc457d0c7a0759a614551b11a6409e5951f6c7537be1f1b7682b9ae9230368"
|
||||||
|
dependencies = [
|
||||||
|
"proc-macro2",
|
||||||
|
]
|
||||||
|
|
||||||
|
[[package]]
|
||||||
|
name = "r-efi"
|
||||||
|
version = "6.0.0"
|
||||||
|
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||||
|
checksum = "f8dcc9c7d52a811697d2151c701e0d08956f92b0e24136cf4cf27b57a6a0d9bf"
|
||||||
|
|
||||||
|
[[package]]
|
||||||
|
name = "rand"
|
||||||
|
version = "0.10.2"
|
||||||
|
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||||
|
checksum = "c7f5fa3a058cd35567ef9bfa5e75732bee0f9e4c55fa90477bef2dfcdbc4be80"
|
||||||
|
dependencies = [
|
||||||
|
"chacha20",
|
||||||
|
"getrandom",
|
||||||
|
"rand_core",
|
||||||
|
]
|
||||||
|
|
||||||
|
[[package]]
|
||||||
|
name = "rand_core"
|
||||||
|
version = "0.10.1"
|
||||||
|
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||||
|
checksum = "63b8176103e19a2643978565ca18b50549f6101881c443590420e4dc998a3c69"
|
||||||
|
|
||||||
|
[[package]]
|
||||||
|
name = "redox_syscall"
|
||||||
|
version = "0.5.18"
|
||||||
|
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||||
|
checksum = "ed2bf2547551a7053d6fdfafda3f938979645c44812fbfcda098faae3f1a362d"
|
||||||
|
dependencies = [
|
||||||
|
"bitflags",
|
||||||
|
]
|
||||||
|
|
||||||
|
[[package]]
|
||||||
|
name = "scopeguard"
|
||||||
|
version = "1.2.0"
|
||||||
|
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||||
|
checksum = "94143f37725109f92c262ed2cf5e59bce7498c01bcc1502d7b9afe439a4e9f49"
|
||||||
|
|
||||||
|
[[package]]
|
||||||
|
name = "serde"
|
||||||
|
version = "1.0.228"
|
||||||
|
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||||
|
checksum = "9a8e94ea7f378bd32cbbd37198a4a91436180c5bb472411e48b5ec2e2124ae9e"
|
||||||
|
dependencies = [
|
||||||
|
"serde_core",
|
||||||
|
"serde_derive",
|
||||||
|
]
|
||||||
|
|
||||||
|
[[package]]
|
||||||
|
name = "serde_core"
|
||||||
|
version = "1.0.228"
|
||||||
|
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||||
|
checksum = "41d385c7d4ca58e59fc732af25c3983b67ac852c1a25000afe1175de458b67ad"
|
||||||
|
dependencies = [
|
||||||
|
"serde_derive",
|
||||||
|
]
|
||||||
|
|
||||||
|
[[package]]
|
||||||
|
name = "serde_derive"
|
||||||
|
version = "1.0.228"
|
||||||
|
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||||
|
checksum = "d540f220d3187173da220f885ab66608367b6574e925011a9353e4badda91d79"
|
||||||
|
dependencies = [
|
||||||
|
"proc-macro2",
|
||||||
|
"quote",
|
||||||
|
"syn",
|
||||||
|
]
|
||||||
|
|
||||||
|
[[package]]
|
||||||
|
name = "serde_json"
|
||||||
|
version = "1.0.150"
|
||||||
|
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||||
|
checksum = "e8014e44b4736ed0538adeecded0fce2a272f22dc9578a7eb6b2d9993c74cfb9"
|
||||||
|
dependencies = [
|
||||||
|
"itoa",
|
||||||
|
"memchr",
|
||||||
|
"serde",
|
||||||
|
"serde_core",
|
||||||
|
"zmij",
|
||||||
|
]
|
||||||
|
|
||||||
|
[[package]]
|
||||||
|
name = "signal-hook-registry"
|
||||||
|
version = "1.4.8"
|
||||||
|
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||||
|
checksum = "c4db69cba1110affc0e9f7bcd48bbf87b3f4fc7c61fc9155afd4c469eb3d6c1b"
|
||||||
|
dependencies = [
|
||||||
|
"errno",
|
||||||
|
"libc",
|
||||||
|
]
|
||||||
|
|
||||||
|
[[package]]
|
||||||
|
name = "smallvec"
|
||||||
|
version = "1.15.2"
|
||||||
|
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||||
|
checksum = "8ed6a63f02c8539c91a8685a86f4099661ba3da017932f6ebbea6de3f0fa7c90"
|
||||||
|
|
||||||
|
[[package]]
|
||||||
|
name = "socket2"
|
||||||
|
version = "0.6.5"
|
||||||
|
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||||
|
checksum = "c3d1e2c7f27f8d4cb10542a02c49005dbd6e93095799d6f3be745fae9f8fedd4"
|
||||||
|
dependencies = [
|
||||||
|
"libc",
|
||||||
|
"windows-sys",
|
||||||
|
]
|
||||||
|
|
||||||
|
[[package]]
|
||||||
|
name = "syn"
|
||||||
|
version = "2.0.119"
|
||||||
|
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||||
|
checksum = "872831b642d1a07999a962a351ed35b955ea2cfc8f3862091e2a240a84f17297"
|
||||||
|
dependencies = [
|
||||||
|
"proc-macro2",
|
||||||
|
"quote",
|
||||||
|
"unicode-ident",
|
||||||
|
]
|
||||||
|
|
||||||
|
[[package]]
|
||||||
|
name = "tokio"
|
||||||
|
version = "1.52.4"
|
||||||
|
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||||
|
checksum = "317fafbbe3f02fc663dad00ea6186197de963cd4190e86a26d8d0fae095539af"
|
||||||
|
dependencies = [
|
||||||
|
"bytes",
|
||||||
|
"libc",
|
||||||
|
"mio",
|
||||||
|
"parking_lot",
|
||||||
|
"pin-project-lite",
|
||||||
|
"signal-hook-registry",
|
||||||
|
"socket2",
|
||||||
|
"tokio-macros",
|
||||||
|
"windows-sys",
|
||||||
|
]
|
||||||
|
|
||||||
|
[[package]]
|
||||||
|
name = "tokio-macros"
|
||||||
|
version = "2.7.0"
|
||||||
|
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||||
|
checksum = "385a6cb71ab9ab790c5fe8d67f1645e6c450a7ce006a33de03daa956cf70a496"
|
||||||
|
dependencies = [
|
||||||
|
"proc-macro2",
|
||||||
|
"quote",
|
||||||
|
"syn",
|
||||||
|
]
|
||||||
|
|
||||||
|
[[package]]
|
||||||
|
name = "unicode-ident"
|
||||||
|
version = "1.0.24"
|
||||||
|
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||||
|
checksum = "e6e4313cd5fcd3dad5cafa179702e2b244f760991f45397d14d4ebf38247da75"
|
||||||
|
|
||||||
|
[[package]]
|
||||||
|
name = "wasi"
|
||||||
|
version = "0.11.1+wasi-snapshot-preview1"
|
||||||
|
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||||
|
checksum = "ccf3ec651a847eb01de73ccad15eb7d99f80485de043efb2f370cd654f4ea44b"
|
||||||
|
|
||||||
|
[[package]]
|
||||||
|
name = "windows-link"
|
||||||
|
version = "0.2.1"
|
||||||
|
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||||
|
checksum = "f0805222e57f7521d6a62e36fa9163bc891acd422f971defe97d64e70d0a4fe5"
|
||||||
|
|
||||||
|
[[package]]
|
||||||
|
name = "windows-sys"
|
||||||
|
version = "0.61.2"
|
||||||
|
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||||
|
checksum = "ae137229bcbd6cdf0f7b80a31df61766145077ddf49416a728b02cb3921ff3fc"
|
||||||
|
dependencies = [
|
||||||
|
"windows-link",
|
||||||
|
]
|
||||||
|
|
||||||
|
[[package]]
|
||||||
|
name = "zmij"
|
||||||
|
version = "1.0.23"
|
||||||
|
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||||
|
checksum = "29666d0abbfad1e3dc4dcf6144730dd3a3ab225bbbdac83319345b1b44ccfc1b"
|
||||||
@@ -0,0 +1,11 @@
|
|||||||
|
[package]
|
||||||
|
name = "problem09"
|
||||||
|
version = "0.1.0"
|
||||||
|
edition = "2024"
|
||||||
|
|
||||||
|
[dependencies]
|
||||||
|
anyhow = "1"
|
||||||
|
tokio = {version = "1", features = ["full"]}
|
||||||
|
serde = {version = "1", features = ["derive"]}
|
||||||
|
serde_json = "1"
|
||||||
|
rand = "0.10"
|
||||||
@@ -0,0 +1,410 @@
|
|||||||
|
use std::{
|
||||||
|
collections::BTreeMap,
|
||||||
|
sync::{Arc, atomic::AtomicU64},
|
||||||
|
};
|
||||||
|
|
||||||
|
use serde::{Deserialize, Serialize};
|
||||||
|
use serde_json::Value;
|
||||||
|
use tokio::{
|
||||||
|
io::{AsyncReadExt, AsyncWriteExt},
|
||||||
|
net::{TcpListener, TcpStream},
|
||||||
|
sync::{
|
||||||
|
Mutex,
|
||||||
|
broadcast::{self, Sender},
|
||||||
|
},
|
||||||
|
};
|
||||||
|
|
||||||
|
use anyhow::{Result, bail};
|
||||||
|
|
||||||
|
type Queue = BTreeMap<String, Vec<Job>>;
|
||||||
|
|
||||||
|
static NEXT_CONN_ID: AtomicU64 = AtomicU64::new(0);
|
||||||
|
|
||||||
|
#[tokio::main]
|
||||||
|
async fn main() -> Result<()> {
|
||||||
|
let listener = TcpListener::bind("0.0.0.0:10000").await?;
|
||||||
|
|
||||||
|
let queues = Arc::new(Mutex::new(Queue::new()));
|
||||||
|
let (tx, _) = broadcast::channel(100_000);
|
||||||
|
|
||||||
|
loop {
|
||||||
|
let (socket, _) = listener.accept().await?;
|
||||||
|
|
||||||
|
let conn_id = next_connection_id();
|
||||||
|
let these_queues = queues.clone();
|
||||||
|
let this_tx = tx.clone();
|
||||||
|
|
||||||
|
tokio::spawn(async move {
|
||||||
|
match handle_connection(socket, these_queues.clone(), this_tx.clone(), conn_id).await {
|
||||||
|
Ok(()) => (),
|
||||||
|
Err(e) => {
|
||||||
|
dbg!(e);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
match release_jobs_for_connection(these_queues, conn_id, this_tx).await {
|
||||||
|
Ok(()) => (),
|
||||||
|
Err(e) => {
|
||||||
|
dbg!(e);
|
||||||
|
}
|
||||||
|
};
|
||||||
|
});
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn release_jobs_for_connection(
|
||||||
|
queues: Arc<Mutex<Queue>>,
|
||||||
|
conn_id: ConnectionId,
|
||||||
|
tx: Sender<Notification>,
|
||||||
|
) -> Result<()> {
|
||||||
|
let mut released = false;
|
||||||
|
let mut map = queues.lock().await;
|
||||||
|
for jobs in map.values_mut() {
|
||||||
|
for job in jobs.iter_mut() {
|
||||||
|
if job.owner == Some(conn_id) {
|
||||||
|
job.owner = None;
|
||||||
|
released = true;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
drop(map);
|
||||||
|
if released {
|
||||||
|
tx.send(Notification::QueueExtended)?;
|
||||||
|
}
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
|
fn next_connection_id() -> ConnectionId {
|
||||||
|
ConnectionId(NEXT_CONN_ID.fetch_add(1, std::sync::atomic::Ordering::Relaxed))
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn handle_connection(
|
||||||
|
mut socket: TcpStream,
|
||||||
|
these_queues: Arc<Mutex<Queue>>,
|
||||||
|
tx: Sender<Notification>,
|
||||||
|
conn_id: ConnectionId,
|
||||||
|
) -> Result<()> {
|
||||||
|
let (mut reader, mut writer) = socket.split();
|
||||||
|
|
||||||
|
let mut read_buffer = Vec::with_capacity(128_000);
|
||||||
|
|
||||||
|
loop {
|
||||||
|
read_to_buffer(&mut reader, &mut read_buffer).await?;
|
||||||
|
|
||||||
|
let message = extract_next_message(&mut read_buffer)?;
|
||||||
|
|
||||||
|
let client_message: ClientMessage = match serde_json::from_str(&message) {
|
||||||
|
Ok(parsed) => parsed,
|
||||||
|
Err(e) => {
|
||||||
|
let server_message = ServerMessage::error(e.to_string());
|
||||||
|
|
||||||
|
let reply = serde_json::to_string(&server_message).expect("we are screwed");
|
||||||
|
writer.write_all(&reply.into_bytes()).await?;
|
||||||
|
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
};
|
||||||
|
|
||||||
|
match client_message {
|
||||||
|
ClientMessage::Put { queue, job, pri } => {
|
||||||
|
handle_put_request(&mut writer, queue, job, pri, &these_queues, tx).await?;
|
||||||
|
}
|
||||||
|
ClientMessage::Get { queues, wait } => {
|
||||||
|
handle_get_request(&mut writer, queues, wait, &these_queues, &tx, conn_id).await?;
|
||||||
|
}
|
||||||
|
ClientMessage::Delete { id } => {
|
||||||
|
handle_delete_request(&mut writer, id, &these_queues).await?;
|
||||||
|
}
|
||||||
|
ClientMessage::Abort { id } => {
|
||||||
|
handle_abort_request(&mut writer, id, conn_id).await?;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
todo!()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn handle_abort_request(
|
||||||
|
writer: &mut (impl AsyncWriteExt + Unpin),
|
||||||
|
id: u64,
|
||||||
|
conn_id: ConnectionId,
|
||||||
|
) -> Result<()> {
|
||||||
|
todo!()
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn handle_delete_request(
|
||||||
|
writer: &mut (impl AsyncWriteExt + Unpin),
|
||||||
|
id: u64,
|
||||||
|
these_queues: &Mutex<Queue>,
|
||||||
|
) -> Result<()> {
|
||||||
|
let mut btree_map = these_queues.lock().await;
|
||||||
|
let mut queue_to_delete_in = None;
|
||||||
|
|
||||||
|
'outer: for (queue, jobs) in btree_map.iter_mut() {
|
||||||
|
for job in jobs {
|
||||||
|
if job.id == id {
|
||||||
|
queue_to_delete_in = Some(queue.to_string());
|
||||||
|
break 'outer;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
let server_message = match queue_to_delete_in {
|
||||||
|
Some(queue) => {
|
||||||
|
let vec = btree_map.get_mut(&queue).unwrap();
|
||||||
|
let pos = vec.iter().position(|j| j.id == id).unwrap();
|
||||||
|
vec.remove(pos);
|
||||||
|
|
||||||
|
if vec.is_empty() {
|
||||||
|
btree_map.remove(&queue);
|
||||||
|
}
|
||||||
|
|
||||||
|
ServerMessage::ok()
|
||||||
|
}
|
||||||
|
None => ServerMessage::no_job(),
|
||||||
|
};
|
||||||
|
|
||||||
|
let reply = serde_json::to_string(&server_message)?;
|
||||||
|
writer.write_all(&reply.into_bytes()).await?;
|
||||||
|
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn handle_put_request(
|
||||||
|
writer: &mut (impl AsyncWriteExt + Unpin),
|
||||||
|
queue: String,
|
||||||
|
job: Value,
|
||||||
|
pri: u64,
|
||||||
|
the_queues: &Arc<Mutex<Queue>>,
|
||||||
|
tx: Sender<Notification>,
|
||||||
|
) -> Result<()> {
|
||||||
|
let job = Job::new(job, pri);
|
||||||
|
let id = job.id;
|
||||||
|
|
||||||
|
the_queues
|
||||||
|
.lock()
|
||||||
|
.await
|
||||||
|
.entry(queue)
|
||||||
|
.or_insert_with(Vec::new)
|
||||||
|
.push(job);
|
||||||
|
|
||||||
|
let server_message = ServerMessage::ok_with_id(id);
|
||||||
|
let reply = serde_json::to_string(&server_message)?;
|
||||||
|
writer.write_all(&reply.into_bytes()).await?;
|
||||||
|
|
||||||
|
tx.send(Notification::QueueExtended)?;
|
||||||
|
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn handle_get_request(
|
||||||
|
writer: &mut (impl AsyncWriteExt + Unpin),
|
||||||
|
queues: Vec<String>,
|
||||||
|
wait: bool,
|
||||||
|
the_queues: &Arc<Mutex<Queue>>,
|
||||||
|
tx: &Sender<Notification>,
|
||||||
|
conn_id: ConnectionId,
|
||||||
|
) -> Result<()> {
|
||||||
|
let (queue, job) = loop {
|
||||||
|
let mut btree_map = the_queues.lock().await;
|
||||||
|
|
||||||
|
let best = find_highest_pri_job_across_all_queues(&mut btree_map, &queues);
|
||||||
|
|
||||||
|
match best {
|
||||||
|
Some((queue, mut job)) => {
|
||||||
|
job.owner = Some(conn_id);
|
||||||
|
break (queue, job.clone());
|
||||||
|
}
|
||||||
|
None => {
|
||||||
|
if !wait {
|
||||||
|
let server_message = ServerMessage::no_job();
|
||||||
|
let reply = serde_json::to_string(&server_message)?;
|
||||||
|
writer.write_all(&reply.into_bytes()).await?;
|
||||||
|
|
||||||
|
return Ok(());
|
||||||
|
}
|
||||||
|
let mut rx = tx.subscribe();
|
||||||
|
drop(btree_map);
|
||||||
|
rx.recv().await?;
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
};
|
||||||
|
|
||||||
|
let server_message = ServerMessage::ok_with_job(job.id, job.job.clone(), job.pri, queue);
|
||||||
|
let reply = serde_json::to_string(&server_message)?;
|
||||||
|
writer.write_all(&reply.into_bytes()).await?;
|
||||||
|
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
|
fn find_highest_pri_job_across_all_queues(
|
||||||
|
btree_map: &mut tokio::sync::MutexGuard<'_, BTreeMap<String, Vec<Job>>>,
|
||||||
|
queues: &[String],
|
||||||
|
) -> Option<(String, Job)> {
|
||||||
|
let best = btree_map
|
||||||
|
.iter_mut()
|
||||||
|
.filter(|(s, _)| queues.contains(*s))
|
||||||
|
.filter_map(|(s, vec)| {
|
||||||
|
vec.iter_mut()
|
||||||
|
.filter(|j| j.owner.is_none())
|
||||||
|
.max_by_key(|j| j.pri)
|
||||||
|
.map(|j| (s.clone(), j))
|
||||||
|
})
|
||||||
|
.max_by_key(|(_, j)| j.pri);
|
||||||
|
|
||||||
|
best.map(|(s, j)| (s, j.clone()))
|
||||||
|
}
|
||||||
|
|
||||||
|
fn extract_next_message(read_buffer: &mut Vec<u8>) -> Result<String> {
|
||||||
|
let next_newline = read_buffer
|
||||||
|
.iter()
|
||||||
|
.position(|c| *c == b'\n')
|
||||||
|
.expect("buffer was filled with newline");
|
||||||
|
|
||||||
|
let chunk: Vec<_> = read_buffer.drain(0..=next_newline).collect();
|
||||||
|
let result = String::from_utf8(chunk)?;
|
||||||
|
Ok(result)
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn read_to_buffer(
|
||||||
|
reader: &mut (impl AsyncReadExt + Unpin),
|
||||||
|
read_buffer: &mut Vec<u8>,
|
||||||
|
) -> Result<(), anyhow::Error> {
|
||||||
|
while !read_buffer.contains(&b'\n') {
|
||||||
|
let read_amount = reader.read_buf(read_buffer).await?;
|
||||||
|
if read_amount == 0 {
|
||||||
|
bail!("client disconnected")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
|
#[derive(Clone, Debug, PartialEq)]
|
||||||
|
struct Job {
|
||||||
|
job: Value,
|
||||||
|
pri: u64,
|
||||||
|
id: u64,
|
||||||
|
owner: Option<ConnectionId>,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl Job {
|
||||||
|
fn new(job: Value, pri: u64) -> Self {
|
||||||
|
let id = rand::random::<u64>();
|
||||||
|
let owner = None;
|
||||||
|
Self {
|
||||||
|
job,
|
||||||
|
pri,
|
||||||
|
id,
|
||||||
|
owner,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
#[derive(Serialize, Debug)]
|
||||||
|
struct ServerMessage {
|
||||||
|
status: Status,
|
||||||
|
#[serde(skip_serializing_if = "Option::is_none")]
|
||||||
|
id: Option<u64>,
|
||||||
|
#[serde(skip_serializing_if = "Option::is_none")]
|
||||||
|
job: Option<serde_json::Value>,
|
||||||
|
#[serde(skip_serializing_if = "Option::is_none")]
|
||||||
|
pri: Option<u64>,
|
||||||
|
#[serde(skip_serializing_if = "Option::is_none")]
|
||||||
|
queue: Option<String>,
|
||||||
|
#[serde(skip_serializing_if = "Option::is_none")]
|
||||||
|
error: Option<String>,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl ServerMessage {
|
||||||
|
fn ok_with_id(id: u64) -> Self {
|
||||||
|
Self {
|
||||||
|
status: Status::Ok,
|
||||||
|
id: Some(id),
|
||||||
|
job: None,
|
||||||
|
pri: None,
|
||||||
|
queue: None,
|
||||||
|
error: None,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
fn ok() -> Self {
|
||||||
|
Self {
|
||||||
|
status: Status::Ok,
|
||||||
|
id: None,
|
||||||
|
job: None,
|
||||||
|
pri: None,
|
||||||
|
queue: None,
|
||||||
|
error: None,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
fn ok_with_job(id: u64, job: Value, pri: u64, queue: String) -> Self {
|
||||||
|
Self {
|
||||||
|
status: Status::Ok,
|
||||||
|
id: Some(id),
|
||||||
|
job: Some(job),
|
||||||
|
pri: Some(pri),
|
||||||
|
queue: Some(queue),
|
||||||
|
error: None,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
fn no_job() -> Self {
|
||||||
|
Self {
|
||||||
|
status: Status::NoJob,
|
||||||
|
id: None,
|
||||||
|
job: None,
|
||||||
|
pri: None,
|
||||||
|
queue: None,
|
||||||
|
error: None,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
fn error(error: String) -> Self {
|
||||||
|
Self {
|
||||||
|
status: Status::Error,
|
||||||
|
id: None,
|
||||||
|
job: None,
|
||||||
|
pri: None,
|
||||||
|
queue: None,
|
||||||
|
error: Some(error),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
#[derive(Deserialize, Debug)]
|
||||||
|
#[serde(rename_all = "kebab-case", tag = "request", deny_unknown_fields)]
|
||||||
|
enum ClientMessage {
|
||||||
|
Put {
|
||||||
|
queue: String,
|
||||||
|
job: serde_json::Value,
|
||||||
|
pri: u64,
|
||||||
|
},
|
||||||
|
Get {
|
||||||
|
queues: Vec<String>,
|
||||||
|
#[serde(default)]
|
||||||
|
wait: bool,
|
||||||
|
},
|
||||||
|
Delete {
|
||||||
|
id: u64,
|
||||||
|
},
|
||||||
|
Abort {
|
||||||
|
id: u64,
|
||||||
|
},
|
||||||
|
}
|
||||||
|
|
||||||
|
#[derive(Serialize, Debug)]
|
||||||
|
#[serde(rename_all = "kebab-case")]
|
||||||
|
enum Status {
|
||||||
|
Ok,
|
||||||
|
Error,
|
||||||
|
NoJob,
|
||||||
|
}
|
||||||
|
|
||||||
|
#[derive(Debug, Clone)]
|
||||||
|
enum Notification {
|
||||||
|
QueueExtended,
|
||||||
|
}
|
||||||
|
|
||||||
|
#[derive(Debug, Clone, Copy, PartialEq)]
|
||||||
|
struct ConnectionId(u64);
|
||||||
Reference in New Issue
Block a user