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 serde::Deserialize;
 
#[derive(Debug, Deserialize)]
#[serde(untagged)]

Parsing flexible JSON shapes with serde

deserialization enums json
Intermediate 6 steps
go 
package streaming
 
import (
	"bufio"

Streaming NDJSON logs over HTTP in Go

http-streaming channels select
Advanced 10 steps
rust 
use std::cmp::{Ordering, Reverse};
use std::collections::BinaryHeap;
use std::fs::File;
use std::io::{self, BufRead, BufReader, BufWriter, Lines, Write};

K-way merge of sorted logs in Rust

binary-heap k-way-merge streaming-io
Intermediate 8 steps
rust 
use axum::{
    extract::{Path, State},
    response::sse::{Event, KeepAlive, Sse},
};

Streaming import progress with SSE in Axum

server-sent-events streams watch-channel
Advanced 7 steps
rust 
use std::f64::consts::PI;
 
#[derive(Debug, Clone, Copy)]
pub struct LatLng {

Building geographic bounding boxes in Rust

geospatial value-types option
Intermediate 7 steps
go 
func (w *Watcher) resetDebounce(d time.Duration) {
	if !w.timer.Stop() {
		select {
		case <-w.timer.C:

Debouncing a stream of events in Go

debounce timers channels
Advanced 7 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