change ci cd pipeline to let it pass through nuclei (#248)

o0Ignition0o · web-flow · commit 09d28deba980 · 2020-07-29T09:31:39.000+02:00
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
@@ -59,29 +59,28 @@ jobs:
         uses: actions-rs/cargo@v1
         with:
           command: check
-          args: --all --bins --examples --tests
-
-      - name: tests windows
-        if: matrix.os != 'windows-latest'
-        uses: actions-rs/cargo@v1
-        with:
-          command: test
-          args: --all
+          args: --bins --examples --tests
 
       - name: tests windows
+        env:
+          RUST_BACKTRACE: full
         if: matrix.os == 'windows-latest'
         uses: actions-rs/cargo@v1
         with:
           command: test
 
       - name: tests nightly osx
+        env:
+          RUST_BACKTRACE: full
         if: ${{ matrix.version == 'nightly' && matrix.os == 'macOS-latest' }}
         uses: actions-rs/cargo@v1
         with:
           command: test
           args: --all --all-features
 
       - name: tests nightly linux
+        env:
+          RUST_BACKTRACE: full
         if: ${{ matrix.version == 'nightly' && matrix.os == 'ubuntu-latest' }}
         uses: actions-rs/cargo@v1
         with:
diff --git a/src/bastion/Cargo.toml b/src/bastion/Cargo.toml
@@ -49,9 +49,7 @@ docs = ["distributed", "scaling", "default"]
 features = ["docs"]
 rustdoc-args = ["--cfg", "feature=\"docs\""]
 
-
 [dependencies]
-nuclei = "0.1.1"
 bastion-executor = "0.3.5"
 lightproc = "0.3.5"
 # bastion-executor = { version = "= 0.3.5-alpha", path = "../bastion-executor" }
@@ -77,6 +75,9 @@ tracing-subscriber = "0.2.6"
 tracing = "0.1.15"
 anyhow = "1.0.31"
 
+[target.'cfg(not(windows))'.dependencies]
+nuclei = "0.1.1"
+
 [dev-dependencies]
 env_logger = "0.7"
 proptest = "0.10"
diff --git a/src/bastion/examples/distributed-fwrite.rs b/src/bastion/examples/distributed-fwrite.rs
@@ -1,6 +1,8 @@
 use bastion::prelude::*;
 use futures::*;
 use std::fs::{File, OpenOptions};
+#[cfg(target_os = "windows")]
+use std::io::Write;
 use std::path::PathBuf;
 use std::sync::Arc;
 
@@ -63,7 +65,11 @@ fn main() {
                     .write(true)
                     .open(&path)
                     .unwrap();
+
+                #[cfg(not(target_os = "windows"))]
                 let mut file = Handle::<File>::new(fo).unwrap();
+                #[cfg(target_os = "windows")]
+                let mut file = fo;
 
                 // Distribute your workload to workers
                 for id_worker_pair in workers.elems().iter().enumerate() {
@@ -74,6 +80,9 @@ fn main() {
                         msg: u64 => {
                             // Handle the answer...
                             println!("Source received the computed value: {}", msg);
+                            #[cfg(target_os = "windows")]
+                            file.write_all(msg.to_string().as_bytes()).unwrap();
+                            #[cfg(not(target_os = "windows"))]
                             file.write_all(msg.to_string().as_bytes()).await.unwrap();
                         };
                         _: _ => ();
diff --git a/src/bastion/examples/scaling_groups.rs b/src/bastion/examples/scaling_groups.rs
@@ -35,44 +35,55 @@ fn auto_resize_group_supervisor(supervisor: Supervisor) -> Supervisor {
 }
 
 fn input_group(children: Children) -> Children {
-    children
-        .with_redundancy(1)
-        .with_resizer(OptimalSizeExploringResizer::default().with_lower_bound(0)) // Don't start new actors after finishing execution
-        .with_exec(move |ctx: BastionContext| async move {
-            println!("[Input] Worker started!");
-
-            let mut messages_sent = 0;
-            static INPUT: [u64; 5] = [5u64, 1, 2, 4, 3];
-            let group_name = "Processing".to_string();
-            let target = BroadcastTarget::Group(group_name);
-
-            while messages_sent != 1000 {
-                // Emulate the workload. The number means how
-                // long it must wait before processing.
-                for value in INPUT.iter() {
-                    ctx.broadcast_message(target.clone(), value);
-                    Delay::new(Duration::from_millis(75 * value)).await;
-                }
-
-                messages_sent += INPUT.len();
+    // we would have fully chained the children builder if it wasn't for the feature flag
+    let mut children = children.with_redundancy(1);
+    #[cfg(feature = "scaling")]
+    {
+        // Don't start new actors after finishing execution
+        children =
+            children.with_resizer(OptimalSizeExploringResizer::default().with_lower_bound(0));
+    }
+    children.with_exec(move |ctx: BastionContext| async move {
+        println!("[Input] Worker started!");
+
+        let mut messages_sent = 0;
+        static INPUT: [u64; 5] = [5u64, 1, 2, 4, 3];
+        let group_name = "Processing".to_string();
+        let target = BroadcastTarget::Group(group_name);
+
+        while messages_sent != 1000 {
+            // Emulate the workload. The number means how
+            // long it must wait before processing.
+            for value in INPUT.iter() {
+                ctx.broadcast_message(target.clone(), value);
+                Delay::new(Duration::from_millis(75 * value)).await;
             }
 
-            Ok(())
-        })
+            messages_sent += INPUT.len();
+        }
+
+        Ok(())
+    })
 }
 
 fn auto_resize_group(children: Children) -> Children {
-    children
+    // we would have fully chained the children builder if it wasn't for the feature flag
+    let mut children = children
         .with_redundancy(3) // Start with 3 actors
-        .with_heartbeat_tick(Duration::from_secs(5)) // Do heartbeat each 5 seconds
-        .with_resizer(
+        .with_heartbeat_tick(Duration::from_secs(5)); // Do heartbeat each 5 seconds
+
+    #[cfg(feature = "scaling")]
+    {
+        children = children.with_resizer(
             OptimalSizeExploringResizer::default()
                 .with_lower_bound(1) // A minimal acceptable size of group
                 .with_upper_bound(UpperBound::Limit(10)) // Max 10 actors in runtime
                 .with_upscale_strategy(UpscaleStrategy::MailboxSizeThreshold(3)) // Scale up when a half of actors have more than 3 messages
                 .with_upscale_rate(0.1) // Increase the size of group on 10%, if necessary to scale up
                 .with_downscale_rate(0.2), // Decrease the size of group on 20%, if too many free actors
-        )
+        );
+    }
+    children
         .with_dispatcher(Dispatcher::with_type(DispatcherType::Named(
             "Processing".to_string(),
         )))
diff --git a/src/bastion/examples/tcp-servers.rs b/src/bastion/examples/tcp-servers.rs
@@ -1,13 +1,62 @@
 use bastion::prelude::*;
+#[cfg(not(target_os = "windows"))]
 use futures::io;
-use std::net::{TcpListener, TcpStream};
+#[cfg(target_os = "windows")]
+use std::io::{self, Read, Write};
+use std::net::{TcpListener, TcpStream, ToSocketAddrs};
 use std::sync::atomic::{AtomicUsize, Ordering};
 
+#[cfg(not(target_os = "windows"))]
+async fn run(addr: impl ToSocketAddrs) -> io::Result<()> {
+    let listener = Handle::<TcpListener>::bind(addr).unwrap();
+    println!("Listening on {}", listener.get_ref().local_addr().unwrap());
+
+    // Accept clients in a loop.
+    loop {
+        let (stream, peer_addr) = listener.accept().await.unwrap();
+        println!("Accepted client: {}", peer_addr);
+
+        // Spawn a task that echoes messages from the client back to it.
+        spawn(echo(stream));
+    }
+    Ok(())
+}
+
+#[cfg(target_os = "windows")]
+async fn run(addr: impl ToSocketAddrs) -> io::Result<()> {
+    let listener = std::net::TcpListener::bind(addr).unwrap();
+    println!("Listening on {}", listener.local_addr().unwrap());
+
+    // Accept clients in a loop.
+    for stream in listener.incoming() {
+        match stream {
+            Ok(stream) => {
+                println!("Accepted client");
+                spawn!(echo(stream));
+            }
+            _ => {
+                break;
+            }
+        }
+    }
+    Ok(())
+}
+
+#[cfg(not(target_os = "windows"))]
 async fn echo(stream: Handle<TcpStream>) -> io::Result<()> {
     io::copy(&stream, &mut &stream).await?;
     Ok(())
 }
 
+#[cfg(target_os = "windows")]
+async fn echo(mut stream: TcpStream) -> io::Result<()> {
+    let mut buf = [0 as u8; 256];
+    while let Ok(size) = stream.read(&mut buf) {
+        stream.write(&buf[0..size]).unwrap();
+    }
+    Ok(())
+}
+
 const TCP_SERVER_COUNT: usize = 10;
 static TCP_SERVERS: AtomicUsize = AtomicUsize::new(TCP_SERVER_COUNT);
 
@@ -17,7 +66,8 @@ static TCP_SERVERS: AtomicUsize = AtomicUsize::new(TCP_SERVER_COUNT);
 /// Prologue:
 ///
 /// This example demonstrates using 10 parallel tcp servers
-
+/// non windows versions use io::copy from nuclei
+/// windows versions use a regular stream copy
 fn main() {
     env_logger::init();
 
@@ -26,26 +76,14 @@ fn main() {
     let _tcp_servers = Bastion::children(|children: Children| {
         children
             .with_redundancy(TCP_SERVER_COUNT) // Let's have 10 tcp echo servers :)
-            .with_exec(move |_ctx: BastionContext| {
-                async move {
-                    println!("Server is starting!");
-                    let port = TCP_SERVERS.fetch_sub(1, Ordering::SeqCst) + 2000;
-                    let addr = format!("127.0.0.1:{}", port);
-
-                    let listener = Handle::<TcpListener>::bind(addr).unwrap();
-                    println!("Listening on {}", listener.get_ref().local_addr().unwrap());
-
-                    // Accept clients in a loop.
-                    loop {
-                        let (stream, peer_addr) = listener.accept().await.unwrap();
-                        println!("Accepted client: {}", peer_addr);
-
-                        // Spawn a task that echoes messages from the client back to it.
-                        spawn(echo(stream));
-                    }
-
-                    Ok(())
-                }
+            .with_exec(move |_ctx: BastionContext| async move {
+                println!("Server is starting!");
+                let port = TCP_SERVERS.fetch_sub(1, Ordering::SeqCst) + 2000;
+                let addr = format!("127.0.0.1:{}", port);
+
+                run(addr);
+
+                Ok(())
             })
     })
     .expect("Couldn't start a new children group.");
diff --git a/src/bastion/src/bastion.rs b/src/bastion/src/bastion.rs
@@ -118,7 +118,14 @@ distributed_api! {
 ///     // of reference to its elements...
 ///     let children = Bastion::children(|children| {
 ///         // ...
-///         # children
+///         # children.with_exec(|ctx: BastionContext| {
+///         #   async move {
+///         #       msg! { ctx.recv().await?,
+///         #            _: _ => ();
+///         #        }
+///         #        Ok(())
+///         #    }
+///         # })
 ///     }).expect("Couldn't create the children group.");
 ///     let elems: &[ChildRef] = children.elems();
 ///
diff --git a/src/bastion/src/child_ref.rs b/src/bastion/src/child_ref.rs
@@ -226,14 +226,32 @@ impl ChildRef {
     /// # use bastion::prelude::*;
     /// #
     /// # Bastion::init();
-    /// #
-    /// # let children_ref = Bastion::children(|children| children).unwrap();
-    /// # let child_ref = &children_ref.elems()[0];
-    /// child_ref.stop().expect("Couldn't send the message.");
-    /// #
-    /// # Bastion::start();
-    /// # Bastion::stop();
-    /// # Bastion::block_until_stopped();
+    /// # let children_ref =
+    /// # Bastion::children(|children| {
+    ///     children.with_exec(|ctx: BastionContext| {
+    ///         async move {
+    ///             // ...which will receive the message asked...
+    ///             msg! { ctx.recv().await?,
+    ///                 msg: &'static str =!> {
+    ///                     // Handle the message...
+    ///
+    ///                     // ...and eventually answer to it...
+    ///                 };
+    ///                 // This won't happen because this example
+    ///                 // only "asks" a `&'static str`...
+    ///                 _: _ => ();
+    ///             }
+    ///
+    ///             Ok(())
+    ///         }
+    ///     })
+    /// }).expect("Couldn't create the children group.");
+    ///     # Bastion::start();
+    ///     # let child_ref = &children_ref.elems()[0];
+    ///     child_ref.stop().expect("Couldn't send the message.");
+    ///     #
+    ///     # Bastion::stop();
+    ///     # Bastion::block_until_stopped();
     /// ```
     pub fn stop(&self) -> Result<(), ()> {
         debug!("ChildRef({}): Stopping.", self.id);
diff --git a/src/bastion/src/children.rs b/src/bastion/src/children.rs
@@ -887,7 +887,7 @@ impl Children {
         let children = self.as_ref();
         let supervisor = self.bcast.parent().clone().into_supervisor();
 
-        #[warn(unused_mut)]
+        #[allow(unused_mut)]
         let mut state = ContextState::new();
         #[cfg(feature = "scaling")]
         self.init_data_for_scaling(&mut state);
diff --git a/src/bastion/src/lib.rs b/src/bastion/src/lib.rs
@@ -79,6 +79,7 @@ pub mod context;
 pub mod dispatcher;
 pub mod envelope;
 pub mod executor;
+#[cfg(not(target_os = "windows"))]
 pub mod io;
 pub mod message;
 pub mod path;
@@ -106,6 +107,7 @@ pub mod prelude {
         DispatcherType, NotificationType,
     };
     pub use crate::envelope::{RefAddr, SignedMessage};
+    #[cfg(not(target_os = "windows"))]
     pub use crate::io::*;
     pub use crate::message::{Answer, AnswerSender, Message, Msg};
     pub use crate::msg;
diff --git a/src/bastion/tests/message_signatures.rs b/src/bastion/tests/message_signatures.rs
