Port to async/await

appveyor.yml

@@ -7,7 +7,7 @@ test: off
 
 environment:
   CLI_VERSION: '5.0.165'
-  RUST_TOOLCHAIN: 'stable'
+  RUST_TOOLCHAIN: 'nightly-2019-09-10'
   DOCKER_TOKEN:
    secure: QKr2YEuliXdFKe3jN7w97w==
   DOCKER_USER:

ci/build-deps.ps1

@@ -163,6 +163,7 @@ function Start-SeqEnvironment($protocol) {
         --network sqelf-test `
         -e SEQ_ADDRESS=http://sqelf-test-seq:5341 `
         -e GELF_ADDRESS="${protocol}://0.0.0.0:12201" `
+        -e GELF_ENABLE_DIAGNOSTICS="True" `
         -itd `
         -p "12202:${portArg}" `
         sqelf-ci:latest
@@ -224,7 +225,7 @@ function Invoke-TestApp($protocol) {
     if ($LASTEXITCODE) { exit 1 }
 
     # Give sqelf enough time to batch and send
-    Start-Sleep -Seconds 2
+    Start-Sleep -Seconds 5
 }
 
 function Check-ClefOutput {

ci/linux-x64/build.ps1

@@ -10,11 +10,34 @@ Push-Location "$PSScriptRoot/../../"
 function Invoke-SmokeTest($protocol) {
     Write-BeginStep $MYINVOCATION
 
-    Start-SeqEnvironment($protocol)
-    Invoke-TestApp($protocol)
-    Check-SqelfLogs
-    Check-SeqLogs
-    Check-ClefOutput
+    $finished = $false
+    $retries = 0
+
+    do {
+        try {
+            Start-SeqEnvironment($protocol)
+            Invoke-TestApp($protocol)
+            Check-SqelfLogs
+            Check-SeqLogs
+            Check-ClefOutput
+
+            $finished = $true
+        }
+        catch {
+            Stop-SeqEnvironment
+
+            if ($retries -gt 3) {
+                exit 1
+            }
+            else {
+                $retries = $retries + 1
+                Write-Host "Retrying (attempt $retries)"
+            }
+        }
+
+    }
+    while ($finished -eq $false)
+
     Stop-SeqEnvironment
 }
 

ci/linux-x64/setup.sh

@@ -1,6 +1,6 @@
 #!/bin/bash
 
-RequiredRustToolchain="stable"
+RequiredRustToolchain=$RUST_TOOLCHAIN
 
 curl https://sh.rustup.rs -sSf | sh -s -- --default-host x86_64-unknown-linux-gnu --default-toolchain $RequiredRustToolchain -y
 

ci/win-x64/setup.ps1

@@ -1,6 +1,6 @@
 $ErrorActionPreference = "Stop"
 
-$RequiredRustToolchain = "stable"
+$RequiredRustToolchain = $env:RUST_TOOLCHAIN
 
 Invoke-WebRequest -OutFile ./rustup-init.exe -Uri https://win.rustup.rs
 $ErrorActionPreference = "Continue"

rust-toolchain

@@ -0,0 +1 @@
+nightly-2019-09-10

rustfmt.toml

@@ -0,0 +1,2 @@
+imports_indent = "Block"
+imports_layout = "Vertical"

sqelf/Cargo.toml

@@ -5,14 +5,16 @@ authors = ["Datalust"]
 edition = "2018"
 license = "Apache-2.0"
 
-[dependencies.futures]
-version = "0.1"
+[dependencies.futures-preview]
+version = "0.3.0-alpha.18"
+features = ["nightly", "async-await"]
 
-[dependencies.tokio]
-version = "0.1"
+[dependencies.pin-utils]
+version = "0.1.0-alpha.4"
 
-[dependencies.tokio-signal]
-version = "0.2"
+[dependencies.tokio]
+version = "0.2.0-alpha.4"
+features = ["signal"]
 
 [dependencies.bytes]
 version = "0.4"
@@ -44,3 +46,6 @@ version = "1"
 [dependencies.chrono]
 version = "0.4"
 features = ["serde"]
+
+[dependencies.lazy_static]
+version = "1"

sqelf/src/config.rs

@@ -1,6 +1,15 @@
-use std::{env, str::FromStr};
+use std::{
+    env,
+    str::FromStr,
+};
 
-use crate::{diagnostics, process, receive, server, Error};
+use crate::{
+    diagnostics,
+    process,
+    receive,
+    server,
+    Error,
+};
 
 #[derive(Debug, Default, Clone)]
 pub struct Config {
@@ -35,7 +44,7 @@ impl Config {
     }
 }
 
-pub(crate) fn is_seq_app() -> bool {
+pub fn is_seq_app() -> bool {
     env::var("SEQ_APP_ID").is_ok()
 }
 

sqelf/src/diagnostics.rs

@@ -1,18 +1,34 @@
-use crate::error::{err_msg, Error};
-use chrono::{DateTime, Utc};
+use crate::error::{
+    err_msg,
+    Error,
+};
+use chrono::{
+    DateTime,
+    Utc,
+};
 use std::{
     collections::HashMap,
     fmt::Display,
     ops::Drop,
     str::FromStr,
     sync::{
-        atomic::{AtomicUsize, Ordering},
+        atomic::{
+            AtomicUsize,
+            Ordering,
+        },
         mpsc,
+        Mutex,
     },
     thread,
     time::Duration,
 };
 
+pub(crate) static MIN_LEVEL: MinLevel = MinLevel(AtomicUsize::new(0));
+
+lazy_static! {
+    static ref DIAGNOSTICS: Mutex<Option<Diagnostics>> = Mutex::new(None);
+}
+
 /**
 Diagnostics configuration.
 */
@@ -37,33 +53,17 @@ impl Default for Config {
     }
 }
 
-pub(crate) struct Diagnostics {
-    metrics: Option<(mpsc::Sender<()>, thread::JoinHandle<()>)>,
-}
-
-impl Diagnostics {
-    pub fn stop_metrics(&mut self) -> Result<(), Error> {
-        if let Some((tx, handle)) = self.metrics.take() {
-            tx.send(())?;
-
-            handle
-                .join()
-                .map_err(|_| err_msg("failed to join diagnostics handle"))?;
-        }
-
-        Ok(())
-    }
-}
+/**
+Initialize process-wide diagnostics.
+*/
+pub fn init(config: Config) {
+    let mut diagnostics = DIAGNOSTICS.lock().expect("failed to lock diagnostics");
 
-impl Drop for Diagnostics {
-    fn drop(&mut self) {
-        if let Some((tx, _)) = self.metrics.take() {
-            let _ = tx.send(());
-        }
+    if diagnostics.is_some() {
+        drop(diagnostics);
+        panic!("GELF diagnostics have already been initialized");
     }
-}
 
-pub(crate) fn init(config: Config) -> Diagnostics {
     MIN_LEVEL.set(config.min_level);
 
     // Only set up metrics if the minimum level is Debug
@@ -75,8 +75,11 @@ pub(crate) fn init(config: Config) -> Diagnostics {
         let metrics_timeout = Duration::from_millis(config.metrics_interval_ms);
         let handle = thread::spawn(move || loop {
             match rx.recv_timeout(metrics_timeout) {
-                Ok(()) | Err(mpsc::RecvTimeoutError::Disconnected) => return,
-                _ => {
+                Ok(()) | Err(mpsc::RecvTimeoutError::Disconnected) => {
+                    emit_metrics();
+                    return;
+                }
+                Err(mpsc::RecvTimeoutError::Timeout) => {
                     emit_metrics();
                 }
             }
@@ -87,7 +90,46 @@ pub(crate) fn init(config: Config) -> Diagnostics {
         None
     };
 
-    Diagnostics { metrics }
+    *diagnostics = Some(Diagnostics { metrics });
+}
+
+/**
+Stop process-wide diagnostics.
+*/
+pub fn stop() -> Result<(), Error> {
+    let mut diagnostics = DIAGNOSTICS.lock().expect("failed to lock diagnostics");
+
+    if let Some(mut diagnostics) = diagnostics.take() {
+        diagnostics.stop_metrics()?;
+    }
+
+    Ok(())
+}
+
+struct Diagnostics {
+    metrics: Option<(mpsc::Sender<()>, thread::JoinHandle<()>)>,
+}
+
+impl Diagnostics {
+    fn stop_metrics(&mut self) -> Result<(), Error> {
+        if let Some((tx, handle)) = self.metrics.take() {
+            tx.send(())?;
+
+            handle
+                .join()
+                .map_err(|_| err_msg("failed to join diagnostics handle"))?;
+        }
+
+        Ok(())
+    }
+}
+
+impl Drop for Diagnostics {
+    fn drop(&mut self) {
+        if let Some((tx, _)) = self.metrics.take() {
+            let _ = tx.send(());
+        }
+    }
 }
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
@@ -177,7 +219,7 @@ pub fn emit_err(error: &impl Display, message_template: &'static str) {
     }
 }
 
-pub fn emit_metrics() {
+fn emit_metrics() {
     if MIN_LEVEL.includes(Level::Debug) {
         #[derive(Serialize)]
         struct EmitMetrics {
@@ -214,56 +256,7 @@ pub fn emit_metrics() {
     }
 }
 
-/// For use with `map_err`
-pub(crate) fn emit_err_abort<TInner>(message_template: &'static str) -> impl Fn(TInner) -> ()
-where
-    TInner: Display,
-{
-    emit_err_abort_with(message_template, || ())
-}
-
-/// For use with `map_err`
-pub(crate) fn emit_err_abort_with<TInner, TError>(
-    message_template: &'static str,
-    err: impl Fn() -> TError,
-) -> impl Fn(TInner) -> TError
-where
-    TInner: Display,
-{
-    move |e| {
-        emit_err(&e, message_template);
-
-        err()
-    }
-}
-
-/// For use with `or_else`
-pub(crate) fn emit_err_continue<TInner, TOuter>(
-    message_template: &'static str,
-) -> impl Fn(TInner) -> Result<(), TOuter>
-where
-    TInner: Display,
-{
-    emit_err_continue_with(message_template, || ())
-}
-
-/// For use with `or_else`
-pub(crate) fn emit_err_continue_with<TInner, TOk, TOuter>(
-    message_template: &'static str,
-    ok: impl Fn() -> TOk,
-) -> impl Fn(TInner) -> Result<TOk, TOuter>
-where
-    TInner: Display,
-{
-    move |err| {
-        emit_err(&err, message_template);
-
-        Ok(ok())
-    }
-}
-
 pub(crate) struct MinLevel(AtomicUsize);
-pub(crate) static MIN_LEVEL: MinLevel = MinLevel(AtomicUsize::new(0));
 
 impl MinLevel {
     fn set(&self, min: Level) {