Add Sync requirement to handlers. (#110)

Cargo.toml

@@ -14,3 +14,6 @@ bitflags = "1.1"
 jack-sys = { path = "./jack-sys", version = "0.2" }
 lazy_static = "1.4"
 libc = "0.2"
+
+[dev-dependencies]
+crossbeam-channel = "0.3"

README.md

@@ -49,35 +49,24 @@ Missing categories include, JACK threading, synchronous processing, transport an
 
 ## Testing
 
-Testing is a little awkward to setup since it relies on a JACK server.
-
-### Setting Up JACK Dummy Server
+Testing requires setting up a dummy server and running the tests using a single
+thread.
 
 ```bash
+$ # Set up a dummy server for tests.
 $ ./dummy_jack_server.sh
+$ # Run tests with limited concurrency.
+$ RUST_TEST_THREADS=1 cargo test
 ```
 
-which runs the command
-
-```bash
-$ jackd -r -ddummy -r44100 -p1024 & # Start the dummy JACK server
-```
-
-Testing expects there to be an available JACK server running at a sample rate of
-44.1kHz and a buffer size of 1024 samples.
-
-### Running the tests
-
-```bash
-$ cargo test
-```
+**Note:** We use a single thread for tests since too multiple client
+instantiations in short periods of time cause the JACK server to become flaky.
 
 #### Possible Issues
 
 If the tests are failing, a possible gotcha may be timing issues.
 
-1. Rust runs tests in parallel, it may be possible that the JACK server is not keeping up. Set the environment variable `RUST_TEST_THREADS` to 1.
-2. Increase the value used by `sleep_on_test` in `client/common.rs`.
+1. Increase the value used by `sleep_on_test` in `client/common.rs`.
 
 Another case is that libjack may be broken on your setup.  Try switching between
 libjack and libjack2 (they have the same API and libjack2 isn't necessarily

examples/sine.rs

@@ -1,9 +1,11 @@
 //! Sine wave generator with frequency configuration exposed through standard
 //! input.
+extern crate crossbeam_channel;
 extern crate jack;
+
+use crossbeam_channel::bounded;
 use std::io;
 use std::str::FromStr;
-use std::sync::mpsc::channel;
 
 fn main() {
     // 1. open a client
@@ -20,7 +22,7 @@ fn main() {
     let sample_rate = client.sample_rate();
     let frame_t = 1.0 / sample_rate as f64;
     let mut time = 0.0;
-    let (tx, rx) = channel();
+    let (tx, rx) = bounded(1_000_000);
     let process = jack::ClosureProcessHandler::new(
         move |_: &jack::Client, ps: &jack::ProcessScope| -> jack::Control {
             // Get output buffer

src/client/async_client.rs

@@ -32,8 +32,8 @@ pub struct AsyncClient<N, P> {
 
 impl<N, P> AsyncClient<N, P>
 where
-    N: 'static + Send + NotificationHandler,
-    P: 'static + Send + ProcessHandler,
+    N: 'static + Send + Sync + NotificationHandler,
+    P: 'static + Send + Sync + ProcessHandler,
 {
     /// Tell the JACK server that the program is ready to start processing audio. JACK will call the
     /// methods specified by the `NotificationHandler` and `ProcessHandler` objects.

src/client/callbacks.rs

@@ -162,8 +162,8 @@ pub trait ProcessHandler: Send {
 
 unsafe extern "C" fn thread_init_callback<N, P>(data: *mut libc::c_void)
 where
-    N: 'static + Send + NotificationHandler,
-    P: 'static + Send + ProcessHandler,
+    N: 'static + Send + Sync + NotificationHandler,
+    P: 'static + Send + Sync + ProcessHandler,
 {
     let ctx = CallbackContext::<N, P>::from_raw(data);
     ctx.notification.thread_init(&ctx.client)
@@ -174,8 +174,8 @@ unsafe extern "C" fn shutdown<N, P>(
     reason: *const libc::c_char,
     data: *mut libc::c_void,
 ) where
-    N: 'static + Send + NotificationHandler,
-    P: 'static + Send + ProcessHandler,
+    N: 'static + Send + Sync + NotificationHandler,
+    P: 'static + Send + Sync + ProcessHandler,
 {
     let ctx = CallbackContext::<N, P>::from_raw(data);
     let cstr = ffi::CStr::from_ptr(reason);
@@ -191,8 +191,8 @@ unsafe extern "C" fn shutdown<N, P>(
 
 unsafe extern "C" fn process<N, P>(n_frames: Frames, data: *mut libc::c_void) -> libc::c_int
 where
-    N: 'static + Send + NotificationHandler,
-    P: 'static + Send + ProcessHandler,
+    N: 'static + Send + Sync + NotificationHandler,
+    P: 'static + Send + Sync + ProcessHandler,
 {
     let ctx = CallbackContext::<N, P>::from_raw(data);
     let scope = ProcessScope::from_raw(n_frames, ctx.client.raw());
@@ -201,8 +201,8 @@ where
 
 unsafe extern "C" fn freewheel<N, P>(starting: libc::c_int, data: *mut libc::c_void)
 where
-    N: 'static + Send + NotificationHandler,
-    P: 'static + Send + ProcessHandler,
+    N: 'static + Send + Sync + NotificationHandler,
+    P: 'static + Send + Sync + ProcessHandler,
 {
     let ctx = CallbackContext::<N, P>::from_raw(data);
     let is_starting = match starting {
@@ -214,17 +214,17 @@ where
 
 unsafe extern "C" fn buffer_size<N, P>(n_frames: Frames, data: *mut libc::c_void) -> libc::c_int
 where
-    N: 'static + Send + NotificationHandler,
-    P: 'static + Send + ProcessHandler,
+    N: 'static + Send + Sync + NotificationHandler,
+    P: 'static + Send + Sync + ProcessHandler,
 {
     let ctx = CallbackContext::<N, P>::from_raw(data);
     ctx.notification.buffer_size(&ctx.client, n_frames).to_ffi()
 }
 
 unsafe extern "C" fn sample_rate<N, P>(n_frames: Frames, data: *mut libc::c_void) -> libc::c_int
 where
-    N: 'static + Send + NotificationHandler,
-    P: 'static + Send + ProcessHandler,
+    N: 'static + Send + Sync + NotificationHandler,
+    P: 'static + Send + Sync + ProcessHandler,
 {
     let ctx = CallbackContext::<N, P>::from_raw(data);
     ctx.notification.sample_rate(&ctx.client, n_frames).to_ffi()
@@ -235,8 +235,8 @@ unsafe extern "C" fn client_registration<N, P>(
     register: libc::c_int,
     data: *mut libc::c_void,
 ) where
-    N: 'static + Send + NotificationHandler,
-    P: 'static + Send + ProcessHandler,
+    N: 'static + Send + Sync + NotificationHandler,
+    P: 'static + Send + Sync + ProcessHandler,
 {
     let ctx = CallbackContext::<N, P>::from_raw(data);
     let name = ffi::CStr::from_ptr(name).to_str().unwrap();
@@ -253,8 +253,8 @@ unsafe extern "C" fn port_registration<N, P>(
     register: libc::c_int,
     data: *mut libc::c_void,
 ) where
-    N: 'static + Send + NotificationHandler,
-    P: 'static + Send + ProcessHandler,
+    N: 'static + Send + Sync + NotificationHandler,
+    P: 'static + Send + Sync + ProcessHandler,
 {
     let ctx = CallbackContext::<N, P>::from_raw(data);
     let register = match register {
@@ -273,8 +273,8 @@ unsafe extern "C" fn port_rename<N, P>(
     data: *mut libc::c_void,
 ) -> libc::c_int
 where
-    N: 'static + Send + NotificationHandler,
-    P: 'static + Send + ProcessHandler,
+    N: 'static + Send + Sync + NotificationHandler,
+    P: 'static + Send + Sync + ProcessHandler,
 {
     let ctx = CallbackContext::<N, P>::from_raw(data);
     let old_name = ffi::CStr::from_ptr(old_name).to_str().unwrap();
@@ -290,8 +290,8 @@ unsafe extern "C" fn port_connect<N, P>(
     connect: libc::c_int,
     data: *mut libc::c_void,
 ) where
-    N: 'static + Send + NotificationHandler,
-    P: 'static + Send + ProcessHandler,
+    N: 'static + Send + Sync + NotificationHandler,
+    P: 'static + Send + Sync + ProcessHandler,
 {
     let ctx = CallbackContext::<N, P>::from_raw(data);
     let are_connected = match connect {
@@ -304,26 +304,26 @@ unsafe extern "C" fn port_connect<N, P>(
 
 unsafe extern "C" fn graph_order<N, P>(data: *mut libc::c_void) -> libc::c_int
 where
-    N: 'static + Send + NotificationHandler,
-    P: 'static + Send + ProcessHandler,
+    N: 'static + Send + Sync + NotificationHandler,
+    P: 'static + Send + Sync + ProcessHandler,
 {
     let ctx = CallbackContext::<N, P>::from_raw(data);
     ctx.notification.graph_reorder(&ctx.client).to_ffi()
 }
 
 unsafe extern "C" fn xrun<N, P>(data: *mut libc::c_void) -> libc::c_int
 where
-    N: 'static + Send + NotificationHandler,
-    P: 'static + Send + ProcessHandler,
+    N: 'static + Send + Sync + NotificationHandler,
+    P: 'static + Send + Sync + ProcessHandler,
 {
     let ctx = CallbackContext::<N, P>::from_raw(data);
     ctx.notification.xrun(&ctx.client).to_ffi()
 }
 
 unsafe extern "C" fn latency<N, P>(mode: j::jack_latency_callback_mode_t, data: *mut libc::c_void)
 where
-    N: 'static + Send + NotificationHandler,
-    P: 'static + Send + ProcessHandler,
+    N: 'static + Send + Sync + NotificationHandler,
+    P: 'static + Send + Sync + ProcessHandler,
 {
     let ctx = CallbackContext::<N, P>::from_raw(data);
     let mode = match mode {
@@ -361,8 +361,8 @@ pub struct CallbackContext<N, P> {
 
 impl<N, P> CallbackContext<N, P>
 where
-    N: 'static + Send + NotificationHandler,
-    P: 'static + Send + ProcessHandler,
+    N: 'static + Send + Sync + NotificationHandler,
+    P: 'static + Send + Sync + ProcessHandler,
 {
     pub unsafe fn from_raw<'a>(ptr: *mut libc::c_void) -> &'a mut CallbackContext<N, P> {
         debug_assert!(!ptr.is_null());

src/client/client.rs

@@ -70,8 +70,8 @@ impl Client {
         process_handler: P,
     ) -> Result<AsyncClient<N, P>, Error>
     where
-        N: 'static + Send + NotificationHandler,
-        P: 'static + Send + ProcessHandler,
+        N: 'static + Send + Sync + NotificationHandler,
+        P: 'static + Send + Sync + ProcessHandler,
     {
         AsyncClient::new(self, notification_handler, process_handler)
     }

src/lib.rs

@@ -37,6 +37,8 @@ extern crate bitflags;
 extern crate jack_sys;
 #[macro_use]
 extern crate lazy_static;
+#[cfg(test)]
+extern crate crossbeam_channel;
 extern crate libc;
 
 pub use client::{

src/port/audio.rs

@@ -99,7 +99,7 @@ impl Port<AudioOut> {
 
 #[cfg(test)]
 mod test {
-    use std::sync::mpsc::channel;
+    use crossbeam_channel::bounded;
 
     use super::*;
     use Client;
@@ -118,7 +118,7 @@ mod test {
         let in_b = c.register_port("ib", AudioIn::default()).unwrap();
         let mut out_a = c.register_port("oa", AudioOut::default()).unwrap();
         let mut out_b = c.register_port("ob", AudioOut::default()).unwrap();
-        let (signal_succeed, did_succeed) = channel();
+        let (signal_succeed, did_succeed) = bounded(1_000);
         let process_callback = move |_: &Client, ps: &ProcessScope| -> Control {
             let exp_a = 0.312_443;
             let exp_b = -0.612_120;

src/port/midi.rs

@@ -229,11 +229,11 @@ mod test {
     use client::Client;
     use client::ClosureProcessHandler;
     use client::ProcessHandler;
+    use crossbeam_channel::bounded;
     use jack_enums::Control;
     use primitive_types::Frames;
     use std::iter::Iterator;
     use std::sync::atomic::{AtomicUsize, Ordering};
-    use std::sync::mpsc::channel;
     use std::sync::Mutex;
     use std::{thread, time};
     use ClientOptions;
@@ -327,7 +327,7 @@ mod test {
         let mut out_b = c.register_port("ob", MidiOut::default()).unwrap();
 
         // set callback routine
-        let (signal_succeed, did_succeed) = channel();
+        let (signal_succeed, did_succeed) = bounded(1_000);
         let process_callback = move |_: &Client, ps: &ProcessScope| -> Control {
             let exp_a = RawMidi {
                 time: 0,