Initial draft: rabbitmq & async

[Robin5605]

This rewrite consists of both a migration to an asynchronous model and a migration to the proposed RabbitMQ architecture, because the lapin crate is one of the most mature AMQP client crates for Rust, except it's asynchronous

Here is the gist of the new structure:

• This task will continuously refresh the access token, this makes it so that each task doesn't have to send a request, check if it's authentication failed, then re-send it. Now, all operations can simply assume the authentication is update to date. This task will refresh the access tokens 10 seconds early just to have some built-in tolerance.
  

  dragonfly-client-rs/src/main.rs

  Lines 109 to 119 in 510e63a

  	tokio::spawn({
  	let client = Arc::clone(&client);
  	async move {
  	loop {
  	let sleep = client.authentication.read().await.expires_in
  	- tokio::time::Duration::from_secs(10);
  	tokio::time::sleep(sleep).await;
  	client.reauthenticate().await;
  	}
  	}
  	});

• This task listens on the exclusive queue that will get rule updates. It doesn't really matter what comes over this queue, since it will be discarded. Each client will establish it's own, exclusive queue and bind it to the rule_updates fanout exchange. Mainframe will dispatch a message to all on this queue whenever the rules are updated. Clients will then have to fetch the new rules from /rules, compile the rules, and update their local state.
  

  dragonfly-client-rs/src/main.rs

  Lines 121 to 131 in 510e63a

  	tokio::spawn({
  	let client = Arc::clone(&client);
  	async move {
  	loop {
  	match client.receive_rule_update().await {
  	Ok(_) => client.update_rules().await,
  	Err(err) => error!("Error from rule updates queue: {err}"),
  	}
  	}
  	}
  	});

• This loop listens on the jobs queue for incoming jobs, and creates a task for each of them.
  

  dragonfly-client-rs/src/main.rs

  Lines 133 to 151 in 510e63a

  	loop {
  	trace!("Waiting for message");
  	match client.receive_delivery().await {
  	Ok(delivery) => {
  	trace!("Message received");
  	let client = Arc::clone(&client);
  	tokio::task::spawn(async move {
  	if let Err(err) = handle_delivery(Arc::clone(&client), &delivery).await {
  	error!("Error while scanning: {err}");
  	match delivery.reject(BasicRejectOptions { requeue: false }).await {
  	Ok(()) => warn!("Rejected message"),
  	Err(err) => error!("Error while rejecting: {err}"),
  	}
  	}
  	});
  	}
  	Err(err) => error!("Error while consuming: {err}"),
  	}
  	}

[AbooMinister25]

Everything looks mostly good - I left a couple of suggestions, but nothing that's a real problem. I am yet to run this, though.

[Robin5605]

TODO:

• Deduplication
• Authentication (with Keycloak)