rust 57 lines · 9 steps

A channel-driven worker loop in Rust

One worker multiplexes commands, events, and a heartbeat over crossbeam channels with a single select loop.

Explained by highlit
Intermediate channels select message-passing event-loop binary-encoding
1use crossbeam_channel::{Receiver, Sender, select, tick};
2use std::time::Duration;
3 
4pub enum Command {
5 Publish { topic: String, payload: Vec<u8> },
6 Subscribe(String),
7 Shutdown,
8}
9 
10pub struct Worker {
11 commands: Receiver<Command>,
12 events: Receiver<Event>,
13 outbound: Sender<Vec<u8>>,
14}
15 
16pub struct Event {
17 pub topic: String,
18 pub data: Vec<u8>,
19}
20 
21impl Worker {
22 pub fn run(&self) {
23 let heartbeat = tick(Duration::from_secs(30));
24 let mut subscriptions: Vec<String> = Vec::new();
25 
26 loop {
27 select! {
28 recv(self.commands) -> msg => match msg {
29 Ok(Command::Publish { topic, payload }) => {
30 let frame = encode_frame(&topic, &payload);
31 let _ = self.outbound.send(frame);
32 }
33 Ok(Command::Subscribe(topic)) => subscriptions.push(topic),
34 Ok(Command::Shutdown) | Err(_) => break,
35 },
36 recv(self.events) -> ev => match ev {
37 Ok(event) if subscriptions.contains(&event.topic) => {
38 let _ = self.outbound.send(event.data);
39 }
40 Ok(_) => {}
41 Err(_) => break,
42 },
43 recv(heartbeat) -> _ => {
44 let _ = self.outbound.send(encode_frame("$SYS/ping", b""));
45 }
46 }
47 }
48 }
49}
50 
51fn encode_frame(topic: &str, payload: &[u8]) -> Vec<u8> {
52 let mut buf = Vec::with_capacity(topic.len() + payload.len() + 4);
53 buf.extend_from_slice(&(topic.len() as u16).to_be_bytes());
54 buf.extend_from_slice(topic.as_bytes());
55 buf.extend_from_slice(payload);
56 buf
57}
01 / 01
STEP 01

Walkthrough

Space play ←→ step click any line
Three takeaways
  1. 1select! lets a single thread wait on multiple channels and react to whichever is ready first.
  2. 2Treating a channel receive error as a shutdown signal cleanly handles a disconnected sender.
  3. 3Encoding a length prefix before variable-length data lets the reader know where each field ends.

Related explainers

rust 
use std::time::Duration;
 
use axum::{
    http::{header, HeaderValue, Request},

Serving fingerprinted assets in Axum

static-assets http-caching middleware
Intermediate 7 steps
rust 
use std::collections::VecDeque;
 
#[derive(Debug)]
pub struct Hunk {

Applying a diff hunk in Rust

enums error-handling pattern-matching
Intermediate 8 steps
rust 
use axum::{
    body::Body,
    http::{Method, StatusCode, Uri},
    response::{IntoResponse, Response},

Nesting routers and JSON fallbacks in Axum

routing http json-responses
Intermediate 7 steps
rust 
use axum::{
    extract::FromRequestParts,
    http::{request::Parts, StatusCode, header::ACCEPT_LANGUAGE},
};

A locale extractor for Axum handlers

content-negotiation http-headers custom-extractor
Intermediate 7 steps
rust 
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::thread;
use std::time::Duration;

Graceful thread shutdown with an atomic flag

concurrency atomics memory-ordering
Advanced 7 steps
rust 
#[derive(Debug, PartialEq)]
enum State {
    FieldStart,
    InUnquoted,

Parsing a CSV line with a state machine

state machine parsing enums
Intermediate 9 steps

Share this explainer

Here's the card — post it anywhere.

A channel-driven worker loop in Rust — share card
Made with highlit — turn any snippet into a walkthrough like this in about a minute.
Explain your code