Merge pull request #27 from chef/wonder-upstream

Merged change 17836373-63dd-4ba4-9914-607f2a185b41

From review branch wonder-upstream into master

Signed-off-by: fnichol <fnichol@chef.io>

src/bldr/util/signals.rs

@@ -23,10 +23,9 @@
 
 use std::sync::{Once, ONCE_INIT};
 use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering, ATOMIC_USIZE_INIT, ATOMIC_BOOL_INIT};
-use std::sync::mpsc::Sender;
 
 use wonder::actor;
-use wonder::actor::{HandleResult, InitResult, StopReason};
+use wonder::actor::{ActorSender, HandleResult, InitResult, StopReason};
 
 use error::{BldrResult, BldrError};
 
@@ -91,7 +90,7 @@ impl actor::GenServer for SignalNotifier {
     type S = ();
     type E = BldrError;
 
-    fn init(&self, _tx: &Sender<actor::Message<Self::T>>, _: &mut Self::S) -> InitResult<Self::E> {
+    fn init(&self, _tx: &ActorSender<Self::T>, _: &mut Self::S) -> InitResult<Self::E> {
         unsafe {
             INIT.call_once(|| {
                 self::set_signal_handlers();
@@ -105,14 +104,14 @@ impl actor::GenServer for SignalNotifier {
         Ok(Some(TIMEOUT_MS))
     }
 
-    fn handle_call(&self, message: Self::T, _: &Sender<actor::Message<Self::T>>, _: &Sender<actor::Message<Self::T>>, _: &mut Self::S) -> HandleResult<Self::T> {
+    fn handle_call(&self, message: Self::T, _: &ActorSender<Self::T>, _: &ActorSender<Self::T>, _: &mut Self::S) -> HandleResult<Self::T> {
         match message {
             Message::Stop => HandleResult::Stop(StopReason::Normal, None),
             msg => HandleResult::Stop(StopReason::Fatal(format!("unexpected call message: {:?}", msg)), None),
         }
     }
 
-    fn handle_timeout(&self, tx: &Sender<actor::Message<Self::T>>, _: &mut Self::S) -> HandleResult<Self::T> {
+    fn handle_timeout(&self, tx: &ActorSender<Self::T>, _: &ActorSender<Self::T>, _: &mut Self::S) -> HandleResult<Self::T> {
         unsafe {
             if CAUGHT.load(Ordering::SeqCst) {
                 match SIGNAL.load(Ordering::SeqCst) {
@@ -133,7 +132,7 @@ impl actor::GenServer for SignalNotifier {
     }
 }
 
-fn send_signal(tx: &Sender<actor::Message<Message>>, signal: Signal) {
+fn send_signal(tx: &ActorSender<Message>, signal: Signal) {
     actor::cast(tx, Message::Signal(signal)).unwrap();
 }
 

vendor/wonder/src/actor.rs

@@ -99,7 +99,7 @@ impl<A: GenServer> Builder<A> {
             loop {
                 if let Some(go_time) = timeout {
                     if go_time >= SteadyTime::now() {
-                        match self.spec.handle_timeout(&itx, &mut state) {
+                        match self.spec.handle_timeout(&otx, &itx, &mut state) {
                             HandleResult::Stop(reason, None) => return shutdown(reason, None, &otx),
                             HandleResult::NoReply(Some(0)) => {
                                 set_timeout(0, &mut timeout);
@@ -132,7 +132,7 @@ impl<A: GenServer> Builder<A> {
                         }
                     },
                     Ok(Message::Cast(msg)) => {
-                        match self.spec.handle_cast(msg, &itx, &mut state) {
+                        match self.spec.handle_cast(msg, &otx, &itx, &mut state) {
                             HandleResult::Stop(reason, reply) => return shutdown(reason, reply, &otx),
                             HandleResult::NoReply(new_timeout) => {
                                 if let Some(ms) = new_timeout {
@@ -146,6 +146,14 @@ impl<A: GenServer> Builder<A> {
                     Err(mpsc::TryRecvError::Disconnected) => { break; },
                     Err(mpsc::TryRecvError::Empty) => { },
                 }
+                // This is absolutely the wrong solution. I need to park the thread or call
+                // recv instead of try_recv and schedule the timeout mechanism another way.
+                // This is a quick and dirty workaround that should be short lived while the API
+                // stabilizes and is leveraged in our other applications.
+                //
+                // I'm so sorry for doing this.
+                //      - Jamie
+                thread::sleep_ms(30)
             }
             Ok(())
         }).unwrap();
@@ -209,13 +217,15 @@ pub trait GenServer : Send + 'static {
 
     fn init(&self, _tx: &ActorSender<Self::T>, state: &mut Self::S) -> InitResult<Self::E>;
 
-    fn handle_call(&self, _message: Self::T, _tx: &ActorSender<Self::T>, _caller: &ActorSender<Self::T>, _state: &mut Self::S) -> HandleResult<Self::T> {
-        panic!("handle_call callback not implemented");
+    fn handle_call(&self, message: Self::T, _tx: &ActorSender<Self::T>, _me: &ActorSender<Self::T>, _state: &mut Self::S) -> HandleResult<Self::T> {
+        panic!("handle_call callback not implemented; received: {:?}", message);
     }
-    fn handle_cast(&self, _message: Self::T, _tx: &ActorSender<Self::T>, _state: &mut Self::S) -> HandleResult<Self::T> {
-        panic!("handle_cast callback not implemented");
+
+    fn handle_cast(&self, message: Self::T, _tx: &ActorSender<Self::T>, _me: &ActorSender<Self::T>, _state: &mut Self::S) -> HandleResult<Self::T> {
+        panic!("handle_cast callback not implemented; received: {:?}", message);
     }
-    fn handle_timeout(&self, _tx: &ActorSender<Self::T>, _state: &mut Self::S) -> HandleResult<Self::T> {
+
+    fn handle_timeout(&self, _tx: &ActorSender<Self::T>, _me: &ActorSender<Self::T>, _state: &mut Self::S) -> HandleResult<Self::T> {
         HandleResult::NoReply(None)
     }
 }
@@ -294,7 +304,7 @@ mod tests {
             }
         }
 
-        fn handle_cast(&self, msg: Self::T, _: &ActorSender<Self::T>, state: &mut Self::S) -> HandleResult<Self::T> {
+        fn handle_cast(&self, msg: Self::T, _: &ActorSender<Self::T>, _: &ActorSender<Self::T>, state: &mut Self::S) -> HandleResult<Self::T> {
             match msg {
                 MyMessage::SetState(value) => {
                     state.initialized = value;