Advanced Rust Programming Guide

WebSockets and Real-Time Communication

Real-time Rust servers often use asynchronous WebSocket crates (e.g. Tokio-tungstenite or warp::ws)
to handle bidirectional streams. A common pattern is to maintain a thread-safe connection registry
(often an Arc<Mutex<HashMap<…>>> ) that tracks all active client connections 1 . For example, one
can store each client’s send-channel in a shared map:

use std::collections::HashMap;
use tokio::sync::mpsc;
use std::sync::Arc;
use tokio::sync::Mutex;

type ConnectionId = String;

type Tx = mpsc::UnboundedSender<String>; // channel to send messages to
client
type ConnMap = Arc<Mutex<HashMap<ConnectionId, Tx>>>;

async fn handle_new_connection(conn_map: ConnMap, id: ConnectionId, mut

ws_receiver: tokio_tungstenite::WebSocketStream<tokio::net::TcpStream>) {
let (tx, mut rx) = mpsc::unbounded_channel();
conn_map.lock().[Link]([Link](), tx);
// Spawn task to forward received messages to the WebSocket
tokio::spawn(async move {
while let Some(msg) = [Link]().await {

ws_receiver.send(tokio_tungstenite::tungstenite::Message::Text(msg)).[Link]();
}
});
// Process incoming messages from the client
while let Some(msg) = ws_receiver.next().await {
if let Ok(tokio_tungstenite::tungstenite::Message::Text(text)) = msg
{
// Broadcast to all connected clients
let map = conn_map.lock().await;
for sender in [Link]() {
let _ = [Link]([Link]());
}
}
}
// On disconnect, remove the connection
conn_map.lock().[Link](&id);
}

1
This pattern (Arc + Mutex protecting a HashMap of client channels) ensures safe, mutable shared state
1 . One can also use tokio::sync::broadcast channels for efficient event distribution to many

subscribers. For example, creating a broadcast channel ( let (tx, _rx) =

broadcast::channel(32) ) lets a sender .send(...) to multiple receivers that each use
[Link]() 2 .

To handle client heartbeats and status, the server can periodically send Ping frames or application-
layer pings. For instance, use tokio::time::interval() in a task to send a ping or heartbeat
message to each client channel:

use tokio::time::{interval, Duration};

let mut ticker = interval(Duration::from_secs(5));
loop {
[Link]().await;
let map = conn_map.lock().await;
for tx in [Link]() {
let _ = [Link]("__heartbeat__".into());
}
}

Clients should reply with a Pong or application response. If a client fails to respond (or its channel
closes), the server can mark it offline and remove it from the registry, thus synchronizing “online/offline”
status.

Group Messaging and Chat Rooms

For group chat or “rooms,” maintain a map of room IDs to sets of client IDs. Each client may belong to
multiple rooms. For example:

use std::collections::{HashMap, HashSet};

type RoomId = String;
type RoomMap = Arc<Mutex<HashMap<RoomId, HashSet<ConnectionId>>>>;

// Add client to room

async fn join_room(room_map: RoomMap, room: RoomId, client_id: ConnectionId)
{
let mut rooms = room_map.lock().await;
[Link](room).or_default().insert(client_id);
}

// Broadcast to a specific room

async fn send_to_room(room_map: RoomMap, conn_map: ConnMap, room: &RoomId,
message: &str) {
let rooms = room_map.lock().await;
if let Some(clients) = [Link](room) {
let clients = [Link]();
drop(rooms);
let map = conn_map.lock().await;
for client_id in clients {

2
 if let Some(tx) = [Link](client_id) {
let _ = [Link](message.to_string());
}
}
}
}

Above, room_map tracks membership. On a message, the server looks up the target room and
broadcasts only to clients in that set. This avoids sending to all connections. If using Tokio’s broadcast
channel for each room, then each subscriber simply listens on a per-room broadcast receiver,
automatically receiving the group’s messages 2 .

Session Lifecycle and Reconnection

Maintain clean lifecycles by removing connections on disconnect (as shown above). To support
reconnection or session recovery, the server can generate persistent session IDs and re-associate new
WebSocket streams with old state. For example, on initial HTTP registration one issues a UUID; the
client connects with [Link] . On connect, call register_client(uuid) to insert into
the map 1 . On disconnect, simply delete or mark offline. If a client reconnects with the same ID, you
restore any queued messages or re-send missed events.

Asynchronous Event Distribution

Tokio provides multiple async primitives for messaging: besides broadcast channels, one can use
tokio::sync::mpsc or create an event bus. A common pattern is to spawn tasks that read from
channels and dispatch to clients. For instance, a dedicated task could read from a
tokio::sync::mpsc::Receiver<ServerEvent> and, upon each event, iterate the active
connections or relevant room and forward the message.

Using tokio::sync::broadcast , multiple tasks or services can subscribe to events and push into
the WebSocket senders. For example, to broadcast notifications to all WebSocket clients, one could
[Link](notification) and have each client handler wait on [Link]().await 2 . This
architecture decouples event producers from consumers.

File Processing
Rust excels at high-performance I/O. For large-file streaming or memory-mapped I/O, use crates like
memmap2 or std::fs::File . Example:

use memmap2::MmapOptions;
use std::fs::File;

// Memory-map a file for fast access

let file = File::open("[Link]")?;
let mmap = unsafe { MmapOptions::new().map(&file)? };
// Now `mmap` is a &[u8] referencing file contents
for line in [Link](|&b| b == b'\n') {
if let Ok(s) = std::str::from_utf8(line) {
process_line(s);

3
 }
}

For streaming reads, use buffered I/O and iterate lines or chunks to avoid loading entire file into
memory:

use std::io::{BufReader, BufRead};

use std::fs::File;

let file = File::open("[Link]")?;

let reader = BufReader::with_capacity(64*1024, file); // 64KB buffer
for line_res in [Link]() {
let line = line_res?;
// Process each CSV line
}

CSV/JSON Parsing with Error Recovery

The csv crate offers robust CSV parsing. A typical pattern is to create a csv::Reader , iterate
records, and handle parse errors. For example, to batch parse and skip invalid lines:

use csv::StringRecord;
let mut rdr = csv::ReaderBuilder::new()
.flexible(true) // allow variable-length records
.has_headers(true)
.from_path("[Link]")?;
for result in [Link]() {
match result {
Ok(record) => {
// Convert record to struct or process fields
let row: Vec<&str> = [Link]().collect();
// ...
}
Err(e) => {
eprintln!("CSV parse error: {}", e);
// Optionally skip or retry
}
}
}

For JSON, use Serde’s streaming interface. To parse a large JSON array from a file without allocating all
at once:

use serde_json::Deserializer;
use std::fs::File;

let file = File::open("[Link]")?;

4
 let stream = Deserializer::from_reader(file).into_iter::<MyStruct>();
for item in stream {
match item {
Ok(val) => process(val),
Err(e) => {
eprintln!("JSON parse error, skipping: {}", e);
// Continue to next item
}
}
}

Here StreamDeserializer yields each JSON object. If a parse error occurs, handle or skip as above.

Parallel File Operations

For CPU-bound file processing or multi-file workloads, Rayon enables easy parallelism 3 . By calling
par_iter() on a collection, you can distribute file reads across threads. Example: computing
checksums of many files in parallel:

use rayon::prelude::*;
use std::fs;

let paths = vec!["[Link]", "[Link]", "[Link]"];

paths.par_iter().for_each(|path| {
let data = fs::read(path).unwrap();
let checksum = compute_checksum(&data);
println!("{} -> {:#x}", path, checksum);
});

Rayon’s par_iter() will run the loop body concurrently, splitting work among threads 3 . For
asynchronous file I/O (e.g. with tokio::fs ), you can also spawn multiple async tasks to read different
files concurrently.

Real-world ETL and Logging

In practice, file processing often integrates with logging, transformation, or ETL pipelines. For
instance, reading CSV logs, transforming fields, and writing out to a database or another file. Typical
structure:

for batch in [Link]().chunks(100) {

let mut batch_data = Vec::with_capacity(100);
for record in batch {
let record = record?;
// transform record fields
let row: MyRow = [Link](None)?;
batch_data.push(row);
}

5
 // Bulk insert batch_data into database or write to disk
}

When writing, prefer buffered I/O ( BufWriter ) or asynchronous writes

( tokio::fs::File::create + write_all ). Logging can use the tracing crate for high-
performance, structured logging in async contexts.

Background Job Processing

For scheduled tasks in Rust services, use crates like tokio-cron-scheduler or implement your own
loop. A simple approach using Tokio:

use tokio::time::{sleep, Duration};

use cronexpr::CronExpr;
use chrono::Utc;

async fn start_cron_loop() {
let expr = CronExpr::parse("0 0 * * * *").unwrap(); // every hour
loop {
let now = Utc::now();
if [Link](&now) {
tokio::spawn(async {
// Your scheduled job here
println!("Hourly job running");
});
}
sleep(Duration::from_secs(1)).await;
}
}

Alternatively, the cronexpr crate can parse cron strings and compute next run times. One tutorial
shows building an “embeddable cron system” by using cronexpr with Tokio timers 4 .

For task queues with retries, you can use in-memory channels or a persistent queue. For a simple in-
memory queue, use a tokio::sync::mpsc channel and worker tasks. For durability, libraries like
Fang provide a full-featured job queue that backs tasks to Postgres or SQLite. Fang supports scheduled
and periodic jobs, unique tasks, and retries 5 6 . For example, to enqueue a job in Fang:

// Pseudo-code, actual API may differ

use fang::{Executor, Job};
let executor = Executor::builder().connect("postgres://...").build().await?;
[Link](Job::new("send_email", job_data)).await?;

Fang will persist the job, run it with async workers, and can retry on failure 5 6 .

For persistent job state, one can store job status in a database table. For instance, use Diesel or SQLx
to write a jobs table with fields (id, status, attempts, payload) . Each worker updates the

6
status from “pending” to “running” to “done” or “failed,” enabling retries and restarts. The Hangfire or
RQ patterns apply: pick up tasks where you left off.

Cancellation and Observability

Implement graceful cancellation by listening for shutdown signals and using a cancellation token.
Tokio provides a CancellationToken that can be cloned and polled in tasks. When you call
[Link]() , all clones see cancellation 7 . For example:

use tokio_util::sync::CancellationToken;
let token = CancellationToken::new();
let child_token = [Link]();
tokio::spawn(async move {
loop {
tokio::select! {
_ = child_token.cancelled() => {
println!("Job cancelled");
break;
}
_ = do_some_work() => {}
}
}
});
// Later, to cancel:
[Link]();

This allows tasks to exit cleanly. Additionally, use tokio::signal::ctrl_c() to catch Ctrl-C and
trigger shutdown logic 8 . For observability, integrate tracing or metrics crates to record job status,
execution time, and failures.

Domain-Driven Design (DDD) in Rust

While Rust is not object-oriented, one can apply DDD by modeling aggregates and value objects as
structs and enums. Use Rust’s strong type system: wrap IDs in newtypes, define domain errors with
thiserror , etc. For example, an aggregate and repository trait:

// Domain layer: aggregate and value objects

#[derive(Debug)]
pub struct UserId(uuid::Uuid);

#[derive(Debug)]
pub enum UserStatus { Active, Suspended }

#[derive(Debug)]
pub struct User {
pub id: UserId,
pub name: String,
pub status: UserStatus,
}

7
 impl User {
// domain logic, e.g. change status
pub fn suspend(&mut self) -> Result<(), String> {
if [Link] == UserStatus::Active {
[Link] = UserStatus::Suspended;
Ok(())
} else {
Err("User not active".into())
}
}
}

// Repository interface (port)

pub trait UserRepository {
fn find(&self, id: &UserId) -> Result<User, anyhow::Error>;
fn save(&self, user: &User) -> Result<(), anyhow::Error>;
}

This follows the DDD idea of a clear domain layer (with aggregates like User , methods with invariants)
and an interface for persistence. The UserRepository trait abstracts storage, aligning with the Ports
and Adapters (Hexagonal) architecture. An implementation in the infrastructure layer could use SQLx
or Diesel to fulfill the trait.

For example, an in-memory repository for tests:

use std::sync::RwLock;
pub struct InMemoryUserRepo {
data: RwLock<HashMap<uuid::Uuid, User>>,
}
impl UserRepository for InMemoryUserRepo {
fn find(&self, id: &UserId) -> Result<User, anyhow::Error> {
[Link]().unwrap()
.get(&id.0).cloned().ok_or_else(|| anyhow::anyhow!("Not found"))
}
fn save(&self, user: &User) -> Result<(), anyhow::Error> {
[Link]().unwrap().insert([Link].0, [Link]());
Ok(())
}
}

This shows a repository pattern and separation of concerns 9 . The domain doesn’t depend on
storage specifics.

Use case boundaries can be modeled as application services or command handlers. For instance, a
UserService that orchestrates use cases:

8
 pub struct UserService<R: UserRepository> {
repo: Arc<R>,
}
impl<R: UserRepository> UserService<R> {
pub fn create_user(&self, name: String) -> Result<UserId, anyhow::Error>
{
let mut user = User {
id: UserId(uuid::Uuid::new_v4()),
name,
status: UserStatus::Active,
};
[Link](&user)?;
Ok([Link])
}
// more use cases...
}

This places application logic (service) distinct from domain entity logic.

Event Sourcing / CQRS

Rust crates like cqrs-es enable CQRS/event sourcing patterns. Typically, you define commands and
events as enums and an aggregate that applies events to mutate state. For example, a simple bank
account:

pub enum AccountCommand {

Deposit { amount: u64 },
Withdraw { amount: u64 },
}

pub enum AccountEvent {

Deposited { amount: u64 },
Withdrawn { amount: u64 },
}

pub struct Account {

balance: u64,
}
impl Account {
pub fn apply(&mut self, event: &AccountEvent) {
match *event {
AccountEvent::Deposited { amount } => [Link] += amount,
AccountEvent::Withdrawn { amount } => [Link] -= amount,
}
}
pub fn handle(&self, cmd: AccountCommand) -> Vec<AccountEvent> {
match cmd {
AccountCommand::Deposit { amount } => vec!
[AccountEvent::Deposited { amount }],

9
 AccountCommand::Withdraw { amount } if [Link] >= amount =>
vec![AccountEvent::Withdrawn { amount }],
_ => vec![],
}
}
}

One can then persist each event to an event store (e.g. Postgres or even append-only files). The cqrs-
es crate provides the infrastructure to save and replay events 10 . This is an advanced pattern; for
many apps a simpler CRUD model suffices.

Functional Programming Patterns

Rust’s enums are powerful algebraic data types. For example:

enum Shape {
Circle(f64), // radius
Rectangle(f64, f64), // width, height
Polygon { sides: u32, length: f64 },
}

fn area(shape: &Shape) -> f64 {

match *shape {
Shape::Circle(r) => std::f64::consts::PI * r * r,
Shape::Rectangle(w, h) => w * h,
Shape::Polygon { sides, length } => {
// e.g. regular polygon approximation
let angle = std::f64::consts::PI * 2.0 / (sides as f64);
(sides as f64) * length * length / (4.0 * [Link]())
}
}
}

This uses enums to model a sum type (circle vs rectangle vs polygon).

Rust’s iterators and combinators ( map , filter , and_then , etc.) enable monadic transformations.
For example, chaining Option and Result:

fn parse_and_double(s: &str) -> Result<i32, std::num::ParseIntError> {

[Link]::<i32>().map(|x| x * 2)
}

Here .parse() returns a Result<i32, E> , and .map() applies on success. Chaining multiple
operations with ? or .and_then() acts like monadic binding 11 12 :

10
 fn compute(s: &str) -> Option<i32> {
Some(s).and_then(|s| [Link]().ok()).map(|x: i32| x + 1)
}

This pipelines parsing and addition without explicit match . The Rust By Example tutorial shows how
and_then() flattens nested Option<Option<T>> 12 .

For persistent (immutable) data structures, use the im crate. It provides immutable vectors,
hashmaps, etc. Example:

use im::HashMap;
let mut m1 = HashMap::new();
[Link]("a", 1);
let m2 = [Link]("a", 2); // m1 unchanged, m2 has updated value
assert_eq!([Link]("a"), Some(&1));
assert_eq!([Link]("a"), Some(&2));

The im crate ensures cheap cloning via structural sharing 13 .

You can build composable pipelines of pure functions easily. For example, processing a list of orders:

let total: u32 = orders

.iter()
.filter(|o| [Link] == Status::Completed)
.map(|o| [Link])
.sum();

This functional style ( filter , map , sum ) avoids mutation. Enums often carry data (tagged unions)
and pattern matching handles each case, keeping code declarative.

Embedded Systems
Rust supports bare-metal (no_std) development on microcontrollers. A typical setup uses #!
[no_std] , a suitable linker script, and an entry macro:

#![no_std]
#![no_main]

use cortex_m_rt::entry;
use panic_halt as _;
use stm32f4xx_hal::{pac, prelude::*};

#[entry]
fn main() -> ! {
let dp = pac::Peripherals::take().unwrap();
let gpioc = [Link]();

11
 let mut led = gpioc.pc13.into_push_pull_output();
loop {
led.set_high().unwrap();
cortex_m::asm::delay(8_000_000);
led.set_low().unwrap();
cortex_m::asm::delay(8_000_000);
}
}

Here we use stm32f4xx-hal crate for GPIO control. This example blinks an LED (on PC13 for many
Nucleo boards) by toggling the output pin. The HAL crate handles peripheral setup (clock, GPIO port)
safely. This approach (found in HAL examples) demonstrates bare-metal initialization and control 14 .

For interrupts and DMA, Rust crates like cortex-m-rtic or embassy provide frameworks. Using
RTIC:

#[rtic::app(device = stm32f4xx_hal::pac, peripherals = true)]

mod app {
#[resources]
struct Resources {
#[init(false)]
flag: bool,
}

#[task(binds = TIM2, resources = [flag])]

fn on_timer(cx: on_timer::Context) {
*[Link] = !*[Link];
}
}

RTIC handles safe interrupt setup and resource sharing. With #[interrupt] one can also write safe
handlers manually, ensuring #[interrupt] fn TIM2() reads/writes memory without data races.
The HAL crates define safe abstractions for DMA transfers as well, e.g., setting up a peripheral to DMA
memory buffers automatically.

Real-time scheduling uses crates like RTIC or Embassy: Embassy’s #[task] and async/await model
allow writing non-blocking interrupt-driven code. For example, Embassy on nRF microcontrollers
supports async/await without std , using .await inside interrupts.

In summary, embedded Rust code carefully configures hardware (via HAL), uses #![no_std] , and
relies on tools like RTIC to enforce timing and safety.

Advanced Concurrency Patterns

Rust’s async ecosystem provides many primitives for shared state and coordination. For asynchronous
state management, use locks from tokio::sync , such as RwLock or Notify . For instance, a
shared in-memory cache might be:

12
 use tokio::sync::RwLock;
use std::collections::HashMap;

let cache = Arc::new(RwLock::new(HashMap::<String, String>::new()));

// In one task:
{
let mut map = [Link]().await;
[Link]("key".into(), "value".into());
}
// In another:
{
let map = [Link]().await;
if let Some(v) = [Link]("key") { println!("{}", v); }
}

Tokio also provides Notify and watch for signaling. A Notify can awaken one or all waiting
tasks:

use tokio::sync::Notify;
let notify = Arc::new(Notify::new());
// Task A:
let notify_a = [Link]();
tokio::spawn(async move {
notify_a.notified().await;
println!("Received notification");
});
// Task B:
notify.notify_one(); // wakes up one waiting task

For task coordination, tokio::select! is invaluable 15 . It waits on multiple async futures:

tokio::select! {
_ = [Link]() => println!("Received on rx1"),
_ = [Link]() => println!("Received on rx2"),
}

This runs until one branch completes, akin to select in other languages 15 .

For parallel CPU-bound work, Rust’s built-in threads or Rayon (as above) apply. For example, use
rayon::spawn() or par_iter when doing heavy computation.

To build actor-like models, use tokio::sync::mpsc as a mailbox. Each actor is a task owning an
mpsc::Receiver<Msg> . Other parts of the system hold a mpsc::Sender<Msg> . For example:

use tokio::sync::{mpsc, oneshot};

13
 enum Command {
Add { x: i32, y: i32, resp: oneshot::Sender<i32> },
Concat { a: String, b: String, resp: oneshot::Sender<String> },
}

struct Actor {
rx: mpsc::Receiver<Command>,
name: String,
}

impl Actor {
async fn run(mut self) {
while let Some(cmd) = [Link]().await {
match cmd {
Command::Add { x, y, resp } => {
let _ = [Link](x + y);
}
Command::Concat { a, b, resp } => {
let _ = [Link](a + &b);
}
}
}
}
}

// Usage:
let (tx, rx) = mpsc::channel(32);
let actor = Actor { rx, name: "actor1".into() };
tokio::spawn([Link]());

// Send a message using a proxy:

let (resp_tx, resp_rx) = oneshot::channel();
[Link](Command::Add { x: 2, y: 3, resp: resp_tx }).[Link]();
assert_eq!(resp_rx.[Link](), 5);

This code, inspired by examples, shows an actor processing two commands from its mailbox 16 17 .
Each command carries a one-shot channel to send back a response. This isolates state within the actor
and avoids shared mutability, following the actor model pattern 16 .

These advanced concurrency tools allow building robust multi-task systems: RwLock , Notify , and
channels for internal communication; select! for coordination 15 ; Rayon or threads for parallelism
3 ; and actor/mailbox patterns for structured message-passing 16 .

14
 1 How to build a WebSocket server with Rust - LogRocket Blog
[Link]

2 tokio::sync::broadcast - Rust
[Link]

3 Implementing data parallelism with Rayon Rust - LogRocket Blog

[Link]

4 Building a Cron Job System in Rust with Tokio and Cronexpr - DEV Community
[Link]

5 6 GitHub - ayrat555/fang: Background processing for Rust

[Link]

7 8 Graceful Shutdown | Tokio - An asynchronous Rust runtime

[Link]

9 Building an API Server with Rust and DDD | by katayama8000 | Medium

[Link]

10 CQRS and Event Sourcing using Rust

[Link]

11 12 Combinators: and_then - Rust By Example

[Link]

13 im - Rust
[Link]

14 stm32f4xx-hal/examples/[Link] at master · stm32-rs/stm32f4xx-hal · GitHub

[Link]

15 Select | Tokio - An asynchronous Rust runtime

[Link]

16 17 Building an Asynchronous Actor Model in Rust using Tokio | by p546489 | Medium

[Link]

15