Call process function with the trait instance

Author: jonnor

examples/repeat.rs

@@ -18,12 +18,14 @@ impl ParticipantTrait for Repeat {
 
 }
 
-fn process(input: Vec<u8>) -> Result<Vec<u8>, Vec<u8>> {
+unsafe impl Sync for Repeat {} // FIXME: figure out how to avoid
+
+fn process(r: &ParticipantTrait, input: Vec<u8>) -> Result<Vec<u8>, Vec<u8>> {
     println!("repeat process():");
     return Ok(input);
 }
 
 fn main() {
-    let r = Repeat { state: None };
+    static r: Repeat = Repeat { state: None };
     msgflo::participant::main(&r, process);
 }

src/participant.rs

@@ -64,18 +64,16 @@ impl InfoBuilder {
     }
 }
 
-
-type SendFunction = fn(String, Vec<u8>);
-pub type ProcessFunction = fn(Vec<u8>) -> Result<Vec<u8>, Vec<u8>>;
-
-pub trait ParticipantTrait {
+pub trait ParticipantTrait : Sync {
     fn info(&self) -> Info;
 //    fn process(&self, Vec<u8>) -> Result<Vec<u8>, Vec<u8>>;
 }
+pub type ProcessFunction = fn(&ParticipantTrait, Vec<u8>) -> Result<Vec<u8>, Vec<u8>>;
 
 struct Participant {
     info: Info,
-    process: ProcessFunction
+    process: ProcessFunction,
+    _trait: & 'static ParticipantTrait,
 }
 
 struct Connection {
@@ -105,10 +103,15 @@ fn send_discovery(channel: &mut Channel, info: &Info) {
 
 struct PortConsumer {
     process: ProcessFunction,
+    participant: & 'static ParticipantTrait,
     portname: String,
     outqueue: String, // FIXME: allow sending on any port, also multiple times
 }
 
+//unsafe impl Sync for PortConsumer {}
+//unsafe impl Send for ParticipantTrait {}
+//unsafe impl Sync for ParticipantTrait {}
+
 fn send_out(channel: &mut Channel, exchange: String, data: Vec<u8>) {
 
     let routing_key = "".to_string();
@@ -128,7 +131,7 @@ impl Consumer for PortConsumer {
 
         debug!("calling process()");
         let f = self.process;
-        let res = f(body);
+        let res = f(self.participant, body);
         debug!("process() returned");
 
         if res.is_ok() {
@@ -149,6 +152,7 @@ fn setup_inport(participant: &Participant, port: &Port, connection: &mut Connect
 
     let consumer = PortConsumer {
         process: participant.process,
+        participant: participant._trait,
         portname: port.id.to_string(),
         outqueue: participant.info.outports[0].queue.to_string(),
     };
@@ -271,15 +275,15 @@ fn parse(options: &mut Options) {
 // XXX: seems rust-amqp makes program hangs forever if error occurs / channel is borked?
 // TODO: pass port info in/out of process()
 // TODO: nicer way to declare ports? ideally they are enums not stringly typed?
-pub fn main(orig: &ParticipantTrait, func: ProcessFunction) {
+pub fn main(orig: & 'static ParticipantTrait, func: ProcessFunction) {
 
     let mut options = Options { .. Default::default() };
     parse(&mut options);
 
     let i : Info = orig.info();
     let info = normalize_info(&i, &options);
 
-    let p = Participant { info: info, process: func }; // XXX: hack
+    let p = Participant { info: info, process: func, _trait: orig }; // XXX: hack
 
     let mut c = start_participant(&p, &options);
     println!("{}({}) started", &p.info.role, &p.info.component);