-1

Hyper.rs を使用した Web サービス リクエストにより、センサー値を記録するスレッドを開始したいと考えています。次に、別のリクエストでスレッドを停止し、記録されたすべてのデータを取得したいと思います。

のおかげでスレッドが既に実行されているかどうかはわかっているMutexので、 に入れようとしSender<T>ましたが、「借用したコンテンツから移動できません」という理由でMutex、この値を使用できません。Mutex

これは私のコードです：

extern crate futures;
extern crate hyper;

use futures::future::Future;
use hyper::{Method, StatusCode};
use hyper::server::{Http, Request, Response, Service};
use std::sync::Mutex;
use std::sync::mpsc::{channel, Sender};
use std::thread;
use std::thread::ThreadId;

fn main() {
    let addr = "127.0.0.1:3000".parse().unwrap();
    let server = Http::new().bind(&addr, || Ok(Sensor::new())).unwrap();
    server.run().unwrap();
}

struct Sensor {
    current_thread: Mutex<Option<ThreadId>>,
    current_tx: Mutex<Option<Sender<String>>>,
}

impl Sensor {
    pub fn new() -> Sensor {
        Sensor {
            current_thread: Mutex::new(None),
            current_tx: Mutex::new(None),
        }
    }

    fn start_sensors(&self) -> <Sensor as Service>::Future {
        let mut current_thread = self.current_thread.lock().unwrap();
        match *current_thread {
            Some(id) => {
                self.response_with_body(format!("thread with id {:?} is already running", id))
            }
            None => {
                let (tx, rx) = channel();
                let builder = thread::Builder::new();
                let join_handle = builder
                    .spawn(|| {
                        for received in rx {
                            println!("Got: {}", received);
                            if received == "stop" {
                                break;
                            }
                        }
                    })
                    .unwrap();
                let thread = join_handle.thread();

                let mut current_tx = self.current_tx.lock().unwrap();
                *current_tx = Some(tx);

                let message = format!("thread id: {:?}", &thread.id());
                *current_thread = Some(thread.id());

                self.response_with_body(message)
            }
        }
    }

    fn stop_sensors(&self) -> <Sensor as Service>::Future {
        let mut current_tx = self.current_tx.lock().unwrap();
        match *current_tx {
            Some(tx) => {
                let mut current_thread = self.current_thread.lock().unwrap();

                // just to trying to communicate with the thread
                let vals = vec![String::from("hi"), String::from("stop")];

                for val in vals {
                    tx.send(val).unwrap();
                }

                *current_tx = None;
                *current_thread = None;
                Box::new(futures::future::ok(Response::new()))
            }
            None => self.response_with_body(format!("No thread running")),
        }
    }

    fn response_with_body(&self, body: String) -> <Sensor as Service>::Future {
        Box::new(futures::future::ok(Response::new().with_body(body)))
    }
}

impl Service for Sensor {
    type Request = Request;
    type Response = Response;
    type Error = hyper::Error;
    type Future = Box<Future<Item = Self::Response, Error = Self::Error>>;

    fn call(&self, req: Request) -> Self::Future {
        //let mut current_thread: Option<ThreadId> = None;
        match (req.method(), req.path()) {
            (&Method::Get, "/sensors/start") => self.start_sensors(),
            (&Method::Get, "/sensors/stop") => self.stop_sensors(),
            _ => Box::new(futures::future::ok(
                Response::new().with_status(StatusCode::NotFound),
            )),
        }
    }
}

error[E0507]: cannot move out of borrowed content
  --> src/main.rs:65:15
   |
65 |         match *current_tx {
   |               ^^^^^^^^^^^ cannot move out of borrowed content
66 |             Some(tx) => {
   |                  -- hint: to prevent move, use `ref tx` or `ref mut tx`

どうすればいいですか？

4

1 に答える 1

0

少し変更するだけで、Noneなんとか入れることができます：*current_tx

fn stop_sensors(&self) -> <Sensor as Service>::Future {
    let mut current_tx = self.current_tx.lock().unwrap();

    match std::mem::replace(&mut *current_tx, None) {
        Some(tx) => {
            let mut current_thread = self.current_thread.lock().unwrap();

            // just to trying to communicate with the thread
            let vals = vec![String::from("hi"), String::from("stop")];

            for val in vals {
                tx.send(val).unwrap();
            }

            // *current_tx = None;
            *current_thread = None;
            Box::new(futures::future::ok(Response::new()))
        }
        None => self.response_with_body(format!("No thread running")),
    }
}

std::mem::replaceNone を入れて*current_tx、古い値を返します。

于 2018-03-01T09:35:47.857 に答える