pub fn process_messages(&mut self) -> Result<(), String> {
        // XXX TODO: With some reasonable rate limiting (like N message per pass)
        // Read some messages from the upstream
        // Handle each request
        match self.protocol.get_message(&mut self.stream, &mut self.buffer) {
            Ok(rpc_msg) => {
                match rpc_msg {
                    Some(message) => {
                        trace!("Worker {} - Got Message: {:?}", self.uuid(), message);
                        // let v: Value = serde_json::from_str(&message).unwrap();
                        let req: RpcRequest = match serde_json::from_str(&message) {
                            Ok(r) => r,
                            Err(e) => {
                                // Do we want to diconnect the user for invalid RPC message ???
                                self.error = true;
                                debug!("Worker {} - Got Invalid Message", self.uuid());
                                // XXX TODO: Invalid request
                                return Err(e.to_string());
                            }
                        };
                        trace!(
                            "Worker {} - Received request type: {}",
                            self.uuid(),
                            req.method
                        );
                        // Add this request id to the queue
                        self.request_ids.add(req.id.clone());
                        match req.method.as_str() {
                            "login" => {
                                debug!("Worker {} - Accepting Login request", self.uuid());
                                if self.user_id != 0 {
                                    // dont log in again, just say ok
                                    debug!("User already logged in: {}", self.user_id.clone());
                                    self.send_ok(req.method);
                                    return Ok(());
                                }
                                let params: Value = match req.params {
                                    Some(p) => p,
                                    None => {
                                        self.error = true;
                                        debug!("Worker {} - Missing Login request parameters", self.uuid());
                                        return self.send_err(
                                            "login".to_string(),
                                            "Missing Login request parameters".to_string(),
                                            -32500,
                                        );
                                        // XXX TODO: Invalid request
                                        //return Err("Invalid Login request".to_string());
                                    }
                                };
                                let login_params: LoginParams = match serde_json::from_value(params) {
                                    Ok(p) => p,
                                    Err(e) => {
                                        self.error = true;
                                        debug!("Worker {} - Invalid Login request parameters", self.uuid());
                                        return self.send_err(
                                            "login".to_string(),
                                            "Invalid Login request parameters".to_string(),
                                            -32500,
                                        );
                                        // XXX TODO: Invalid request
                                        //return Err(e.to_string());
                                    }
                                };
                                // Call do_login()
                                match self.do_login(login_params) {
                                    Ok(_) => {
                                        // We accepted the login, send ok result
					                    self.authenticated = true;
                                        self.needs_job = false; // not until requested
                                        self.status = WorkerStatus::new(self.uuid());
                                        self.send_ok(req.method);
                                    },
                                    Err(e) => {
                                        return self.send_err(
                                            "login".to_string(),
                                            e,
                                            -32500,
                                        );
                                    }
                                }
                            }
                            "getjobtemplate" => {
                                trace!("Worker {} - Accepting request for job", self.uuid());
                                self.needs_job = true;
                                self.requested_job = true;
                            }
                            "submit" => {
                                trace!("Worker {} - Accepting share", self.uuid());
                                match serde_json::from_value(req.params.unwrap()) {
                                    Result::Ok(share) => {
                           			    self.shares.push(share);
                                    },
                                    Result::Err(err) => { }
                                };
                            }
                            "status" => {
                                trace!("Worker {} - Accepting status request", self.uuid());
                                let status = self.status.clone();
                                self.send_status(status);
                            }
                            "keepalive" => {
                                trace!("Worker {} - Accepting keepalive request", self.uuid());
                                self.send_ok(req.method);
                            }
                            _ => {
                                warn!(
                                    "Worker {} - Unknown request: {}",
                                    self.uuid(),
                                    req.method.as_str()
                                );
                                self.error = true;
                                return Err("Unknown request".to_string());
                            }
                        };
                    }
                    None => {} // Not an error, just no messages for us right now
                }
            }
            Err(e) => {
                error!(
                    "Worker {} - Error reading message: {}",
                    self.uuid(),
                    e.to_string()
                );
                self.error = true;
                return Err(e.to_string());
            }
        }
        return Ok(());
    }
/// Get and process messages from the connected worker
// Method to handle requests from the downstream worker
Get and process messages from the connected worker
Method to handle requests from the downstream worker
Get and process messages from the connected worker Method to handle requests from the downstream worker
pub fn process_messages(&mut self) -> Result<(), String> {
        // XXX TODO: With some reasonable rate limiting (like N message per pass)
        // Read some messages from the upstream
        // Handle each request
        match self.protocol.get_message(&mut self.stream, &mut self.buffer) {
            Ok(rpc_msg) => {
                match rpc_msg {
                    Some(message) => {
                        trace!("Worker {} - Got Message: {:?}", self.uuid(), message);
                        // let v: Value = serde_json::from_str(&message).unwrap();
                        let req: RpcRequest = match serde_json::from_str(&message) {
                            Ok(r) => r,
                            Err(e) => {
                                // Do we want to diconnect the user for invalid RPC message ???
                                self.error = true;
                                debug!("Worker {} - Got Invalid Message", self.uuid());
                                // XXX TODO: Invalid request
                                return Err(e.to_string());
                            }
                        };
                        trace!(
                            "Worker {} - Received request type: {}",
                            self.uuid(),
                            req.method
                        );
                        // Add this request id to the queue
                        self.request_ids.add(req.id.clone());
                        match req.method.as_str() {
                            "login" => {
                                debug!("Worker {} - Accepting Login request", self.uuid());
                                if self.user_id != 0 {
                                    // dont log in again, just say ok
                                    debug!("User already logged in: {}", self.user_id.clone());
                                    self.send_ok(req.method);
                                    return Ok(());
                                }
                                let params: Value = match req.params {
                                    Some(p) => p,
                                    None => {
                                        self.error = true;
                                        debug!("Worker {} - Missing Login request parameters", self.uuid());
                                        return self.send_err(
                                            "login".to_string(),
                                            "Missing Login request parameters".to_string(),
                                            -32500,
                                        );
                                        // XXX TODO: Invalid request
                                        //return Err("Invalid Login request".to_string());
                                    }
                                };
                                let login_params: LoginParams = match serde_json::from_value(params) {
                                    Ok(p) => p,
                                    Err(e) => {
                                        self.error = true;
                                        debug!("Worker {} - Invalid Login request parameters", self.uuid());
                                        return self.send_err(
                                            "login".to_string(),
                                            "Invalid Login request parameters".to_string(),
                                            -32500,
                                        );
                                        // XXX TODO: Invalid request
                                        //return Err(e.to_string());
                                    }
                                };
                                // Call do_login()
                                match self.do_login(login_params) {
                                    Ok(_) => {
                                        // We accepted the login, send ok result
					                    self.authenticated = true;
                                        self.needs_job = false; // not until requested
                                        self.status = WorkerStatus::new(self.uuid());
                                        self.send_ok(req.method);
                                    },
                                    Err(e) => {
                                        return self.send_err(
                                            "login".to_string(),
                                            e,
                                            -32500,
                                        );
                                    }
                                }
                            }
                            "getjobtemplate" => {
                                trace!("Worker {} - Accepting request for job", self.uuid());
                                self.needs_job = true;
                                self.requested_job = true;
                            }
                            "submit" => {
                                trace!("Worker {} - Accepting share", self.uuid());
                                match serde_json::from_value(req.params.unwrap()) {
                                    Result::Ok(share) => {
                           			    self.shares.push(share);
                                    },
                                    Result::Err(err) => { }
                                };
                            }
                            "status" => {
                                trace!("Worker {} - Accepting status request", self.uuid());
                                let status = self.status.clone();
                                self.send_status(status);
                            }
                            "keepalive" => {
                                trace!("Worker {} - Accepting keepalive request", self.uuid());
                                self.send_ok(req.method);
                            }
                            _ => {
                                warn!(
                                    "Worker {} - Unknown request: {}",
                                    self.uuid(),
                                    req.method.as_str()
                                );
                                self.error = true;
                                return Err("Unknown request".to_string());
                            }
                        };
                    }
                    None => {} // Not an error, just no messages for us right now
                }
            }
            Err(e) => {
                error!(
                    "Worker {} - Error reading message: {}",
                    self.uuid(),
                    e.to_string()
                );
                self.error = true;
                return Err(e.to_string());
            }
        }
        return Ok(());
    }
pub fn process_messages ( & mut self ) -> Result < ( ) , String > { // XXX TODO: With some reasonable rate limiting (like N message per pass) // Read some messages from the upstream // Handle each request match self . protocol . get_message ( & mut self . stream , & mut self . buffer ) { Ok ( rpc_msg ) => { match rpc_msg { Some ( message ) => { trace ! ( " " self uuid ( ) message ) ; // let v: Value = serde_json::from_str(&message).unwrap(); let req : RpcRequest = match serde_json :: from_str ( & message ) { Ok ( r ) => r , Err ( e ) => { // Do we want to diconnect the user for invalid RPC message ??? self . error = true ; debug ! ( " " self uuid ( ) ) ; // XXX TODO: Invalid request return Err ( e . to_string ( ) ) ; } } ; trace ! ( " " self uuid ( ) req method ) ; // Add this request id to the queue self . request_ids . add ( req . id . clone ( ) ) ; match req . method . as_str ( ) { " " => { debug ! ( " " self uuid ( ) ) ; if self . user_id != 0 { // dont log in again, just say ok debug ! ( " " self user_id clone ( ) ) ; self . send_ok ( req . method ) ; return Ok ( ( ) ) ; } let params : Value = match req . params { Some ( p ) => p , None => { self . error = true ; debug ! ( " " self uuid ( ) ) ; return self . send_err ( " " . to_string ( ) , " " . to_string ( ) , - 32500 , ) ; // XXX TODO: Invalid request //return Err("Invalid Login request".to_string()); } } ; let login_params : LoginParams = match serde_json :: from_value ( params ) { Ok ( p ) => p , Err ( e ) => { self . error = true ; debug ! ( " " self uuid ( ) ) ; return self . send_err ( " " . to_string ( ) , " " . to_string ( ) , - 32500 , ) ; // XXX TODO: Invalid request //return Err(e.to_string()); } } ; // Call do_login() match self . do_login ( login_params ) { Ok ( _ ) => { // We accepted the login, send ok result self . authenticated = true ; self . needs_job = false ; // not until requested self . status = WorkerStatus :: new ( self . uuid ( ) ) ; self . send_ok ( req . method ) ; } , Err ( e ) => { return self . send_err ( " " . to_string ( ) , e , - 32500 , ) ; } } } " " => { trace ! ( " " self uuid ( ) ) ; self . needs_job = true ; self . requested_job = true ; } " " => { trace ! ( " " self uuid ( ) ) ; match serde_json :: from_value ( req . params . unwrap ( ) ) { Result :: Ok ( share ) => { self . shares . push ( share ) ; } , Result :: Err ( err ) => { } } ; } " " => { trace ! ( " " self uuid ( ) ) ; let status = self . status . clone ( ) ; self . send_status ( status ) ; } " " => { trace ! ( " " self uuid ( ) ) ; self . send_ok ( req . method ) ; } _ => { warn ! ( " " self uuid ( ) req method as_str ( ) ) ; self . error = true ; return Err ( " " . to_string ( ) ) ; } } ; } None => { } // Not an error, just no messages for us right now } } Err ( e ) => { error ! ( " " self uuid ( ) e to_string ( ) ) ; self . error = true ; return Err ( e . to_string ( ) ) ; } } return Ok ( ( ) ) ; }
Get and process messages from the connected worker
Method to handle requests from the downstream worker
Get and process messages from the connected worker Method to handle requests from the downstream worker
&mut self