migrate to tokio 0.3

ffi/rodbus-ffi/Cargo.toml

@@ -16,7 +16,7 @@ crate-type = ["cdylib"]
 [dependencies]
 rodbus = { path = "../../rodbus" }
 log = { version = "0.4", features = ["std"] }
-tokio = { version = "^0.2.11", features = ["rt-threaded"]}
+tokio = { version = "0.3", features = ["rt-multi-thread"]}
 
 [build-dependencies]
 rodbus-schema = { path = "../rodbus-schema" }

ffi/rodbus-ffi/src/channel.rs

@@ -3,7 +3,7 @@ use std::ptr::null_mut;
 
 pub struct Channel {
     pub(crate) inner: rodbus::client::channel::Channel,
-    pub(crate) runtime: tokio::runtime::Handle,
+    pub(crate) runtime: crate::Runtime,
 }
 
 pub(crate) unsafe fn create_tcp_client(
@@ -25,11 +25,11 @@ pub(crate) unsafe fn create_tcp_client(
         rodbus::client::channel::strategy::default(),
     );
 
-    rt.spawn(task);
+    rt.inner.spawn(task);
 
     Box::into_raw(Box::new(Channel {
         inner: handle,
-        runtime: rt.handle().clone(),
+        runtime: rt.clone(),
     }))
 }
 
@@ -66,7 +66,7 @@ pub(crate) unsafe fn channel_read_coils_async(
     let mut session = param.build_session(channel);
 
     channel
-        .runtime
+        .runtime.inner
         .block_on(session.read_coils(range, callback));
 }
 
@@ -98,6 +98,7 @@ pub(crate) unsafe fn channel_read_discrete_inputs_async(
 
     channel
         .runtime
+        .inner
         .block_on(session.read_discrete_inputs(range, callback));
 }
 
@@ -129,6 +130,7 @@ pub(crate) unsafe fn channel_read_holding_registers_async(
 
     channel
         .runtime
+        .inner
         .block_on(session.read_holding_registers(range, callback));
 }
 
@@ -160,6 +162,7 @@ pub(crate) unsafe fn channel_read_input_registers_async(
 
     channel
         .runtime
+        .inner
         .block_on(session.read_input_registers(range, callback));
 }
 
@@ -181,6 +184,7 @@ pub(crate) unsafe fn channel_write_single_coil_async(
 
     channel
         .runtime
+        .inner
         .block_on(session.write_single_coil(bit.into(), callback.convert_to_fn_once()));
 }
 
@@ -202,6 +206,7 @@ pub(crate) unsafe fn channel_write_single_register_async(
 
     channel
         .runtime
+        .inner
         .block_on(session.write_single_register(register.into(), callback.convert_to_fn_once()));
 }
 
@@ -242,6 +247,7 @@ pub(crate) unsafe fn channel_write_multiple_coils_async(
 
     channel
         .runtime
+        .inner
         .block_on(session.write_multiple_coils(argument, callback));
 }
 
@@ -282,5 +288,6 @@ pub(crate) unsafe fn channel_write_multiple_registers_async(
 
     channel
         .runtime
+        .inner
         .block_on(session.write_multiple_registers(argument, callback));
 }

ffi/rodbus-ffi/src/runtime.rs

@@ -1,20 +1,25 @@
-pub use tokio::runtime::Runtime;
+use std::sync::Arc;
+
+#[derive(Clone)]
+pub struct Runtime {
+    pub(crate) inner: Arc<tokio::runtime::Runtime>
+}
 
 pub(crate) unsafe fn runtime_new(
     config: Option<&crate::ffi::RuntimeConfig>,
-) -> *mut tokio::runtime::Runtime {
-    let mut builder = tokio::runtime::Builder::new();
+) -> *mut crate::Runtime {
+    let mut builder = tokio::runtime::Builder::new_multi_thread();
 
-    builder.enable_all().threaded_scheduler();
+    builder.enable_all();
 
     if let Some(x) = config.as_ref() {
         if x.num_core_threads > 0 {
-            builder.core_threads(x.num_core_threads as usize);
+            builder.worker_threads(x.num_core_threads as usize);
         }
     }
 
     match builder.build() {
-        Ok(r) => Box::into_raw(Box::new(r)),
+        Ok(r) => Box::into_raw(Box::new(crate::Runtime { inner: Arc::new(r) })),
         Err(_) => std::ptr::null_mut(),
     }
 }

ffi/rodbus-ffi/src/server.rs

@@ -120,7 +120,7 @@ impl RequestHandler for RequestHandlerWrapper {
 }
 
 pub struct Server {
-    runtime: tokio::runtime::Handle,
+    runtime: crate::Runtime,
     // never used but we have to hang onto it otherwise the server shuts down
     _server: rodbus::shutdown::TaskHandle,
     map: ServerHandlerMap<RequestHandlerWrapper>,
@@ -190,7 +190,7 @@ pub(crate) unsafe fn create_tcp_server(
     };
 
     // at this point, we know that all the arguments are good, so we can go ahead and try to bind a listener
-    let listener = match runtime.block_on(TcpListener::bind(address)) {
+    let listener = match runtime.inner.block_on(TcpListener::bind(address)) {
         Ok(x) => x,
         Err(err) => {
             log::error!("error binding listener: {}", err);
@@ -207,11 +207,11 @@ pub(crate) unsafe fn create_tcp_server(
         listener,
         handler_map.clone(),
     );
-    let join_handle = runtime.spawn(task);
+    let join_handle = runtime.inner.spawn(task);
 
     let server_handle = Server {
         _server: TaskHandle::new(tx, join_handle),
-        runtime: runtime.handle().clone(),
+        runtime: runtime.clone(),
         map: handler_map,
     };
 
@@ -244,7 +244,7 @@ pub(crate) unsafe fn server_update_database(
         transaction.callback(&mut lock.database);
     };
 
-    server.runtime.block_on(transaction);
+    server.runtime.inner.block_on(transaction);
 
     true
 }

rodbus-client/Cargo.toml

@@ -17,6 +17,6 @@ path = "src/main.rs"
 [dependencies]
 rodbus = { path = "../rodbus", version = "0.1.1" }
 clap = "2.33"
-tokio = { version = "^0.2.11", features = ["macros", "time"] }
+tokio = { version = "0.3", features = ["macros", "time"] }
 simple_logger = "1.9"
 log = "0.4"

rodbus-client/src/main.rs

@@ -50,7 +50,7 @@ impl Args {
     }
 }
 
-#[tokio::main(basic_scheduler)]
+#[tokio::main(flavor = "current_thread")]
 async fn main() -> Result<(), Box<dyn std::error::Error>> {
     // print log messages to the console
     SimpleLogger::new()
@@ -74,7 +74,7 @@ async fn run() -> Result<(), Error> {
         None => run_command(&args.command, &mut session).await,
         Some(period) => loop {
             run_command(&args.command, &mut session).await?;
-            tokio::time::delay_for(period).await
+            tokio::time::sleep(period).await
         },
     }
 }

rodbus/Cargo.toml

@@ -15,10 +15,10 @@ codecov = { repository = "automatak/rodbus", branch = "master", service = "githu
 maintenance = { status = "actively-developed" }
 
 [dependencies]
-tokio = { version = "^0.2.20", features = ["tcp", "sync", "io-util", "time", "rt-core", "rt-threaded", "macros"]}
+tokio = { version = "0.3", features = ["net", "sync", "io-util", "time", "rt", "rt-multi-thread", "macros"]}
 log = "0.4"
 no-panic = { version = "0.1", optional = true }
 
 [dev-dependencies]
-tokio-test = "0.2"
+tokio-test = "0.3"
 simple_logger = "1.9"

rodbus/examples/client.rs

@@ -3,7 +3,7 @@ use std::time::Duration;
 
 use rodbus::prelude::*;
 
-#[tokio::main(basic_scheduler)]
+#[tokio::main(flavor = "current_thread")]
 async fn main() -> Result<(), Box<dyn Error>> {
     // Create a channel
     let channel = spawn_tcp_client_task("127.0.0.1:502".parse().unwrap(), 1, strategy::default());

rodbus/examples/perf.rs

@@ -20,7 +20,7 @@ impl RequestHandler for Handler {
     }
 }
 
-#[tokio::main(threaded_scheduler)]
+#[tokio::main(flavor = "multi_thread")]
 async fn main() -> Result<(), Box<dyn std::error::Error>> {
     let args: Vec<String> = std::env::args().collect();
 

rodbus/examples/server.rs

@@ -82,7 +82,7 @@ impl RequestHandler for SimpleHandler {
     }
 }
 
-#[tokio::main(threaded_scheduler)]
+#[tokio::main(flavor = "multi_thread")]
 async fn main() -> Result<(), Box<dyn std::error::Error>> {
     let args: Vec<String> = std::env::args().collect();
 
@@ -127,6 +127,6 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
                 *c = !*c;
             }
         }
-        tokio::time::delay_until(next).await;
+        tokio::time::sleep_until(next).await;
     }
 }

rodbus/src/lib.rs

@@ -43,7 +43,7 @@
 //!use std::str::FromStr;
 //!
 //!
-//!use tokio::time::delay_for;
+//!use tokio::time::sleep;
 //!
 //!#[tokio::main]
 //!async fn main() -> Result<(), Box<dyn std::error::Error>> {
@@ -70,7 +70,7 @@
 //!            Err(err) => println!("Error: {:?}", err)
 //!        }
 //!
-//!        delay_for(std::time::Duration::from_secs(3)).await
+//!        sleep(std::time::Duration::from_secs(3)).await
 //!    }
 //!}
 //! ```
@@ -102,7 +102,7 @@
 //!    }
 //! }
 //!
-//! #[tokio::main(threaded_scheduler)]
+//! #[tokio::main(flavor = "multi_thread")]
 //! async fn main() -> Result<(), Box<dyn std::error::Error>> {
 //!
 //!    let handler = CoilsOnlyHandler::new().wrap();
@@ -129,7 +129,7 @@
 //!                *c = !*c;
 //!            }
 //!        }
-//!        tokio::time::delay_until(next).await;
+//!        tokio::time::sleep_until(next).await;
 //!    }
 //!}
 //!```

rodbus/tests/integration_test.rs

@@ -212,6 +212,6 @@ async fn test_requests_and_responses() {
 
 #[test]
 fn can_read_and_write_values() {
-    let mut rt = Runtime::new().unwrap();
+    let rt = Runtime::new().unwrap();
     rt.block_on(test_requests_and_responses())
 }