在旧版本的mio文档中,我找到了mio::channel,它似乎被用来创建一个实现EventedFd
的通道,可以注册到Poll
。我还在reddit上看到,这被更改为其他东西,但我无法弄清楚注册频道的新方式是什么。
到目前为止,使用 mio 等待频道(std::sync::mpsc
或其他东西(的当前方式是什么?我在谷歌上能找到的只是指向旧版本的mio文档的链接。谢谢!
当我遇到同样的问题时,我希望可以我提供一些代码以使问题更容易理解:
use std::io;
use std::net::*; //{TcpListener,TcpStream,IpAddr,Ipv4Addr,SocketAddr};
use std::thread::*;
use std::sync::mpsc::*; //{Receiver,SyncSender,Sender,channel,sync_channel};
fn poll_quit( rx : &Receiver::<u8> ) -> bool {
match rx.try_recv() {
Ok(_) => true,
Err(TryRecvError::Empty) => false,
Err(TryRecvError::Disconnected) => true
}
}
fn guard_external_tcp_port( rx : Receiver::<u8>) -> () {
let listener = TcpListener::bind("127.0.0.1:8384").expect("tcp guard - bind failed!");
listener.set_nonblocking(true).expect("cannot set tcp listener to nonblocking!");
while false == poll_quit(&rx) {
match listener.accept() {
Ok((_s,pa)) => {
println!("tcp client connected: {} - closing it down."
, pa);
}
Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
//wait_for_fd();
continue;
}
Err(e) => panic!("encountered IO error: {}", e),
}
}
}
fn start_tcpguard() -> (JoinHandle<()>,SyncSender::<u8>) {
let (tx,rx) = sync_channel::<u8>(0);
let thrd = spawn(move || {
guard_external_tcp_port(rx);
});
(thrd,tx)
}
到目前为止,我试图在没有 mio 的情况下做,但问题很明显:您希望阻止,直到关闭通道消息到达或侦听器套接字收到信号。有足够的动力切换到mio。除非 mio 无法等待Receiver
.
现在还有另一个 mio 板条箱,它具有与通道相关的功能:
朔朔-额外
因此,您所要做的就是将该板条箱添加到您的 cargo.toml [dependencies]
.
然后,在您的main.rs
中,添加extern crate mio_extras
,您应该准备好使用它。
因此,如果我将代码片段添加到问题中,它最终看起来像这样:
extern crate mio;
extern crate mio_extras;
use std::io;
use mio::*;
use mio::net::*;
use mio_extras::channel::*;
use std::thread::*;
fn guard_external_tcp_port( rx : mio_extras::channel::Receiver::<u8>) -> () {
let poll = Poll::new().expect("could not create a new Poll instance.");
const QUIT : Token = Token(0);
const CONNECT : Token = Token(1);
let mut events = Events::with_capacity(5);
poll.register(&rx,QUIT,Ready::readable(),PollOpt::level() ).expect("could not register channel Receiver.");
let addr : std::net::SocketAddr = "127.0.0.1:8384".parse().unwrap();
let listener = TcpListener::bind(&addr).expect("tcp guard - bind failed!");
poll.register(&listener,CONNECT,Ready::all(),PollOpt::edge()).expect("could not register our listening socket.");
let mut running : bool = true;
//listener.set_nonblocking(true).expect("cannot set tcp listener to nonblocking!");
while running {
let _nevents = poll.poll(&mut events,None).unwrap();
println!("poll returned!");
for event in &events {
match event.token() {
QUIT => running = false,
CONNECT => {
match listener.accept() {
Ok((_s,pa)) => {
println!("tcp client connected: {} - closing it down."
, pa);
}
Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
//wait_for_fd();
continue;
}
Err(e) => panic!("encountered IO error: {}", e),
}
},
Token(_) => continue
}
}
}
}
fn start_tcpguard() -> (JoinHandle<()>,Sender::<u8>) {
let (tx,rx) = mio_extras::channel::channel::<u8>();
let thrd = spawn(move || {
guard_external_tcp_port(rx);
});
(thrd,tx)
}
我也尝试了sync_channel(0)
版本,但它无法按预期工作。上面代码中的版本虽然适用于接收关闭事件的轮询。