Move tooling to deib-polimi/noir-extra

.github/workflows/rust.yml

@@ -28,7 +28,7 @@ jobs:
         uses: actions-rs/clippy-check@v1
         with:
           token: ${{ secrets.GITHUB_TOKEN }}
-          args: --all-targets --tests --all -- -D warnings
+          args: --all-targets --all -- -D warnings
 
   test:
     name: Test

README.md

@@ -20,10 +20,10 @@ fn main() {
     let (config, args) = EnvironmentConfig::from_args();
     let mut env = StreamEnvironment::new(config);
     env.spawn_remote_workers();
-    let source = FileSource::new(&args[0]);
-    // Create file source
+
     let result = env
-        .stream(source)
+        // Open and read file line by line in parallel
+        .stream_file(&args[0])
         // Split into words
         .flat_map(|line| tokenize(&line))
         // Partition
@@ -32,10 +32,11 @@ fn main() {
         .fold(0, |count, _word| *count += 1)
         // Collect result
         .collect_vec();
+
     env.execute(); // Start execution (blocking)
     if let Some(result) = result.get() {
         // Print word counts
-        println!("{:?}", result);
+        result.into_iter().for_each(|(word, count)| println!("{word}: {count}"));
     }
 }
 
@@ -45,4 +46,6 @@ fn tokenize(s: &str) -> Vec<String> {
 }
 
 // Execute on 6 local hosts `cargo run -- -l 6 input.txt`
-```
+```
+
+# Documentation is WIP

examples/rolling_top_words.rs

@@ -170,7 +170,7 @@ fn main() {
     env.spawn_remote_workers();
 
     let source = ParallelIteratorSource::new(|id, num_replicas| {
-        TopicSource::new(id as u64, num_replicas as u64)
+        TopicSource::new(id, num_replicas)
     });
     env.stream(source)
         // add a timestamp for each item (using the one generated by the source) and add a watermark

examples/rolling_top_words_e2e.rs

@@ -117,7 +117,7 @@ fn main() {
     env.spawn_remote_workers();
 
     let source = ParallelIteratorSource::new(move |id, num_replicas| {
-        TopicSource::new(id as u64, num_replicas as u64, limit)
+        TopicSource::new(id, num_replicas, limit)
     });
     env.stream(source)
         // add a timestamp for each item (using the one generated by the source) and add a watermark

src/operator/source/file.rs

@@ -3,7 +3,6 @@ use std::fs::File;
 use std::io::BufRead;
 use std::io::Seek;
 use std::io::{BufReader, SeekFrom};
-use std::path::Path;
 use std::path::PathBuf;
 
 use crate::block::{BlockStructure, OperatorKind, OperatorStructure};
@@ -169,7 +168,7 @@ impl Clone for FileSource {
 
 impl crate::StreamEnvironment {
     /// Convenience method, creates a `FileSource` and makes a stream using `StreamEnvironment::stream`
-    pub fn stream_file(&mut self, path: &Path) -> Stream<String, FileSource> {
+    pub fn stream_file<P: Into<PathBuf>>(&mut self, path: P) -> Stream<String, FileSource> {
         let source = FileSource::new(path);
         self.stream(source)
     }