From 4060f64b98e0bbfde3ac6a9a8f176d6ee40f0aad Mon Sep 17 00:00:00 2001 From: Md Imran Hossain Date: Mon, 28 Sep 2026 23:34:04 +0600 Subject: [PATCH] basic interactions work --- .gitignore | 1 + Cargo.toml | 2 +- README.md | 110 +++++++++++- coap-inspect.py | 52 ++++-- src/bin/stream.rs | 178 +++++++++++++++++++ src/lib.rs | 220 ++++++++++++++++++++++-- src/main.rs | 427 +++++++++++++++++++++++++++++++++++++++++++++- test-stream.sh | 82 +++++++++ 8 files changed, 1034 insertions(+), 38 deletions(-) create mode 100644 src/bin/stream.rs create mode 100755 test-stream.sh diff --git a/.gitignore b/.gitignore index b5bcc8e..2be0e14 100644 --- a/.gitignore +++ b/.gitignore @@ -1,2 +1,3 @@ target/ nodejs/node_modules/ +media/ diff --git a/Cargo.toml b/Cargo.toml index 0a1b178..d09c561 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -6,7 +6,7 @@ default-run = "coap-hello-udp" [dependencies] coap-lite = "0.13.3" -tokio = { version = "1.53.1", features = ["rt-multi-thread", "net", "macros", "signal", "time", "io-util"] } +tokio = { version = "1.53.1", features = ["rt-multi-thread", "net", "macros", "signal", "time", "io-util", "process"] } [profile.release] opt-level = 3 diff --git a/README.md b/README.md index d5d010b..1887c77 100644 --- a/README.md +++ b/README.md @@ -86,23 +86,122 @@ Or with any other CoAP client, e.g. [`coap-cli`](https://www.npmjs.com/package/c coap get "coap://localhost:5683/hello?name=Imran" ``` +## Streaming media + +The server also streams two static files, loaded into memory at startup from +`MEDIA_DIR` (default `media/`): + +| Route | File | Format | +|--------------|----------------------|-----------------------------------------------| +| `GET /audio` | `media/audio.wav` | PCM WAV, mono, 16-bit, 16 kHz | +| `GET /image` | `media/image.rgb565` | raw RGB565 little-endian, 320×240 (153,600 B) | + +A file that is missing or in the wrong format is reported at startup and its +route answers `4.04` with the reason; the rest of the server still runs. + +Neither file fits in one datagram, so they go out with block-wise transfer +(Block2, RFC 7959): the client asks for block 0, 1, 2 … and each reply carries +1024 bytes. The server keeps no per-client state — every request names the +block it wants, so it just slices the in-memory file. Every block carries an +ETag (so the client can tell the file did not change mid-transfer), block 0 +carries Size2 (the total size), and a block past the end gets `4.02`. + +### Make test media + +```bash +mkdir -p media +``` + +```bash +ffmpeg -f lavfi -i "sine=frequency=440:duration=10" -ac 1 -ar 16000 -c:a pcm_s16le media/audio.wav +``` + +```bash +ffmpeg -f lavfi -i "testsrc=size=320x240" -frames:v 1 -f rawvideo -pix_fmt rgb565le media/image.rgb565 +``` + +Or convert your own files: + +```bash +ffmpeg -i input.mp3 -ac 1 -ar 16000 -c:a pcm_s16le media/audio.wav +``` + +```bash +ffmpeg -i photo.png -vf scale=320:240 -f rawvideo -pix_fmt rgb565le media/image.rgb565 +``` + +`media/` is gitignored. + +### Play it + +`src/bin/stream.rs` downloads a file block by block and pipes it into `ffplay` +(part of ffmpeg), which opens the window: + +```bash +cargo run --release --bin stream -- image +``` + +```bash +cargo run --release --bin stream -- audio +``` + +The image opens in a 320×240 window that stays until you close it; the audio +plays with a waveform window. Closing the window stops the download. + +| Flag | Effect | +|------------------|----------------------------------------------------| +| `--out FILE` | Also save the received bytes | +| `--no-play` | Don't start ffplay (download only) | +| `--timeout-ms N` | First retransmission timeout (default 2000, doubles, 4 retries) | + +Measured on localhost: + +| File | Blocks | Download, `--no-play` | Retransmits | +|-------|--------|-----------------------|-------------| +| image | 150 | under 1 s | 0 | +| audio | 313 | 0.16 s | 0 | + +Each block reply is 1050 B on the wire (26 B of CoAP header and options plus +1024 B of data). When playing, blocks are written to ffplay as they arrive and +a full pipe slows the download down, but ffplay buffers several seconds ahead: +a 10 s clip finishes downloading after about 2.4 s while it plays. + +To see one block byte by byte: + +```bash +./coap-inspect.py get /image --block 0 +``` + +To test streaming end to end (generates its own media and server): + +```bash +./test-stream.sh +``` + +Limits: at most 65,536 blocks per file (about 67 MB, roughly 35 minutes of this +audio), download only, and no encryption. + ## Test ```bash cargo test ``` -Five unit tests cover the greeting, payload parsing (text + JSON) and query -decoding. +Unit tests cover the greeting, payload parsing (text + JSON), query decoding, +block selection, WAV and RGB565 validation, and the Block2/ETag/Size2 reply. ## Layout | Path | What it is | |---------------------|---------------------------------------------------------| | `src/main.rs` | The server: socket, read loop, route table, helpers. | -| `src/lib.rs` | Shared CoAP client used by the CLI and the bridge. | +| `src/lib.rs` | Shared CoAP client, including block-wise downloads. | | `src/bin/client.rs` | CLI client for smoke tests. | | `src/bin/bridge.rs` | HTTP -> CoAP proxy so a browser can reach the server. | +| `src/bin/stream.rs` | Streams `/audio` or `/image` into an ffplay window. | +| `media/` | Audio and image files the server streams (gitignored). | +| `coap-inspect.py` | Sends one request and shows both packets byte by byte. | +| `test-stream.sh` | End-to-end streaming test with its own server and media. | | `static/index.html` | CoAP reference page: routes, codes, copyable commands. | | `nodejs/` | The earlier Node.js version, kept for reference. | @@ -113,6 +212,7 @@ decoding. | `COAP_PORT` | `5683` | server, client, bridge | | `COAP_HOST` | `0.0.0.0` | server, client, bridge | | `HTTP_PORT` | `8080` | bridge | +| `MEDIA_DIR` | `media` | server | ## Design notes @@ -136,8 +236,8 @@ decoding. - **DTLS.** Plaintext CoAP has no encryption or authentication. Production uses `coaps://` on port 5684 — see `webrtc-dtls` or a gateway like `libcoap`. -- **Block-wise transfer** (RFC 7959) for payloads past ~1 KB; `coap-lite` ships - a `BlockHandler` for this. +- **Block-wise upload** (Block1, RFC 7959). Downloads already use Block2 (see + Streaming media); `coap-lite`'s `BlockHandler` can reassemble uploads. - **Observe** (RFC 7641) for server-pushed updates; `coap-lite` provides `Subject` and `create_notification`. - **Deduplication.** A CoAP server should track recent message IDs and replay diff --git a/coap-inspect.py b/coap-inspect.py index 2387137..5b7bf4a 100755 --- a/coap-inspect.py +++ b/coap-inspect.py @@ -4,6 +4,7 @@ ./coap-inspect.py get "/hello?name=Imran" ./coap-inspect.py post /hello Imran ./coap-inspect.py get /health + ./coap-inspect.py get /image --block 3 (one Block2 block of a stream) No CoAP library: the packet is built and decoded right here, so every byte on screen is accounted for. COAP_HOST / COAP_PORT retarget it (127.0.0.1:5683). @@ -19,13 +20,15 @@ METHODS = {"get": 1, "post": 2, "put": 3, "delete": 4} CODES = { "0.00": "Empty", "0.01": "GET", "0.02": "POST", "0.03": "PUT", "0.04": "DELETE", "2.01": "Created", "2.02": "Deleted", "2.03": "Valid", "2.04": "Changed", "2.05": "Content", - "4.00": "Bad Request", "4.04": "Not Found", "4.05": "Method Not Allowed", + "4.00": "Bad Request", "4.02": "Bad Option", "4.04": "Not Found", "4.05": "Method Not Allowed", "5.00": "Internal Server Error", } -OPTIONS = {3: "Uri-Host", 7: "Uri-Port", 11: "Uri-Path", 12: "Content-Format", 15: "Uri-Query", 17: "Accept"} -FORMATS = {0: "text/plain", 50: "application/json"} +OPTIONS = {3: "Uri-Host", 4: "ETag", 7: "Uri-Port", 11: "Uri-Path", 12: "Content-Format", 15: "Uri-Query", + 17: "Accept", 23: "Block2", 27: "Block1", 28: "Size2", 60: "Size1"} +FORMATS = {0: "text/plain", 42: "application/octet-stream", 50: "application/json"} TTY = sys.stdout.isatty() +DUMP_LIMIT = 128 # bytes shown in the hex and binary views def bold(text): @@ -46,7 +49,7 @@ def encode_option(delta, value): return bytes([(d << 4) | l]) + d_ext + l_ext + value -def build_request(method, path, query, payload): +def build_request(method, path, query, payload, block=None): token = os.urandom(4) header = bytes([(1 << 6) | (0 << 4) | len(token), METHODS[method]]) # ver 1, CON, TKL, code header += random.randrange(1 << 16).to_bytes(2, "big") + token # message id, token @@ -54,6 +57,9 @@ def build_request(method, path, query, payload): # Options must be in ascending number order; each stores the gap from the previous one. options = [(11, s.encode()) for s in path.split("/") if s] options += [(15, q.encode()) for q in query.split("&") if q] + if block is not None: # Block2: num << 4 | more << 3 | SZX, SZX 6 = 1024 B + value = block << 4 | 6 + options.append((23, value.to_bytes((value.bit_length() + 7) // 8, "big"))) body, last = b"", 0 for number, value in options: body += encode_option(number - last, value) @@ -90,8 +96,12 @@ def decode(pkt): if pkt[i] == 0xFF: body = pkt[i + 1:] rows.append((i, i + 1, "payload marker", "0xff -> the rest is the body")) - rows.append((i + 1, len(pkt), "payload", - f"{len(body)} bytes: {body.decode('utf-8', 'replace')}")) + try: + text = body.decode("utf-8") + shown = f"{len(body)} bytes: {text}" if text.isprintable() else None + except UnicodeDecodeError: + shown = None + rows.append((i + 1, len(pkt), "payload", shown or f"{len(body)} bytes of binary data")) break start = i @@ -112,6 +122,13 @@ def decode(pkt): if number in (12, 17): n = int.from_bytes(value, "big") if value else 0 shown = f"{n} = {FORMATS.get(n, 'unknown format')}" + elif number in (23, 27): + n = int.from_bytes(value, "big") if value else 0 + shown = f"num {n >> 4}, more {bool(n >> 3 & 1)}, size {1 << ((n & 7) + 4)}" + elif number in (28, 60): + shown = f"{int.from_bytes(value, 'big') if value else 0} B total" + elif number == 4: + shown = f"0x{value.hex()}" else: shown = f'"{value.decode("utf-8", "replace")}"' rows.append((start, i, OPTIONS.get(number, f"option {number}"), @@ -124,18 +141,22 @@ def show(title, pkt): print(bold(f"\n{title} ({len(pkt)} bytes)")) print(bold("\n hex + ascii")) - for off in range(0, len(pkt), 16): - chunk = pkt[off:off + 16] + for off in range(0, min(len(pkt), DUMP_LIMIT), 16): + chunk = pkt[off:min(off + 16, DUMP_LIMIT)] hexes = " ".join(f"{b:02x}" for b in chunk) text = "".join(chr(b) if 32 <= b < 127 else "." for b in chunk) print(f" {off:04x} {hexes:<47} |{text}|") + if len(pkt) > DUMP_LIMIT: + print(f" ... {len(pkt) - DUMP_LIMIT} more bytes not shown") print(bold("\n binary")) - for off in range(0, len(pkt), 4): - chunk = pkt[off:off + 4] + for off in range(0, min(len(pkt), DUMP_LIMIT), 4): + chunk = pkt[off:min(off + 4, DUMP_LIMIT)] bits = " ".join(f"{b:08b}" for b in chunk) text = "".join(chr(b) if 32 <= b < 127 else "." for b in chunk) print(f" {off:04x} {bits:<35} |{text}|") + if len(pkt) > DUMP_LIMIT: + print(f" ... {len(pkt) - DUMP_LIMIT} more bytes not shown") rows, meta = decode(pkt) print(bold("\n fields")) @@ -151,6 +172,15 @@ def show(title, pkt): def main(): args = sys.argv[1:] + block = None + if "--block" in args: + i = args.index("--block") + try: + block = int(args[i + 1]) + except (IndexError, ValueError): + print("--block needs a block number") + return 2 + del args[i:i + 2] if not args or args[0].lower() not in METHODS: print(__doc__.strip()) return 2 @@ -162,7 +192,7 @@ def main(): host = os.environ.get("COAP_HOST", "127.0.0.1") port = int(os.environ.get("COAP_PORT", "5683")) - request = build_request(method, path, query, payload) + request = build_request(method, path, query, payload, block) sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) # SOCK_DGRAM = UDP sock.settimeout(3) diff --git a/src/bin/stream.rs b/src/bin/stream.rs new file mode 100644 index 0000000..9c2d0fd --- /dev/null +++ b/src/bin/stream.rs @@ -0,0 +1,178 @@ +//! Stream a media file from the CoAP server and play it in an ffplay window. +//! +//! cargo run --release --bin stream -- image +//! cargo run --release --bin stream -- audio +//! cargo run --release --bin stream -- audio --no-play --out copy.wav +//! +//! The file arrives as 1024-byte Block2 blocks over UDP. Each block is written +//! to ffplay's stdin as soon as it lands; when the player's pipe is full the +//! write waits, so blocks are requested at playback speed instead of flooded. + +use std::io::{self, IsTerminal, Write}; +use std::process::Stdio; +use std::time::Duration; + +use coap_hello_udp::{BlockFetcher, Error, FetchOptions, TransferStats}; +use tokio::io::AsyncWriteExt; +use tokio::process::{Child, Command}; + +const USAGE: &str = "usage: stream audio|image [--out FILE] [--no-play] [--timeout-ms N]"; +const IMAGE_BYTES: u64 = 320 * 240 * 2; + +#[derive(Clone, Copy, PartialEq)] +enum Media { + Audio, + Image, +} + +impl Media { + fn path(self) -> &'static str { + match self { + Media::Audio => "/audio", + Media::Image => "/image", + } + } + + /// ffplay arguments for reading this media from stdin. + fn player_args(self) -> &'static [&'static str] { + match self { + Media::Audio => &["-autoexit", "-showmode", "waves", "-window_title", "CoAP audio", "-f", "wav"], + Media::Image => &[ + "-f", "rawvideo", "-pixel_format", "rgb565le", "-video_size", "320x240", + "-window_title", "CoAP image", + ], + } + } +} + +struct Args { + media: Media, + out: Option, + play: bool, + timeout: Duration, +} + +fn parse_args() -> Result { + let mut args = std::env::args().skip(1); + let media = match args.next().as_deref() { + Some("audio") => Media::Audio, + Some("image") => Media::Image, + _ => return Err(USAGE.into()), + }; + + let mut parsed = Args { media, out: None, play: true, timeout: FetchOptions::default().timeout }; + while let Some(arg) = args.next() { + match arg.as_str() { + "--out" => parsed.out = Some(args.next().ok_or(USAGE)?), + "--no-play" => parsed.play = false, + "--timeout-ms" => { + let ms = args.next().and_then(|v| v.parse().ok()).ok_or(USAGE)?; + parsed.timeout = Duration::from_millis(ms); + } + _ => return Err(USAGE.into()), + } + } + Ok(parsed) +} + +#[tokio::main] +async fn main() -> Result<(), Error> { + let args = match parse_args() { + Ok(args) => args, + Err(usage) => { + eprintln!("{usage}"); + std::process::exit(2); + } + }; + let host = std::env::var("COAP_HOST").unwrap_or_else(|_| "127.0.0.1".into()); + let port = std::env::var("COAP_PORT").unwrap_or_else(|_| "5683".into()); + let path = args.media.path(); + + let mut player = if args.play { Some(spawn_player(args.media)?) } else { None }; + let mut player_stdin = player.as_mut().and_then(|p| p.stdin.take()); + let mut out = match &args.out { + Some(file) => Some(io::BufWriter::new(std::fs::File::create(file)?)), + None => None, + }; + + let options = FetchOptions { timeout: args.timeout, ..FetchOptions::default() }; + let mut fetcher = BlockFetcher::connect(&host, &port, path, options).await?; + eprintln!("streaming coap://{host}:{port}{path}"); + + let show_progress = io::stderr().is_terminal(); + let mut stopped_by_player = false; + + while let Some(block) = fetcher.next_block().await? { + if let Some(file) = out.as_mut() { + file.write_all(&block)?; + } + if let Some(stdin) = player_stdin.as_mut() + && stdin.write_all(&block).await.is_err() + { + // The player window was closed. + stopped_by_player = true; + break; + } + if show_progress { + let stats = fetcher.stats(); + let total = stats.total.map(|t| format!(" / {t}")).unwrap_or_default(); + eprint!("\r {}{total} B, {} blocks, {} retransmits", stats.bytes, stats.blocks, stats.retransmits); + } + } + + if let Some(mut file) = out { + file.flush()?; + } + let stats = fetcher.stats(); + if show_progress { + eprintln!(); + } + print_summary(&stats, stopped_by_player); + + if args.media == Media::Image + && let Some(total) = stats.total + && total != IMAGE_BYTES + { + eprintln!("warning: server sent {total} B, a 320x240 RGB565 image is {IMAGE_BYTES} B"); + } + + // Closing stdin tells ffplay the stream has ended. + drop(player_stdin); + if let Some(player) = player.as_mut() { + if !stopped_by_player && args.media == Media::Image { + eprintln!("close the image window to exit"); + } + player.wait().await?; + } + Ok(()) +} + +fn spawn_player(media: Media) -> Result { + Command::new("ffplay") + .args(["-hide_banner", "-loglevel", "error"]) + .args(media.player_args()) + .args(["-i", "pipe:0"]) + .stdin(Stdio::piped()) + .kill_on_drop(true) + .spawn() + .map_err(|e| match e.kind() { + io::ErrorKind::NotFound => { + "ffplay not found: install ffmpeg, or use --no-play --out FILE".to_string() + } + _ => format!("cannot start ffplay: {e}"), + }) +} + +fn print_summary(stats: &TransferStats, stopped_by_player: bool) { + let secs = stats.elapsed.as_secs_f64(); + let rate = if secs > 0.0 { stats.bytes as f64 / 1024.0 / secs } else { 0.0 }; + let etag = match &stats.etag { + Some(tag) => tag.iter().map(|b| format!("{b:02x}")).collect(), + None => "none".to_string(), + }; + let what = if stopped_by_player { "stopped by player after" } else { "received" }; + eprintln!( + "{what} {} B in {} blocks, {secs:.2} s ({rate:.0} KiB/s), {} retransmits, ETag {etag}", + stats.bytes, stats.blocks, stats.retransmits + ); +} diff --git a/src/lib.rs b/src/lib.rs index ba078e0..eaa9f0f 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -1,11 +1,14 @@ -//! Shared CoAP client used by the CLI (`src/bin/client.rs`) and the HTTP -//! bridge (`src/bin/bridge.rs`). +//! Shared CoAP client used by the CLI (`src/bin/client.rs`), the HTTP bridge +//! (`src/bin/bridge.rs`) and the media streamer (`src/bin/stream.rs`). -use std::time::{Duration, SystemTime, UNIX_EPOCH}; +use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; +use coap_lite::block_handler::BlockValue; use coap_lite::{CoapOption, MessageClass, MessageType, Packet, RequestType}; use tokio::net::UdpSocket; +pub type Error = Box; + /// What a CoAP server sent back. pub struct CoapReply { /// Response code in CoAP's `class.detail` form, e.g. `"2.05"`. @@ -46,24 +49,24 @@ pub fn parse_method(name: &str) -> Result { } } -/// Send one confirmable CoAP request and wait for the reply. +/// Build a confirmable request. /// /// A CoAP URI is carried as repeated options rather than one string: each path /// segment becomes its own Uri-Path option, each `k=v` its own Uri-Query. -pub async fn request( - host: &str, - port: &str, +fn request_packet( method: RequestType, path: &str, query: Option<&str>, payload: &[u8], -) -> Result> { + message_id: u16, + token: &[u8], +) -> Packet { let mut packet = Packet::new(); packet.header.set_version(1); packet.header.set_type(MessageType::Confirmable); packet.header.code = MessageClass::Request(method); - packet.header.message_id = nonce() as u16; - packet.set_token(nonce().to_be_bytes()[..4].to_vec()); + packet.header.message_id = message_id; + packet.set_token(token.to_vec()); for segment in path.split('/').filter(|s| !s.is_empty()) { packet.add_option(CoapOption::UriPath, segment.as_bytes().to_vec()); @@ -74,6 +77,25 @@ pub async fn request( if !payload.is_empty() { packet.payload = payload.to_vec(); } + packet +} + +/// A response code in `class.detail` form, e.g. `"2.05"`. +fn code_string(packet: &Packet) -> String { + let raw: u8 = packet.header.code.into(); + format!("{}.{:02}", raw >> 5, raw & 0x1f) +} + +/// Send one confirmable CoAP request and wait for the reply. +pub async fn request( + host: &str, + port: &str, + method: RequestType, + path: &str, + query: Option<&str>, + payload: &[u8], +) -> Result { + let packet = request_packet(method, path, query, payload, nonce() as u16, &nonce().to_be_bytes()); let socket = UdpSocket::bind("0.0.0.0:0").await?; socket.connect(format!("{host}:{port}")).await?; @@ -88,14 +110,188 @@ pub async fn request( }; let response = Packet::from_bytes(&buf[..len])?; - let raw: u8 = response.header.code.into(); Ok(CoapReply { - code: format!("{}.{:02}", raw >> 5, raw & 0x1f), + code: code_string(&response), payload: String::from_utf8_lossy(&response.payload).into_owned(), }) } +/// Settings for a block-wise download. Defaults follow RFC 7252 §4.8. +#[derive(Clone, Copy)] +pub struct FetchOptions { + /// How long to wait for a reply before retransmitting; doubles each retry. + pub timeout: Duration, + pub max_retransmit: u32, + /// Block size to ask for. The server may answer with a smaller one. + pub block_size: usize, +} + +impl Default for FetchOptions { + fn default() -> Self { + Self { timeout: Duration::from_secs(2), max_retransmit: 4, block_size: 1024 } + } +} + +/// How a block-wise download went so far. +#[derive(Clone, Default)] +pub struct TransferStats { + pub blocks: usize, + pub bytes: usize, + pub retransmits: u32, + pub elapsed: Duration, + /// Total size the server announced in Size2, if any. + pub total: Option, + pub etag: Option>, +} + +/// Downloads one resource with Block2 (RFC 7959), a block at a time. +/// +/// Blocks are requested one after another on a single socket (NSTART = 1). +/// The caller pulls them with [`BlockFetcher::next_block`], so a slow consumer +/// — a media player whose pipe is full — sets the pace of the whole transfer. +pub struct BlockFetcher { + socket: UdpSocket, + path: String, + options: FetchOptions, + token: [u8; 4], + message_id: u16, + next_num: usize, + block_size: usize, + done: bool, + started: Instant, + stats: TransferStats, +} + +impl BlockFetcher { + pub async fn connect( + host: &str, + port: &str, + path: &str, + options: FetchOptions, + ) -> Result { + let socket = UdpSocket::bind("0.0.0.0:0").await?; + socket.connect(format!("{host}:{port}")).await?; + Ok(Self { + socket, + path: path.to_string(), + options, + token: nonce().to_be_bytes(), + message_id: nonce() as u16, + next_num: 0, + block_size: options.block_size, + done: false, + started: Instant::now(), + stats: TransferStats::default(), + }) + } + + pub fn stats(&self) -> TransferStats { + TransferStats { elapsed: self.started.elapsed(), ..self.stats.clone() } + } + + /// Fetch the next block. `Ok(None)` once the last block has arrived. + pub async fn next_block(&mut self) -> Result>, Error> { + if self.done { + return Ok(None); + } + + let requested_size = self.block_size; + let block2 = BlockValue::new(self.next_num, false, requested_size) + .map_err(|e| format!("cannot request block {}: {e:?}", self.next_num))?; + + self.message_id = self.message_id.wrapping_add(1); + let mut packet = + request_packet(RequestType::Get, &self.path, None, &[], self.message_id, &self.token); + packet.add_option_as(CoapOption::Block2, block2); + let response = self.exchange(&packet.to_bytes()?).await?; + + let code = code_string(&response); + if code != "2.05" { + return Err(format!("{code} {}", String::from_utf8_lossy(&response.payload)).into()); + } + + let etag = response.get_first_option(CoapOption::ETag).cloned(); + if self.stats.blocks == 0 { + self.stats.etag = etag; + self.stats.total = response + .get_first_option(CoapOption::Size2) + .map(|v| v.iter().fold(0u64, |n, &b| n << 8 | u64::from(b))); + } else if etag != self.stats.etag { + return Err("the resource changed during the transfer (ETag differs)".into()); + } + + match response.get_first_option_as::(CoapOption::Block2) { + // The whole resource fitted in one message. + None if self.next_num == 0 => self.done = true, + None => return Err("server stopped sending Block2 mid-transfer".into()), + Some(Err(_)) => return Err("malformed Block2 option in the response".into()), + Some(Ok(block)) => { + let num = usize::from(block.num); + // The server may shrink the block size and renumber; the byte + // offset must still be the one we asked for. + if num * block.size() != self.next_num * requested_size { + return Err(format!( + "asked for byte {} but got block {num} of size {}", + self.next_num * requested_size, + block.size() + ) + .into()); + } + if block.more && response.payload.len() != block.size() { + return Err(format!( + "block {num} is {} B, expected {}", + response.payload.len(), + block.size() + ) + .into()); + } + self.block_size = block.size(); + self.next_num = num + 1; + self.done = !block.more; + } + } + + self.stats.blocks += 1; + self.stats.bytes += response.payload.len(); + Ok(Some(response.payload)) + } + + /// Send a confirmable request and wait for the reply carrying the same + /// message ID and token, retransmitting with exponential back-off. + async fn exchange(&mut self, request: &[u8]) -> Result { + let mut timeout = self.options.timeout; + let mut buf = [0u8; 1152]; + + for attempt in 0..=self.options.max_retransmit { + if attempt > 0 { + self.stats.retransmits += 1; + } + self.socket.send(request).await?; + + let deadline = tokio::time::Instant::now() + timeout; + loop { + let len = match tokio::time::timeout_at(deadline, self.socket.recv(&mut buf)).await { + Ok(result) => result?, + Err(_) => break, + }; + // Late replies to an earlier block, or anything unparsable, are not ours. + let Ok(reply) = Packet::from_bytes(&buf[..len]) else { continue }; + if reply.header.message_id == self.message_id && reply.get_token() == self.token { + return Ok(reply); + } + } + timeout *= 2; + } + + Err(format!( + "no reply for block {} after {} retransmissions", + self.next_num, self.options.max_retransmit + ) + .into()) + } +} + fn nonce() -> u32 { SystemTime::now() .duration_since(UNIX_EPOCH) diff --git a/src/main.rs b/src/main.rs index b344299..a4b62ca 100644 --- a/src/main.rs +++ b/src/main.rs @@ -4,10 +4,15 @@ //! and query strings, but each request/response is a single datagram carrying a //! compact binary header. `coap-lite` handles that binary codec; the UDP socket //! and the routing are right here so nothing is hidden. +//! +//! It also streams two static media files with block-wise transfer (RFC 7959): +//! `GET /audio` (16 kHz mono 16-bit WAV) and `GET /image` (320x240 RGB565). use std::net::SocketAddr; +use std::path::Path; use std::time::Instant; +use coap_lite::block_handler::BlockValue; use coap_lite::{ CoapOption, CoapRequest, CoapResponse, ContentFormat, Packet, RequestType, ResponseType, @@ -19,19 +24,83 @@ use tokio::net::UdpSocket; /// transfer (RFC 7959). const MAX_DATAGRAM: usize = 1152; +/// Largest block we serve. 1024 B (SZX 6) is the biggest RFC 7959 block size +/// that still fits one datagram once the header and options are added. +const MAX_BLOCK: usize = 1024; + +/// coap-lite stores the block number as a u16, capping a resource at 65,536 +/// blocks. +const MAX_MEDIA_BYTES: usize = (u16::MAX as usize + 1) * MAX_BLOCK; + +const IMAGE_WIDTH: usize = 320; +const IMAGE_HEIGHT: usize = 240; +const IMAGE_BYTES: usize = IMAGE_WIDTH * IMAGE_HEIGHT * 2; + +/// Everything a request handler can read. Built once, before the receive loop. +struct State { + started: Instant, + media: Media, +} + +/// Static media, loaded into memory at startup so serving a block is a slice +/// of a buffer, never file I/O on the receive loop. +struct Media { + audio: Result, + image: Result, +} + +struct Resource { + bytes: Vec, + etag: [u8; 8], +} + +impl Resource { + fn load(path: &Path, validate: fn(&[u8]) -> Result<(), String>) -> Result { + let bytes = std::fs::read(path).map_err(|e| format!("{}: {e}", path.display()))?; + if bytes.len() > MAX_MEDIA_BYTES { + return Err(format!( + "{}: {} B exceeds the {MAX_MEDIA_BYTES} B block-wise limit", + path.display(), + bytes.len() + )); + } + validate(&bytes).map_err(|e| format!("{}: {e}", path.display()))?; + Ok(Self { etag: etag(&bytes), bytes }) + } +} + #[tokio::main] async fn main() -> std::io::Result<()> { let host = std::env::var("COAP_HOST").unwrap_or_else(|_| "0.0.0.0".into()); let port = std::env::var("COAP_PORT").unwrap_or_else(|_| "5683".into()); + let media_dir = std::env::var("MEDIA_DIR").unwrap_or_else(|_| "media".into()); let addr = format!("{host}:{port}"); + let media_dir = Path::new(&media_dir); + let state = State { + started: Instant::now(), + media: Media { + audio: Resource::load(&media_dir.join("audio.wav"), validate_wav), + image: Resource::load(&media_dir.join("image.rgb565"), validate_image), + }, + }; + let socket = UdpSocket::bind(&addr).await?; - let started = Instant::now(); println!("CoAP server (UDP) listening on coap://{addr}"); println!(" GET /hello?name=Imran"); println!(" POST /hello (payload: Imran)"); println!(" GET /health"); + for (path, resource) in [("/audio", &state.media.audio), ("/image", &state.media.image)] { + match resource { + Ok(r) => println!( + " GET {path} ({} B in {} blocks)", + r.bytes.len(), + r.bytes.len().div_ceil(MAX_BLOCK) + ), + Err(e) => eprintln!(" GET {path} unavailable: {e}"), + } + } let mut buf = [0u8; MAX_DATAGRAM]; @@ -50,7 +119,7 @@ async fn main() -> std::io::Result<()> { // Handling is pure CPU work with no awaits, so doing it inline // keeps the hot path allocation-light and ordered. Spawn a task // here instead once a handler needs to do real I/O. - if let Some(reply) = handle(&buf[..len], peer, started) + if let Some(reply) = handle(&buf[..len], peer, &state) && let Err(e) = socket.send_to(&reply, peer).await { eprintln!("send error to {peer}: {e}"); @@ -62,7 +131,7 @@ async fn main() -> std::io::Result<()> { /// Decode one datagram, route it, and encode the reply. /// Returns `None` when no reply should be sent (unparsable or non-request). -fn handle(datagram: &[u8], peer: SocketAddr, started: Instant) -> Option> { +fn handle(datagram: &[u8], peer: SocketAddr, state: &State) -> Option> { let packet = match Packet::from_bytes(datagram) { Ok(p) => p, Err(e) => { @@ -78,9 +147,13 @@ fn handle(datagram: &[u8], peer: SocketAddr, started: Instant) -> Option let path = format!("/{}", request.get_path()); let method = *request.get_method(); - println!("{method:?} {}{} from {peer}", path, query_string(&request.message)); - route(method, &path, &mut request, &mut response, started); + // Media streams log once per transfer in `serve_media`, not once per block. + if !matches!((method, path.as_str()), (RequestType::Get, "/audio" | "/image")) { + println!("{method:?} {}{} from {peer}", path, query_string(&request.message)); + } + + route(method, &path, &mut request, &mut response, state); match response.message.to_bytes() { Ok(bytes) => Some(bytes), @@ -96,7 +169,7 @@ fn route( path: &str, request: &mut CoapRequest, response: &mut CoapResponse, - started: Instant, + state: &State, ) { match (method, path) { (RequestType::Get, "/hello") => { @@ -127,13 +200,16 @@ fn route( (RequestType::Get, "/health") => { let body = format!( r#"{{"status":"ok","uptime":{}}}"#, - started.elapsed().as_secs() + state.started.elapsed().as_secs() ); reply(response, ResponseType::Content, ContentFormat::ApplicationJSON, body); } + (RequestType::Get, "/audio") => serve_media(request, response, &state.media.audio, path), + (RequestType::Get, "/image") => serve_media(request, response, &state.media.image, path), + // Path exists but the method doesn't: 4.05, per RFC 7252. - (_, "/hello") | (_, "/health") => reply( + (_, "/hello" | "/health" | "/audio" | "/image") => reply( response, ResponseType::MethodNotAllowed, ContentFormat::TextPlain, @@ -149,6 +225,179 @@ fn route( } } +/// Serve one block of a static resource (RFC 7959 Block2). +/// +/// Stateless: every request names the block it wants, so nothing is cached per +/// client, and a lost or repeated request simply gets the same block again. +fn serve_media( + request: &CoapRequest, + response: &mut CoapResponse, + resource: &Result, + path: &str, +) { + let resource = match resource { + Ok(r) => r, + Err(e) => { + return reply( + response, + ResponseType::NotFound, + ContentFormat::TextPlain, + format!("Not Found: {path} is unavailable ({e})"), + ); + } + }; + + let requested = match request.message.get_first_option_as::(CoapOption::Block2) { + None => None, + Some(Ok(b)) => Some((usize::from(b.num), b.size())), + Some(Err(_)) => { + return reply( + response, + ResponseType::BadOption, + ContentFormat::TextPlain, + "Bad Option: malformed Block2".into(), + ); + } + }; + + let len = resource.bytes.len(); + let block = match select_block(requested, len) { + Ok(b) => b, + Err(e) => { + return reply(response, ResponseType::BadOption, ContentFormat::TextPlain, format!("Bad Option: {e}")); + } + }; + + // `select_block` only yields sizes and numbers a BlockValue can hold. + let Ok(block2) = BlockValue::new(block.num, block.more, block.size) else { + return reply( + response, + ResponseType::InternalServerError, + ContentFormat::TextPlain, + format!("cannot encode block {} of size {}", block.num, block.size), + ); + }; + + let peer = request.source.map(|p| p.to_string()).unwrap_or_default(); + if block.num == 0 { + println!( + "GET {path} from {peer}: {len} B in {} blocks of {} B", + len.div_ceil(block.size), + block.size + ); + } else if !block.more { + println!("GET {path} from {peer}: sent last block {}", block.num); + } + + response.set_status(ResponseType::Content); + let message = &mut response.message; + message.set_content_format(ContentFormat::ApplicationOctetStream); + message.add_option(CoapOption::ETag, resource.etag.to_vec()); + message.add_option_as(CoapOption::Block2, block2); + if block.num == 0 { + // Total size up front, so the client can show progress (RFC 7959 §4). + message.add_option(CoapOption::Size2, uint_option(len as u32)); + } + message.payload = resource.bytes[block.start..block.end].to_vec(); +} + +/// The byte range one Block2 request gets. +#[derive(Debug, PartialEq)] +struct Block { + num: usize, + size: usize, + start: usize, + end: usize, + more: bool, +} + +/// Pick the block for a request's `(num, size)`; no Block2 means block 0. +fn select_block(requested: Option<(usize, usize)>, len: usize) -> Result { + let (num, size) = requested.unwrap_or((0, MAX_BLOCK)); + + // A server may answer with a smaller block than asked for. Renumber so the + // client still gets the byte offset it asked for (RFC 7959 §2.4). + let offset = num * size; + let size = size.min(MAX_BLOCK); + let num = offset / size; + let start = num * size; + + if start >= len { + return Err(format!("block {num} of size {size} is past the end of a {len} B resource")); + } + let end = (start + size).min(len); + Ok(Block { num, size, start, end, more: end < len }) +} + +/// Accept only PCM WAV at 16 kHz, mono, 16-bit. Walks the RIFF chunks instead +/// of assuming a 44-byte header, since encoders like ffmpeg add a LIST chunk. +fn validate_wav(bytes: &[u8]) -> Result<(), String> { + if bytes.len() < 12 || &bytes[0..4] != b"RIFF" || &bytes[8..12] != b"WAVE" { + return Err("not a RIFF/WAVE file".into()); + } + + let mut format = None; + let mut has_data = false; + let mut at = 12; + while at + 8 <= bytes.len() { + let id = &bytes[at..at + 4]; + let size = u32::from_le_bytes([bytes[at + 4], bytes[at + 5], bytes[at + 6], bytes[at + 7]]) as usize; + let body = &bytes[at + 8..(at + 8 + size).min(bytes.len())]; + + match id { + b"fmt " if body.len() >= 16 => { + let u16_at = |i: usize| u16::from_le_bytes([body[i], body[i + 1]]); + let rate = u32::from_le_bytes([body[4], body[5], body[6], body[7]]); + format = Some((u16_at(0), u16_at(2), rate, u16_at(14))); + } + b"data" => has_data = true, + _ => {} + } + + // Chunks are padded to an even length. + at += 8 + size + size % 2; + } + + match format { + None => Err("no fmt chunk".into()), + Some((1, 1, 16_000, 16)) if has_data => Ok(()), + Some((1, 1, 16_000, 16)) => Err("no data chunk".into()), + Some((codec, channels, rate, bits)) => Err(format!( + "need PCM mono 16 kHz 16-bit, got format {codec}, {channels} channel(s), {rate} Hz, {bits}-bit" + )), + } +} + +/// Raw RGB565 has no header, so the exact size is the only possible check. +fn validate_image(bytes: &[u8]) -> Result<(), String> { + if bytes.len() == IMAGE_BYTES { + Ok(()) + } else { + Err(format!( + "need exactly {IMAGE_BYTES} B for {IMAGE_WIDTH}x{IMAGE_HEIGHT} RGB565, got {}", + bytes.len() + )) + } +} + +/// FNV-1a 64-bit hash of the content, sent as the ETag so a client can check +/// that every block came from the same version of the file. +fn etag(bytes: &[u8]) -> [u8; 8] { + let mut hash: u64 = 0xcbf2_9ce4_8422_2325; + for &b in bytes { + hash ^= u64::from(b); + hash = hash.wrapping_mul(0x0000_0100_0000_01b3); + } + hash.to_be_bytes() +} + +/// Encode a CoAP uint option: big-endian with leading zero bytes dropped. +fn uint_option(value: u32) -> Vec { + let bytes = value.to_be_bytes(); + let skip = bytes.iter().take_while(|&&b| b == 0).count(); + bytes[skip..].to_vec() +} + fn greet(name: Option<&str>) -> String { let name = name.map(str::trim).filter(|n| !n.is_empty()).unwrap_or("World"); format!("Hello, {name}!") @@ -254,6 +503,7 @@ fn percent_decode(input: &str) -> String { #[cfg(test)] mod tests { use super::*; + use coap_lite::{MessageClass, MessageType}; #[test] fn greets_world_when_no_name() { @@ -269,7 +519,6 @@ mod tests { } #[test] - fn reads_name_from_text_and_json_payloads() { assert_eq!(name_from_payload(b"Imran").as_deref(), Some("Imran")); assert_eq!( @@ -306,4 +555,164 @@ mod tests { assert_eq!(query_param(&packet, "missing"), None); assert_eq!(query_string(&packet), "?name=Md%20Imran&lang=en"); } + + // 10 s of 16 kHz mono 16-bit audio as written by ffmpeg, header included. + const AUDIO_BYTES: usize = 320_078; + + #[test] + fn serves_block_zero_when_no_block2_is_sent() { + assert_eq!( + select_block(None, IMAGE_BYTES), + Ok(Block { num: 0, size: 1024, start: 0, end: 1024, more: true }) + ); + } + + #[test] + fn serves_a_middle_block() { + assert_eq!( + select_block(Some((5, 1024)), AUDIO_BYTES), + Ok(Block { num: 5, size: 1024, start: 5120, end: 6144, more: true }) + ); + } + + #[test] + fn last_audio_block_is_partial() { + assert_eq!( + select_block(Some((312, 1024)), AUDIO_BYTES), + Ok(Block { num: 312, size: 1024, start: 319_488, end: AUDIO_BYTES, more: false }) + ); + assert!(select_block(Some((313, 1024)), AUDIO_BYTES).is_err()); + } + + #[test] + fn image_ends_exactly_on_a_block_boundary() { + assert_eq!( + select_block(Some((149, 1024)), IMAGE_BYTES), + Ok(Block { num: 149, size: 1024, start: 152_576, end: IMAGE_BYTES, more: false }) + ); + assert!(select_block(Some((150, 1024)), IMAGE_BYTES).is_err()); + } + + #[test] + fn oversized_block_request_is_renumbered() { + // Block 1 of 2048 B starts at byte 2048, which is block 2 of 1024 B. + assert_eq!( + select_block(Some((1, 2048)), IMAGE_BYTES), + Ok(Block { num: 2, size: 1024, start: 2048, end: 3072, more: true }) + ); + } + + #[test] + fn smaller_block_size_is_honoured() { + assert_eq!( + select_block(Some((3, 256)), IMAGE_BYTES), + Ok(Block { num: 3, size: 256, start: 768, end: 1024, more: true }) + ); + } + + fn wav(channels: u16, rate: u32, bits: u16, list_chunk: bool) -> Vec { + let block_align = channels * bits / 8; + let mut body = b"WAVE".to_vec(); + body.extend(b"fmt "); + body.extend(16u32.to_le_bytes()); + body.extend(1u16.to_le_bytes()); + body.extend(channels.to_le_bytes()); + body.extend(rate.to_le_bytes()); + body.extend((rate * u32::from(block_align)).to_le_bytes()); + body.extend(block_align.to_le_bytes()); + body.extend(bits.to_le_bytes()); + if list_chunk { + // Odd-sized, so the parser has to skip the padding byte. + body.extend(b"LIST"); + body.extend(3u32.to_le_bytes()); + body.extend(b"abc\0"); + } + body.extend(b"data"); + body.extend(4u32.to_le_bytes()); + body.extend([0u8; 4]); + + let mut file = b"RIFF".to_vec(); + file.extend((body.len() as u32).to_le_bytes()); + file.extend(body); + file + } + + #[test] + fn accepts_16khz_mono_16bit_wav() { + assert_eq!(validate_wav(&wav(1, 16_000, 16, false)), Ok(())); + assert_eq!(validate_wav(&wav(1, 16_000, 16, true)), Ok(())); + } + + #[test] + fn rejects_other_wav_formats() { + assert!(validate_wav(&wav(2, 16_000, 16, false)).is_err()); + assert!(validate_wav(&wav(1, 44_100, 16, false)).is_err()); + assert!(validate_wav(&wav(1, 16_000, 8, false)).is_err()); + assert!(validate_wav(b"definitely not a wav file").is_err()); + } + + #[test] + fn image_must_be_exactly_320x240_rgb565() { + assert_eq!(validate_image(&vec![0; 153_600]), Ok(())); + assert!(validate_image(&vec![0; 153_599]).is_err()); + } + + #[test] + fn etag_is_deterministic_and_content_sensitive() { + assert_eq!(etag(b"frame"), etag(b"frame")); + assert_ne!(etag(b"frame"), etag(b"framf")); + } + + #[test] + fn encodes_uint_options_without_leading_zeros() { + assert_eq!(uint_option(153_600), vec![0x02, 0x58, 0x00]); + assert_eq!(uint_option(0), Vec::::new()); + } + + fn media_request(block: Option) -> CoapRequest { + let mut packet = Packet::new(); + packet.header.set_type(MessageType::Confirmable); + packet.header.code = MessageClass::Request(RequestType::Get); + if let Some(block) = block { + packet.add_option_as(CoapOption::Block2, block); + } + CoapRequest::from_packet(packet, "127.0.0.1:9".parse().unwrap()) + } + + #[test] + fn media_blocks_carry_block2_etag_and_size2() { + let bytes = vec![7u8; 3000]; + let resource = Ok(Resource { etag: etag(&bytes), bytes }); + + let request = media_request(None); + let mut first = CoapResponse::new(&request.message).unwrap(); + serve_media(&request, &mut first, &resource, "/audio"); + let block = first.message.get_first_option_as::(CoapOption::Block2).unwrap().unwrap(); + assert_eq!(u8::from(first.message.header.code), 0x45); // 2.05 + assert_eq!((block.num, block.more, block.size()), (0, true, 1024)); + assert_eq!(first.message.get_first_option(CoapOption::Size2), Some(&vec![0x0b, 0xb8])); + assert_eq!(first.message.payload.len(), 1024); + + let request = media_request(Some(BlockValue::new(2, false, 1024).unwrap())); + let mut last = CoapResponse::new(&request.message).unwrap(); + serve_media(&request, &mut last, &resource, "/audio"); + let block = last.message.get_first_option_as::(CoapOption::Block2).unwrap().unwrap(); + assert_eq!((block.num, block.more), (2, false)); + assert_eq!(last.message.payload.len(), 3000 - 2048); + assert_eq!( + last.message.get_first_option(CoapOption::ETag), + first.message.get_first_option(CoapOption::ETag) + ); + assert!(last.message.get_option(CoapOption::Size2).is_none()); + } + + #[test] + fn block_past_the_end_is_bad_option() { + let bytes = vec![0u8; 1024]; + let resource = Ok(Resource { etag: etag(&bytes), bytes }); + let request = media_request(Some(BlockValue::new(1, false, 1024).unwrap())); + let mut response = CoapResponse::new(&request.message).unwrap(); + serve_media(&request, &mut response, &resource, "/image"); + assert_eq!(u8::from(response.message.header.code), 0x82); // 4.02 + } } diff --git a/test-stream.sh b/test-stream.sh new file mode 100755 index 0000000..64e0aaf --- /dev/null +++ b/test-stream.sh @@ -0,0 +1,82 @@ +#!/usr/bin/env bash +# End-to-end test of block-wise media streaming over CoAP/UDP. Self-contained: +# generates test media, starts its own server, streams both files back and +# compares them byte for byte. +# ./test-stream.sh (needs: cargo build --release, ffmpeg, python3) +# COAP_PORT picks the test server's port (default 5699). + +set -u +cd "$(dirname "$0")" + +PORT=${COAP_PORT:-5699} +DIR=$(mktemp -d) +pass=0; fail=0 +server_pid= + +cleanup() { [[ -n $server_pid ]] && kill "$server_pid" 2>/dev/null; rm -rf "$DIR"; } +trap cleanup EXIT + +ok() { printf ' ok %s\n' "$1"; ((pass++)); } +bad() { printf ' FAIL %s\n' "$1"; [[ -n ${2:-} ]] && printf '%s\n' "$2" | sed 's/^/ /'; ((fail++)); } + +for bin in coap-hello-udp stream client; do + [[ -x target/release/$bin ]] || { echo "build first: cargo build --release"; exit 1; } +done +command -v ffmpeg >/dev/null || { echo "ffmpeg is needed to generate test media"; exit 1; } + +ffmpeg -hide_banner -loglevel error -f lavfi -i "sine=frequency=440:duration=10" \ + -ac 1 -ar 16000 -c:a pcm_s16le "$DIR/audio.wav" +ffmpeg -hide_banner -loglevel error -f lavfi -i "testsrc=size=320x240" \ + -frames:v 1 -f rawvideo -pix_fmt rgb565le "$DIR/image.rgb565" + +MEDIA_DIR=$DIR COAP_PORT=$PORT target/release/coap-hello-udp >"$DIR/server.log" 2>&1 & +server_pid=$! +for _ in $(seq 50); do + ss -lun | grep -q ":$PORT " && break + sleep 0.1 +done +ss -lun | grep -q ":$PORT " || { echo "server did not start:"; cat "$DIR/server.log"; exit 1; } + +echo "block-wise streaming tests -> udp://127.0.0.1:$PORT" + +check_stream() { # check_stream + local media=$1 src=$2 out="$DIR/received.$1" log + local size; size=$(stat -c %s "$src") + local blocks=$(( (size + 1023) / 1024 )) + + if ! log=$(COAP_PORT=$PORT target/release/stream "$media" --no-play --out "$out" 2>&1); then + bad "stream $media" "$log"; return + fi + if cmp -s "$src" "$out"; then + ok "stream $media: $size B received byte-identical" + else + bad "stream $media: received file differs from the source" + fi + if grep -q "received $size B in $blocks blocks" <<<"$log"; then + ok "stream $media: $blocks blocks, $(grep -o '[0-9]* retransmits' <<<"$log")" + else + bad "stream $media: expected $size B in $blocks blocks" "$log" + fi +} + +check_stream image "$DIR/image.rgb565" +check_stream audio "$DIR/audio.wav" + +reply=$(COAP_PORT=$PORT python3 coap-inspect.py get /image --block 150 2>&1) +if grep -q "4.02 Bad Option" <<<"$reply"; then + ok "block 150 of a 150-block image -> 4.02 Bad Option" +else + bad "block past the end -> 4.02" "$(tail -4 <<<"$reply")" +fi + +reply=$(COAP_PORT=$PORT target/release/client post /image x 2>&1) +if [[ $reply == "<- 4.05 "* ]]; then ok "POST /image -> 4.05"; else bad "POST /image -> 4.05" "$reply"; fi + +if out=$(COAP_PORT=$PORT ./test-coap.sh 2>&1); then + ok "test-coap.sh on the same server: $(tail -1 <<<"$out")" +else + bad "test-coap.sh on the same server" "$(grep FAIL <<<"$out")" +fi + +echo "$pass passed, $fail failed" +[[ $fail -eq 0 ]]