bask 0.2.0

Build Tasks: an async task-queue pipeline engine with pluggable IO, columnar formats, and predefined tasks
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120

/*
 * SPDX-FileCopyrightText: 2026 Wavelens GmbH <info@wavelens.io>
 *
 * SPDX-License-Identifier: MIT OR Apache-2.0
 */
#![cfg(feature = "download")]

use std::io::{Read as _, Write as _};
use std::net::TcpListener;
use std::path::{Path, PathBuf};

use bask::io::{
    Bytes, HttpSource, Read, SinkRegistry, SinkWorker, Source, SourceRegistry, SourceWorker,
};
use bask::prelude::*;

/// A throwaway HTTP/1.1 server that answers every request with `body`.
fn serve(body: &'static [u8]) -> String {
    let listener = TcpListener::bind("127.0.0.1:0").unwrap();
    let url = format!("http://{}/file", listener.local_addr().unwrap());
    std::thread::spawn(move || {
        for stream in listener.incoming() {
            let Ok(mut stream) = stream else { break };
            let mut buf = [0u8; 1024];
            let _ = stream.read(&mut buf);
            let header = format!(
                "HTTP/1.1 200 OK\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
                body.len()
            );
            let _ = stream.write_all(header.as_bytes());
            let _ = stream.write_all(body);
            let _ = stream.flush();
        }
    });
    url
}

fn scratch(name: &str) -> PathBuf {
    let path = std::env::temp_dir().join(name);
    let _ = std::fs::remove_dir_all(&path);
    path
}

fn find_one_file(dir: &Path) -> PathBuf {
    fn walk(dir: &Path) -> Option<PathBuf> {
        let mut entries: Vec<PathBuf> = std::fs::read_dir(dir)
            .ok()?
            .filter_map(|e| e.ok().map(|e| e.path()))
            .collect();
        entries.sort();
        for path in entries {
            if path.is_dir() {
                if let Some(found) = walk(&path) {
                    return Some(found);
                }
            } else {
                return Some(path);
            }
        }
        None
    }
    walk(dir).unwrap_or_else(|| panic!("no file written under {dir:?}"))
}

#[tokio::test]
async fn http_source_fetches_bytes() {
    let url = serve(b"hello world");
    let mut src = HttpSource::open(url);
    let item = src.next().await.unwrap().unwrap();
    assert_eq!(&item.value[..], b"hello world");
    assert!(src.next().await.unwrap().is_none());
}

#[tokio::test]
async fn http_source_resumes_from_cache() {
    let url = serve(b"from-network");
    let cache = scratch("bask_dl_cache");

    let mut first = HttpSource::open(url.clone()).with_cache(&cache);
    assert_eq!(
        &first.next().await.unwrap().unwrap().value[..],
        b"from-network"
    );

    // Tamper with the cached copy: a second fetch must read it, not the network.
    let cached = find_one_file(&cache);
    std::fs::write(&cached, b"from-disk").unwrap();

    let mut second = HttpSource::open(url).with_cache(&cache);
    assert_eq!(
        &second.next().await.unwrap().unwrap().value[..],
        b"from-disk"
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn download_to_directory_through_the_engine() {
    let url = serve(b"engine-body");
    let out = scratch("bask_dl_out");

    let sinks = SinkRegistry::<Bytes>::blobs();
    let report = Engine::builder()
        .worker(SourceWorker::new(SourceRegistry::<Bytes>::blobs()))
        .worker_cfg(
            SinkWorker::open(&sinks, out.to_str().unwrap()).unwrap(),
            WorkerCfg::new().concurrency(1),
        )
        .seed(Read::<Bytes>::new(url))
        .run()
        .await
        .unwrap();

    assert!(
        report.failures.is_empty(),
        "failures: {:?}",
        report.failures
    );
    assert_eq!(report.stats.processed, 2); // 1 fetch + 1 write
    assert_eq!(std::fs::read(find_one_file(&out)).unwrap(), b"engine-body");
}