如何将take_while与futures::Stream一起使用？
在使用Rust编程语言中，可以将take_while与futures::Stream一起使用来实现一种在满足特定条件之前，从流中获取特定数量的元素的功能。
take_while是一个函数，它接受一个闭包作为参数，该闭包定义了一个条件，只要该条件为真，take_while就会从流中获取下一个元素。一旦条件为假，take_while将停止获取元素，并结束流。
以下是一个示例代码，展示如何将take_while与futures::Stream一起使用：
use futures::stream::{self, StreamExt};
#[tokio::main]
async fn main() {
let numbers = stream::iter(0..10); // 创建一个流，包含0到9的数字
let even_numbers = numbers
.take_while(|&x| x % 2 == 0) // 只获取偶数，直到遇到奇数
.collect::<Vec<_>>()
.await;
println!("{:?}", even_numbers);
}在上面的代码中，我们创建了一个包含数字0到9的流。然后，我们使用take_while函数来只获取偶数，直到遇到奇数为止。最后，我们使用.collect()方法将获取的偶数收集到一个向量中，并使用await等待结果。最后，我们打印出收集到的偶数。
上述代码的输出结果将是[0, 2, 4, 6, 8]，这是因为take_while只获取偶数，直到遇到奇数3为止。
需要注意的是，以上示例代码中使用了futures::StreamExt trait，它为流提供了各种有用的方法，如take_while、collect等。
推荐的腾讯云产品链接地址：
相关·内容
我正在尝试理解我应该为futures::Stream; crate (0.1.25)的take_while()使用什么语法。下面是一段代码(on playground)： use futures::{stream, Stream}; // 0.1.25
fn into_many(i: i32) -> impl Stream<Item = i32, Error = ()> {
stream::iter_ok(0..i)
}
fn main() {
println!("start:");
let _ = into_many(10)
浏览 25提问于2019-02-24得票数 0
回答已采纳
为了异步关闭Receiver流，我希望能够沿着这些路线执行一些操作：
extern crate futures;
extern crate tokio;
use futures::future::lazy;
use futures::stream::AndThen;
use futures::sync::mpsc::Receiver;
use futures::{Future, Sink, Stream};
use std::sync::{Arc, Mutex};
use tokio::timer::{Delay, Interval};
fn main() {
tokio::ru
浏览 0提问于2018-12-23得票数 2
回答已采纳
1回答
我想要一个处理一些流的结构。所有流共享同一项。逻辑如下:我需要创建一个唯一的流，其中包含来自所有其他流的所有项。我还需要在“主”流中添加“新”流。我不在乎下一个项目来自哪个流。
为此，我看到有一个函数应该执行上面描述的逻辑。
pub struct WsPool {
merge: Arc<Mutex<SelectAll<Box<dyn Stream<Item=MyItem> + Send + 'static>>>>,
}
impl WsPool {
pub fn new() -> Self {
浏览 3提问于2021-10-10得票数 1
回答已采纳
1回答
如何同时抓取未知目的的分页网页？
、、、、
我正在尝试使用tokio异步运行时在Rust中编写一个web爬虫。我希望异步地获取/处理多个页面，但我也希望爬虫在到达末尾时停止(换句话说，如果没有什么可以抓取的东西)。到目前为止，我已经使用从作为Future提供的异步函数获得集体结果，但这显然需要程序事先知道要爬行的总页面。例如：
async fn fetch(_url: String) -> Result<String, ()> {
tokio::time::sleep(std::time::Duration::from_millis(100)).await;
Ok(String::from("
浏览 3提问于2021-09-24得票数 1
回答已采纳
2回答
我正在编写一个使用Rust从中抓取链接的程序。我正在使用hyper和futures。
我收集了到每个部分的链接，并使用stream::unfold构造对每个页面的访问：
// Returns the links scraped, and probably Uri to the next page.
fn process_body_and_return_next(body: Body) -> (Vec<String>, Option<Uri>) { ... }
// In main()
let mut core = Core::new().unwrap();
le
浏览 0提问于2018-02-14得票数 1
两个整数的序列，一个每次递减，另一个每隔三次递减，直到它们都达到它们的极限(149到145 & 153到150)。
(149, 153)
(148, 153)
(147, 153)
(146, 152)
(145, 152)
(145, 152)
(145, 151)
(145, 151)
(145, 151)
(145, 150)
在这个实现中有一个off-by-one错误，它生成除了最后一个条目之外的所有条目。
#include <iostream>
#include "../Streams-master/source/Stream.h"
#includ
浏览 0提问于2015-12-07得票数 0
我试图使用机箱在Rust中编写一个简单的TCP客户端。我的代码非常接近减去TLS：
extern crate futures;
extern crate tokio_core;
extern crate tokio_io;
use futures::Future;
use tokio_core::net::TcpStream;
use tokio_core::reactor::Core;
use tokio_io::io;
fn main() {
let mut core = Core::new().unwrap();
let handle = core.handle();
浏览 9提问于2017-10-07得票数 11
回答已采纳
我想要一个，这样我就可以做进一步的流处理。问题是我可能有一个迭代器或一个错误要处理。我觉得我对这件事很熟悉：
#[macro_use]
extern crate log;
extern crate futures; // 0.1.21
extern crate tokio;
use futures::prelude::*;
use futures::{future, stream};
use std::fmt::Debug;
use std::net::{SocketAddr, ToSocketAddrs};
fn resolve(addrs: impl ToSocketAddrs + D
浏览 3提问于2018-06-30得票数 4
回答已采纳
1回答
我想使用async函数逐步解析入站流，但是actix需要impl Future<Item = HttpResponse, Error = Error>作为返回值。
如何将async函数返回的未来转换为actix所需的内容？
我每晚使用Rust 1.39和Actix-Web1.0.7。
http_srv.rs：
use futures::compat::Stream01CompatExt;
use futures::future::{FutureExt, TryFutureExt};
use futures::stream::TryStreamExt;
use futures01::
浏览 0提问于2019-09-04得票数 3
回答已采纳
1回答
在我的中，我想异步地读取read。目前，由于的支持，它们被同步读取
self.feeds = self
.feeds
.iter()
.map(|f| f.read(&self.settings))
.collect::<Vec<Feed>>();
我想使代码异步，因为它将允许我更好地处理糟糕的web服务器响应。
据我所知，我可以使用一个Stream，它可以使用Vec创建，使用stream::from_iter(...)将代码转换成如下所示
self.feeds = stream::from_iter(self.feeds.ite
浏览 3提问于2019-12-08得票数 5
回答已采纳
我是Rust的新手，并且使用Tonic编写了一个简单的应用程序，该应用程序将在gRPC上传输一些值。这些值最初是作为BoxStream (Pin<Box<Stream>>)从外部库中获取的，而requires的API需要一些实现Stream的东西(当然Pin不需要)。 Tonic's streaming example使用ReceiverStream将mpsc频道转换为流，并派生出一个线程将值推入其中。这将要求流的生命周期为'static，这不是我的实际实现的选项，因为我的流的生命周期与返回它的类相关联。 最好的方式是提供一些实现流的东西，我可以从我的
浏览 67提问于2021-09-10得票数 0
回答已采纳
我在下面运行一些异步任务，需要等待直到它们全部完成。我不知道为什么，但是join()并没有强迫等待所有的任务，并且代码不需要等待就继续执行。连接流没有按预期工作的原因吗？
CompletableFutures列表只是一个映射supplyAsync的流。
List<Integer> items = Arrays.asList(1, 2, 3);
List<CompletableFuture<Integer>> futures = items
.stream()
.map(item -> C
浏览 3提问于2020-10-12得票数 2
回答已采纳
我正在使用FuturesUnordered将异步工作负载排队到多线程tokio运行程序上。这些期货回报各种不同的结果。为了区分它们，我将每个未来的结果映射到自定义的Event类型。
enum Event {
ResultTypeA {...},
ResultTypeB {...},
ResultTypeC {...},
ResultTypeD {...}
}
let pending_futures: FuturesUnordered<Pin<Box<dyn Future<Output = Event> + Send>>>
浏览 15提问于2022-11-06得票数 0
在这个简化的代码版本中，我希望有时执行标记行，有时不执行，可能会返回一个错误：
extern crate futures; // 0.1.26
extern crate hyper; // 0.12.25
use hyper::rt::{Future, Stream};
use std::str::FromStr;
struct MyStream {}
impl Stream for MyStream {
type Item = hyper::Uri;
type Error = ();
fn poll(&mut self) -> Result<
浏览 0提问于2018-09-03得票数 0
回答已采纳
我是生疏的，如果我使用的术语不正确，我很抱歉。也许我对疑问词的选择是错误的。
我在玩流，我需要在流元素之间有一些延迟。所以我写了这个：
use futures::stream;
use futures::StreamExt;
use tokio::time;
#[tokio::main]
async fn main() {
let mut stream = stream::iter(0..1000).then(|x| async move {
time::delay_for(std::time::Duration::from_millis(500)).await;
浏览 3提问于2020-09-22得票数 4
回答已采纳
我正试图用“铁锈”来包装我的未来，但我对这段代码感到困惑，它应该把到达rx的消息发送到sink
extern crate futures;
extern crate tokio_core;
extern crate websocket;
use websocket::message::OwnedMessage;
use websocket::server::InvalidConnection;
use websocket::async::Server;
use tokio_core::reactor::Core;
use futures::{Future, Sink, Stream};
u
浏览 2提问于2017-11-30得票数 1
回答已采纳
1回答
有一个async connect()函数如下所示。
use actix_web::client::Client;
use futures::compat::Future01CompatExt;
use futures::future::{FutureExt, TryFutureExt};
pub async fn connect() {
let request = Client::default().get("https://www.google.com").send();
// compat() converts a v0.1 futures::futur
浏览 0提问于2019-08-28得票数 4
回答已采纳
我正在尝试连接一个锈蚀的期货流(重复)到一个水槽(排水)。第一个方法工作正常(下面检查一下)，但是第二个带有.forward的方法无法编译(错误在内联中显示)。对我做错了什么有什么建议吗？
use futures::stream::StreamExt;
use futures::sink::SinkExt;
use futures::executor;
fn main() {
executor::block_on( async {
// works
let mut work = futures::stream::repeat(5);
l
浏览 1提问于2020-01-15得票数 3
回答已采纳
1回答
我有一个用Python语言实现的gRPC服务器，我正在从NodeJS调用一个RPC，但它给出了一个错误"Method not found“。当我使用python客户端调用时，请求成功。 stream_csv.proto syntax = "proto3";
package csv;
service Stream {
rpc csvToObject(CSVDataRequest) returns (stream CSVDataResponse) {};
rpc sayHello(HelloRequest) returns (HelloReply);
}
m
浏览 93提问于2020-10-18得票数 3
回答已采纳
2回答
我正在尝试使用async_std从网络接收UDP数据报。
有一个实现UdpSocket的，这个方法返回一个未来，但是我需要一个async_std::stream::Stream来提供一个UDP数据报流，因为它是一个更好的抽象。
我已经找到了完全满足我需要的，但它在目前的tokio版本中是不可用的。
一般来说，问题是如何将Future从给定的异步函数转换为Stream。
浏览 4提问于2020-01-30得票数 5
回答已采纳
我有以下设置：
use core::task::Poll;
use tokio::io::ReadBuf;
use core::task::Context;
use core::pin::Pin;
use std::error::Error;
use tokio::io::AsyncRead;
struct Dummy;
impl AsyncRead for Dummy {
fn poll_read(
self: Pin<&mut Self>,
cx: &mut Context<'_>,
浏览 2提问于2022-11-08得票数 0
我得把卡桑德拉升级到4倍。这是以前用cassandra 3.5编写的代码
protected <T> Stream<T> getAll(Stream<Statement> statements, Mapper<T> mapper) {
List<ResultSetFuture> futures = statements
.peek(p -> cassandraReads.inc())
.map(s -> session.session().e
浏览 2提问于2022-06-10得票数 0
回答已采纳
1回答
考虑以下最小流示例：
use futures::StreamExt;
use async_stream::stream;
#[tokio::main]
async fn main() {
let ticks = stream! {
yield Ok(0);
yield Err("");
};
futures::pin_mut!(ticks);
while let Some(x) = ticks.next().await {
println!("{:?}", x);
}
}
浏览 4提问于2022-06-15得票数 0
回答已采纳
1回答
服务器使用，客户端代码使用修改后的。当客户端连接到服务器时，它会永远阻止自己。
服务器：
extern crate futures;
extern crate futures_io;
extern crate futures_mio;
use std::net::SocketAddr;
use futures::Future;
use futures_io::{copy, TaskIo};
use futures::stream::Stream;
fn main() {
let addr = "127.0.0.1:8080".parse::<SocketAd
浏览 2提问于2016-08-20得票数 4
回答已采纳
1回答
当我尝试使用UnboundedSender和UnboundedReceiver而不使用tokio::run()时，我得到了一个错误：tokio::run()。
这是一个错误：
Press Ctrl-C to abort
thread 'receiver thread' panicked at 'no Task is currently running', src/libcore/option.rs:1166:5
stack backtrace:
0: backtrace::backtrace::libunwind::trace
a
浏览 3提问于2019-10-03得票数 1
回答已采纳
2回答
关于异步和流的生命周期
、、、、
我正在尝试创建一个函数，该函数使用Stream并在max_consecutive_fails连续失败时将其截断。然而，事情并没有很好地工作(E0495)。我将Stream%s更改为Iterator%s(并删除了async%s)，这很简单。为什么会发生这种情况？我如何重构这段代码(使其正常工作)？ use futures::stream::Stream;
pub fn max_fail<'a, T>(stream : impl Stream<Item = Option<T>> +'a , max_consecutive_fails: usiz
浏览 43提问于2020-05-04得票数 1
我有一段将futures::stream::Stream序列化为Write的代码。我希望这段代码在write_all失败时返回Err，但我看不到从for_each的回调中去掉它的方法。我找到了，但我不知道如何让它在我的情况下工作。
use std::io::Write;
use std::error::Error;
use futures::stream::StreamExt;
pub async fn download(url: &str, mut dest: impl Write) -> Result<(), Box<dyn Error>> {
浏览 0提问于2021-07-12得票数 0
我有以下设置
use futures::{
future,
stream::{self, Stream, FuturesUnordered},
};
use tokio;
fn foo(futures: FuturesUnordered<impl futures::Future<Output = std::io::Result<impl std::fmt::Binary>>>) {}
fn bar(futures: FuturesUnordered<impl futures::Future<Output = impl std:
浏览 13提问于2022-11-09得票数 0
回答已采纳
我有一个使用组合子的流，我需要运行它直到完成。我可以使用while循环或for_each组合器。它们都可以工作，但我认为一定有更好的方法。
看起来就像我要找的，尤其是，但我还不能理解如何使用它。
使用while循环
use futures::{StreamExt, TryStreamExt}; // 0.3.6
use tokio; // 0.3.0
#[tokio::main]
async fn main() {
let mut stream = Box::pin(
futures::stream::iter(0..20)
.map(foo)
浏览 0提问于2020-10-20得票数 0
这里有两个选择，只是有点混淆了哪一个是最好的。
选项1:
ExecutorService es = Executors.newFixedThreadPool(4);
List<Runnable> tasks = getTasks();
CompletableFuture<?>[] futures = tasks.stream()
.map(task -> CompletableFuture.runAsync(task, es))
.toArr
浏览 0提问于2018-09-02得票数 1
回答已采纳
我试图“连接”多个对等点，并并行“处理”它们。我有以下实现
pub async fn process(addr: &str) {
// simulate processing by sleeping
tokio::time::sleep(core::time::Duration::from_millis(1)).await;
println!("processed {}", addr);
}
#[tokio::main]
async fn main() {
let mut peers = vec!["127.0.0.1"
浏览 8提问于2022-09-12得票数 0
我有一段代码如下：
fn stream_it(&self) -> Box<dyn Stream<Item=()>> {
Box::new(self.some_field)
}
fn consume_it(&self) {
let a = self.stream_it().map(|i| i);
}
我得到了编译错误：
error: the `map` method cannot be invoked on a trait object
--> ...
|
69 | let a = self.strea
浏览 6提问于2020-04-17得票数 5
回答已采纳
我想实现一个基于FuturesUnordered的Stream，它同样应该计算返回类型为Result<SomeEnum>的async函数，不过为了简单起见，我们假设它只是一个Result<f64>。当async fns最终返回Futures时，我假设我必须用以下方式来定义我的结构： use anyhow::Result;
use futures::{Future, Stream, stream::FuturesUnordered};
use std::{pin::Pin, task::Poll};
use pin_project::pin_project;
#[pi
浏览 20提问于2021-04-27得票数 1
回答已采纳
1回答
当Rust抱怨两个相同的类型不匹配时，这意味着什么？
下面的错误似乎是比较类型..。
<for<'_> fn(&u32) -> impl futures::Future<Output = u32> {f} as FnOnce<(&u32,)>>::Output
...to本身。
error[E0308]: mismatched types
--> src/main.rs:31:18
|
31 | let output = map(f, input);
|
浏览 1提问于2022-06-16得票数 3
在我的SmtpService中，我想要立即发送响应头，并在处理完成时发送正文。这应该遵循SMTP交换：
C: DATA
S: 354 Start mail input
C: ... data ...
C: ... more ...
C: .
S: 250 Ok
我在里有这么多
#[macro_use]
extern crate log;
extern crate bytes;
extern crate tokio_proto;
extern crate tokio_service;
extern crate futures;
use std::io;
use bytes::Bytes;
u
浏览 0提问于2017-10-24得票数 1
2回答
我想使用Hyper的当前主分支编写一个服务器，它保存了一个POST请求传递的消息，并将此消息发送到每个传入的GET请求。
我有这样的例子，主要是从超级示例目录中复制的：
extern crate futures;
extern crate hyper;
extern crate pretty_env_logger;
use futures::future::FutureResult;
use hyper::{Get, Post, StatusCode};
use hyper::header::{ContentLength};
use hyper::server::{Http, Servic
浏览 10提问于2017-04-14得票数 17
回答已采纳
我正在尝试学习使用nom (5.0.1)，并希望获得两个标记之间的字符串：
use nom::{
bytes::complete::{tag_no_case, take_while},
character::{is_alphanumeric},
error::{ParseError},
sequence::{delimited},
IResult,
};
fn root<'a, E: ParseError<&'a str>>(i: &'a str) -> IResult<&am
浏览 1提问于2019-08-27得票数 3
回答已采纳
1回答
我正在尝试用tokio编写一个测试程序，它从网站上抓取一个文件，并将流响应写入一个文件。超级网站显示了一个使用while循环并使用.data()方法作为响应体的示例，但我想使用.map()和其他几个方法来操作流。
我认为下一个合理的尝试应该是使用TryStreamExt中的.into_async_read()方法将流转换为AsyncRead，但这似乎行不通。我必须使用映射将hyper::error::Error转换为std::error::Error以获得TryStream，但现在编译器告诉我转换后的流没有实现AsyncRead。下面是我的main.rs文件和错误：
src/main.rs
u
浏览 0提问于2020-04-01得票数 6
我有这个密码
use futures::Map;
use futures::sink::SendAll;
use futures::sink::SinkFromErr;
use futures::stream::Forward;
use futures::sync::mpsc::Receiver;
use futures::sync::mpsc::Sender;
use futures::{Future, Stream, Sink};
use std::boxed::FnBox;
use tokio_core::reactor::Core;
use websocket::async::futu
浏览 2提问于2017-10-04得票数 0
回答已采纳
1回答
我有N个futures::Stream，其中每个Item都有一个时间戳。我想将它们合并到一个流中，因此我想将所有流转换为peekable：
fn merge_streams(
mut streams: Vec<Box<dyn futures::Stream<Item = Data>>>,
) -> impl futures::Stream<Item = Data> {
async_stream::try_stream! {
let n = streams.len();
let stre
浏览 2提问于2020-08-19得票数 0
回答已采纳
1回答
流程为控制器层、->服务层
这里我从控制器层调用processLOBTransactions (异步方法)方法
如何将加入控制器层中的所有CompletableFuture响应？我的要求是在对来自控制器的每个list元素执行processLOBTransactions之后，在控制器层中，我想写日志之类的东西。
请大家就如何做到这一点提出建议。
**Controller Layer:**
class ControllerLayer{
pktransaction.getLineOfBusinessTransaction().stream().forEac
浏览 5提问于2021-02-07得票数 1
回答已采纳
最终，我需要一个rusoto的ByteStream。所以我想我可以从futures::Stream创建它 pub fn z_stream(r: impl Read) -> Result<ByteStream> {
let outstream = stream::empty();
// TODO, how do I wrap outstream
process(r, &mut wrapped_outstream)?;
Ok(ByteStream::new(outstream))
} 我有一个process方法，它有两个参数，一个用于impl Rea
浏览 20提问于2020-12-28得票数 1
回答已采纳
我需要实现一个返回futures::StreamExt特征的特征。 一般来说，这听起来很简单，有几个答案，例如this here。 我在StreamExt上试过了，但由于某种原因，这不起作用。下面是我的示例代码： // [dependencies]
// futures = "0.3.6"
// async-std = { version = "1.6.5", features = ["attributes", "unstable"] } // interval function is unstable atm.
use std
浏览 8提问于2020-10-13得票数 5
回答已采纳
在执行异步CompletableFuture时，父线程上下文以及org.slf4j.MDC上下文将丢失。
这很糟糕，因为我使用某种“鱼标记”来跟踪多个日志文件中的一个请求中的日志。
MDC.put("fishid", randomId())
问题:一般情况下，在CompletableFutures的任务中，我如何保留这个id？
List<CompletableFuture<UpdateHotelAllotmentsRsp>> futures =
tasks.stream()
.map(task -> CompletableF
浏览 2提问于2018-03-05得票数 12
回答已采纳
1回答
在可完成的未来内部/之后抛出异常
、、、、
我有一个可完成的未来列表，如下所示：
List<CompletableFuture<MyObject>> futures = class_list.stream()
.map(c -> CompletableFuture.supplyAsync(() ->
try {
return helper(c);
} catch (Throwable th) {
throw th; // doesn't work
}
))
.excepti
浏览 12提问于2019-11-02得票数 1
我有一个动态的futures::sync::mpsc::Sender集合，我想为每个传入的连接向他们每个人发送一条消息。
我让它与UnboundedSender一起工作，因为我可以这样做(见下文)，但是Sender会消耗它自己，所以我需要在发送后删除它并重新插入到Vec中。我该怎么做？如果Sender阻塞，它不应该发送更多的消息，而是切换到处理接收端的传入连接。
UnboundedSender实现如下，否则我失败的尝试将内联注释掉(只需将前面的行替换为注释掉的行)。
UnboundedSender (作品)
extern crate tokio;
use tokio::runtime::cur
浏览 1提问于2018-11-30得票数 1
回答已采纳
这样做是可行的：
public Long getMaxSalary(List<CompletableFuture<EmployeeData>> futures) {
CompletableFuture<Void> allDoneFuture = CompletableFuture.allOf(futures.toArray(new CompletableFuture[futures.size()]));
CompletableFuture<List<EmployeeData>> employeeDataList = allDoneF
浏览 0提问于2018-05-24得票数 0
回答已采纳
1回答
我有一个类似于的方法，它可以接受接收器：
pub async fn connect(
addr: &SocketAddr,
mut stdin: impl Stream<Item = Result<Request, io::Error>> + Unpin,
mut stdout: impl Sink<Response, Error = io::Error> + Unpin,
) -> Result<(), Box<dyn Error>> {
是否有一种标准/简单的方
浏览 0提问于2020-04-10得票数 0
回答已采纳
我曾经可以做到这一点，但我正在与0.3的未来作斗争。 下面是我从WebSocket获得的接收器和流： let (mut sink, stream) = ws_stream.split(); 我创建了一个无界通道来在异步tokio任务之间进行通信： let (unbounded_sender, unbounded_receiver) = mpsc::unbounded::<tungstenite::Message>(); 这就是我被困住的地方。我产生了一个异步任务，该任务应该连接未绑定的接收器和接收器；我的想法是通过unbounded_sender向下发送消息 tokio::spa
浏览 12提问于2020-07-22得票数 0
回答已采纳
到目前为止，我认为move |...| {...}会在闭包中移动变量，而闭包只实现FnOnce，因为您只能移动变量一次。然而，令我惊讶的是，我发现这段代码是有效的：
extern crate futures;
use futures::stream;
use futures::stream::{Stream, StreamExt};
use std::rc::Rc;
#[derive(Debug)]
struct Foo(i32);
fn bar(r: Rc<Foo>) -> Box<Stream<Item = (), Error = ()> + &#
浏览 0提问于2018-05-02得票数 5
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
实时音视频
即时通信 IM活动推荐
腾讯云开发者
