rust/keystore: make bip39 unlocking async
What changed, and why it matters
This commit is a large internal refactoring of the BitBox02 firmware's BIP39 wallet-unlocking code. It converts the slow PBKDF2 key-stretching loop from a blocking (synchronous) operation into an asynchronous one, so the device can briefly pause each round to handle other tasks such as USB messages and screen animations. The commit also vendors the futures-core, futures-lite, and pin-project-lite Rust crates so the firmware can use async/await. The change is explicitly described by the authors as not yet fully functional on its own; a follow-up commit is needed to make the unlock animation work with the new async model. There is no direct evidence in the commit that this fixes or introduces a security vulnerability; it is primarily an architectural/performance change.
Treat this as a non-security-critical refactoring commit. Review the follow-up commit that completes the async animation integration to ensure the BIP39 unlock path remains timing-attack resistant and that yielding does not leak intermediate state or allow re-entrancy bugs. Audit the newly vendored futures crates for compatibility with the firmware's no_std/embedded constraints and verify their checksums match published crates.io versions.
Security signals we found
Large dependency addition (futures-core, futures-lite, pin-project-lite) increases firmware attack surface and supply-chain exposure.
Async conversion of a security-critical key-derivation path introduces new concurrency and state-machine correctness requirements.
Commit explicitly notes the change is not yet functional on its own, indicating potential for incomplete/partial security behavior in this snapshot.
No direct diff evidence of a vulnerability fix or introduction; signals are architectural rather than exploit-specific.
Evidence from the diff
The patch makes BIP39 mnemonic-to-seed unlocking asynchronous by replacing to_seed_normalized(...) with to_seed_normalized_async(...).await and propagating async/await through the Rust keystore and unlock workflow. The PBKDF2 stretch loop now yields to the async executor on each of the 2048 rounds. In the simulator build, yielding is disabled and the computation remains blocking to avoid excessive simulator latency. The change pulls in vendored copies of futures-core 0.3.31, futures-lite 2.6.1, and pin-project-lite 0.2.16. The C unlock-animation component is also adjusted to support the new async flow. The commit message states the change is incomplete by itself because the unlock animation still relies on timer interrupts, which conflicts with the async model until the next commit.
Changed components
src/rust/bitbox02-rust/src/keystore.rssrc/rust/bitbox02-rust/src/workflow/unlock.rssrc/rust/bitbox02-rust/src/workflow/unlock_animation.rssrc/rust/bitbox02/src/keystore.rssrc/keystore.csrc/ui/components/unlock_animation.cexternal/vendor/futures-coreexternal/vendor/futures-liteexternal/vendor/pin-project-liteInspect captured patch +17928 / −261
diff --git a/external/vendor/futures-core/.cargo-checksum.json b/external/vendor/futures-core/.cargo-checksum.json
new file mode 100644
index 0000000..e2ac60b
--- /dev/null
+++ b/external/vendor/futures-core/.cargo-checksum.json
@@ -0,0 +1 @@
+{"files":{".cargo_vcs_info.json":"8e7b8a227215b5c1dd862396cd4c6f4bea54f83431aaf38663c3bd2d4dae06af","Cargo.toml":"c0ee4bd5904127f284fff848d01d9e6539e1762e6c1291f55025a9bf0b754827","Cargo.toml.orig":"2a9ede068bd90580bfd9c86cfe6462771b2f61bffe519d10240ccd6551b8ac7a","LICENSE-APACHE":"275c491d6d1160553c32fd6127061d7f9606c3ea25abfad6ca3f6ed088785427","LICENSE-MIT":"6652c868f35dfe5e8ef636810a4e576b9d663f3a17fb0f5613ad73583e1b88fd","README.md":"e8258273fed6f1796485777655118f2369fd3f000191e9d8cdbd10bf052946a9","src/future.rs":"bdbe034548271aef0c3dd8a6d087a5861e5920848c9e33a8949dbc40407c0ca7","src/lib.rs":"e545004177a7cd13257a3a562d2d44a5e0cff45687fc912b69e3d510fa397396","src/stream.rs":"11f0b4360287dd870c1b674db84f2452ddc38fbaf475cca27d374b65211af72d","src/task/__internal/atomic_waker.rs":"0418206de25768f691944c81f61ccec2c362751e56757bf812385c8ef01081fe","src/task/__internal/mod.rs":"1cc15fd61942a29ea558c5f4d5782e46adcfd914cab82be084a6882fa9afc122","src/task/mod.rs":"e213602a2fe5ae78ad5f1ca20e6d32dcbab17aba5b6b072fb927a72da99b4a11","src/task/poll.rs":"74c2717c1f9a37587a367da1b690d1cd2312e95dbaffca42be4755f1cd164bb8"},"package":"05f29059c0c2090612e8d742178b0580d2dc940c837851ad723096f87af6663e"}
\ No newline at end of file
diff --git a/external/vendor/futures-core/.cargo_vcs_info.json b/external/vendor/futures-core/.cargo_vcs_info.json
new file mode 100644
index 0000000..dcab678
--- /dev/null
+++ b/external/vendor/futures-core/.cargo_vcs_info.json
@@ -0,0 +1,6 @@
+{
+ "git": {
+ "sha1": "1e052816b09890925cfdfcbe8d390cdaae5e4c38"
+ },
+ "path_in_vcs": "futures-core"
+}
\ No newline at end of file
diff --git a/external/vendor/futures-core/Cargo.toml b/external/vendor/futures-core/Cargo.toml
new file mode 100644
index 0000000..6842872
--- /dev/null
+++ b/external/vendor/futures-core/Cargo.toml
@@ -0,0 +1,65 @@
+# THIS FILE IS AUTOMATICALLY GENERATED BY CARGO
+#
+# When uploading crates to the registry Cargo will automatically
+# "normalize" Cargo.toml files for maximal compatibility
+# with all versions of Cargo and also rewrite `path` dependencies
+# to registry (e.g., crates.io) dependencies.
+#
+# If you are reading this file be aware that the original Cargo.toml
+# will likely look very different (and much more reasonable).
+# See Cargo.toml.orig for the original contents.
+
+[package]
+edition = "2018"
+rust-version = "1.36"
+name = "futures-core"
+version = "0.3.31"
+build = false
+autobins = false
+autoexamples = false
+autotests = false
+autobenches = false
+description = """
+The core traits and types in for the `futures` library.
+"""
+homepage = "https://rust-lang.github.io/futures-rs"
+readme = "README.md"
+license = "MIT OR Apache-2.0"
+repository = "https://github.com/rust-lang/futures-rs"
+
+[package.metadata.docs.rs]
+all-features = true
+rustdoc-args = [
+ "--cfg",
+ "docsrs",
+]
+
+[lib]
+name = "futures_core"
+path = "src/lib.rs"
+
+[dependencies.portable-atomic]
+version = "1.3"
+features = ["require-cas"]
+optional = true
+default-features = false
+
+[dev-dependencies]
+
+[features]
+alloc = []
+cfg-target-has-atomic = []
+default = ["std"]
+std = ["alloc"]
+unstable = []
+
+[lints.rust]
+missing_debug_implementations = "warn"
+rust_2018_idioms = "warn"
+single_use_lifetimes = "warn"
+unreachable_pub = "warn"
+
+[lints.rust.unexpected_cfgs]
+level = "warn"
+priority = 0
+check-cfg = ["cfg(futures_sanitizer)"]
diff --git a/external/vendor/futures-core/Cargo.toml.orig b/external/vendor/futures-core/Cargo.toml.orig
new file mode 100644
index 0000000..a7d710e
--- /dev/null
+++ b/external/vendor/futures-core/Cargo.toml.orig
@@ -0,0 +1,34 @@
+[package]
+name = "futures-core"
+version = "0.3.31"
+edition = "2018"
+rust-version = "1.36"
+license = "MIT OR Apache-2.0"
+repository = "https://github.com/rust-lang/futures-rs"
+homepage = "https://rust-lang.github.io/futures-rs"
+description = """
+The core traits and types in for the `futures` library.
+"""
+
+[features]
+default = ["std"]
+std = ["alloc"]
+alloc = []
+
+# These features are no longer used.
+# TODO: remove in the next major version.
+unstable = []
+cfg-target-has-atomic = []
+
+[dependencies]
+portable-atomic = { version = "1.3", optional = true, default-features = false, features = ["require-cas"] }
+
+[dev-dependencies]
+futures = { path = "../futures" }
+
+[package.metadata.docs.rs]
+all-features = true
+rustdoc-args = ["--cfg", "docsrs"]
+
+[lints]
+workspace = true
diff --git a/external/vendor/futures-core/LICENSE-APACHE b/external/vendor/futures-core/LICENSE-APACHE
new file mode 100644
index 0000000..9eb0b09
--- /dev/null
+++ b/external/vendor/futures-core/LICENSE-APACHE
@@ -0,0 +1,202 @@
+ Apache License
+ Version 2.0, January 2004
+ http://www.apache.org/licenses/
+
+TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
+
+1. Definitions.
+
+ "License" shall mean the terms and conditions for use, reproduction,
+ and distribution as defined by Sections 1 through 9 of this document.
+
+ "Licensor" shall mean the copyright owner or entity authorized by
+ the copyright owner that is granting the License.
+
+ "Legal Entity" shall mean the union of the acting entity and all
+ other entities that control, are controlled by, or are under common
+ control with that entity. For the purposes of this definition,
+ "control" means (i) the power, direct or indirect, to cause the
+ direction or management of such entity, whether by contract or
+ otherwise, or (ii) ownership of fifty percent (50%) or more of the
+ outstanding shares, or (iii) beneficial ownership of such entity.
+
+ "You" (or "Your") shall mean an individual or Legal Entity
+ exercising permissions granted by this License.
+
+ "Source" form shall mean the preferred form for making modifications,
+ including but not limited to software source code, documentation
+ source, and configuration files.
+
+ "Object" form shall mean any form resulting from mechanical
+ transformation or translation of a Source form, including but
+ not limited to compiled object code, generated documentation,
+ and conversions to other media types.
+
+ "Work" shall mean the work of authorship, whether in Source or
+ Object form, made available under the License, as indicated by a
+ copyright notice that is included in or attached to the work
+ (an example is provided in the Appendix below).
+
+ "Derivative Works" shall mean any work, whether in Source or Object
+ form, that is based on (or derived from) the Work and for which the
+ editorial revisions, annotations, elaborations, or other modifications
+ represent, as a whole, an original work of authorship. For the purposes
+ of this License, Derivative Works shall not include works that remain
+ separable from, or merely link (or bind by name) to the interfaces of,
+ the Work and Derivative Works thereof.
+
+ "Contribution" shall mean any work of authorship, including
+ the original version of the Work and any modifications or additions
+ to that Work or Derivative Works thereof, that is intentionally
+ submitted to Licensor for inclusion in the Work by the copyright owner
+ or by an individual or Legal Entity authorized to submit on behalf of
+ the copyright owner. For the purposes of this definition, "submitted"
+ means any form of electronic, verbal, or written communication sent
+ to the Licensor or its representatives, including but not limited to
+ communication on electronic mailing lists, source code control systems,
+ and issue tracking systems that are managed by, or on behalf of, the
+ Licensor for the purpose of discussing and improving the Work, but
+ excluding communication that is conspicuously marked or otherwise
+ designated in writing by the copyright owner as "Not a Contribution."
+
+ "Contributor" shall mean Licensor and any individual or Legal Entity
+ on behalf of whom a Contribution has been received by Licensor and
+ subsequently incorporated within the Work.
+
+2. Grant of Copyright License. Subject to the terms and conditions of
+ this License, each Contributor hereby grants to You a perpetual,
+ worldwide, non-exclusive, no-charge, royalty-free, irrevocable
+ copyright license to reproduce, prepare Derivative Works of,
+ publicly display, publicly perform, sublicense, and distribute the
+ Work and such Derivative Works in Source or Object form.
+
+3. Grant of Patent License. Subject to the terms and conditions of
+ this License, each Contributor hereby grants to You a perpetual,
+ worldwide, non-exclusive, no-charge, royalty-free, irrevocable
+ (except as stated in this section) patent license to make, have made,
+ use, offer to sell, sell, import, and otherwise transfer the Work,
+ where such license applies only to those patent claims licensable
+ by such Contributor that are necessarily infringed by their
+ Contribution(s) alone or by combination of their Contribution(s)
+ with the Work to which such Contribution(s) was submitted. If You
+ institute patent litigation against any entity (including a
+ cross-claim or counterclaim in a lawsuit) alleging that the Work
+ or a Contribution incorporated within the Work constitutes direct
+ or contributory patent infringement, then any patent licenses
+ granted to You under this License for that Work shall terminate
+ as of the date such litigation is filed.
+
+4. Redistribution. You may reproduce and distribute copies of the
+ Work or Derivative Works thereof in any medium, with or without
+ modifications, and in Source or Object form, provided that You
+ meet the following conditions:
+
+ (a) You must give any other recipients of the Work or
+ Derivative Works a copy of this License; and
+
+ (b) You must cause any modified files to carry prominent notices
+ stating that You changed the files; and
+
+ (c) You must retain, in the Source form of any Derivative Works
+ that You distribute, all copyright, patent, trademark, and
+ attribution notices from the Source form of the Work,
+ excluding those notices that do not pertain to any part of
+ the Derivative Works; and
+
+ (d) If the Work includes a "NOTICE" text file as part of its
+ distribution, then any Derivative Works that You distribute must
+ include a readable copy of the attribution notices contained
+ within such NOTICE file, excluding those notices that do not
+ pertain to any part of the Derivative Works, in at least one
+ of the following places: within a NOTICE text file distributed
+ as part of the Derivative Works; within the Source form or
+ documentation, if provided along with the Derivative Works; or,
+ within a display generated by the Derivative Works, if and
+ wherever such third-party notices normally appear. The contents
+ of the NOTICE file are for informational purposes only and
+ do not modify the License. You may add Your own attribution
+ notices within Derivative Works that You distribute, alongside
+ or as an addendum to the NOTICE text from the Work, provided
+ that such additional attribution notices cannot be construed
+ as modifying the License.
+
+ You may add Your own copyright statement to Your modifications and
+ may provide additional or different license terms and conditions
+ for use, reproduction, or distribution of Your modifications, or
+ for any such Derivative Works as a whole, provided Your use,
+ reproduction, and distribution of the Work otherwise complies with
+ the conditions stated in this License.
+
+5. Submission of Contributions. Unless You explicitly state otherwise,
+ any Contribution intentionally submitted for inclusion in the Work
+ by You to the Licensor shall be under the terms and conditions of
+ this License, without any additional terms or conditions.
+ Notwithstanding the above, nothing herein shall supersede or modify
+ the terms of any separate license agreement you may have executed
+ with Licensor regarding such Contributions.
+
+6. Trademarks. This License does not grant permission to use the trade
+ names, trademarks, service marks, or product names of the Licensor,
+ except as required for reasonable and customary use in describing the
+ origin of the Work and reproducing the content of the NOTICE file.
+
+7. Disclaimer of Warranty. Unless required by applicable law or
+ agreed to in writing, Licensor provides the Work (and each
+ Contributor provides its Contributions) on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+ implied, including, without limitation, any warranties or conditions
+ of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A
+ PARTICULAR PURPOSE. You are solely responsible for determining the
+ appropriateness of using or redistributing the Work and assume any
+ risks associated with Your exercise of permissions under this License.
+
+8. Limitation of Liability. In no event and under no legal theory,
+ whether in tort (including negligence), contract, or otherwise,
+ unless required by applicable law (such as deliberate and grossly
+ negligent acts) or agreed to in writing, shall any Contributor be
+ liable to You for damages, including any direct, indirect, special,
+ incidental, or consequential damages of any character arising as a
+ result of this License or out of the use or inability to use the
+ Work (including but not limited to damages for loss of goodwill,
+ work stoppage, computer failure or malfunction, or any and all
+ other commercial damages or losses), even if such Contributor
+ has been advised of the possibility of such damages.
+
+9. Accepting Warranty or Additional Liability. While redistributing
+ the Work or Derivative Works thereof, You may choose to offer,
+ and charge a fee for, acceptance of support, warranty, indemnity,
+ or other liability obligations and/or rights consistent with this
+ License. However, in accepting such obligations, You may act only
+ on Your own behalf and on Your sole responsibility, not on behalf
+ of any other Contributor, and only if You agree to indemnify,
+ defend, and hold each Contributor harmless for any liability
+ incurred by, or claims asserted against, such Contributor by reason
+ of your accepting any such warranty or additional liability.
+
+END OF TERMS AND CONDITIONS
+
+APPENDIX: How to apply the Apache License to your work.
+
+ To apply the Apache License to your work, attach the following
+ boilerplate notice, with the fields enclosed by brackets "[]"
+ replaced with your own identifying information. (Don't include
+ the brackets!) The text should be enclosed in the appropriate
+ comment syntax for the file format. We also recommend that a
+ file or class name and description of purpose be included on the
+ same "printed page" as the copyright notice for easier
+ identification within third-party archives.
+
+Copyright (c) 2016 Alex Crichton
+Copyright (c) 2017 The Tokio Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
diff --git a/external/vendor/futures-core/LICENSE-MIT b/external/vendor/futures-core/LICENSE-MIT
new file mode 100644
index 0000000..8ad082e
--- /dev/null
+++ b/external/vendor/futures-core/LICENSE-MIT
@@ -0,0 +1,26 @@
+Copyright (c) 2016 Alex Crichton
+Copyright (c) 2017 The Tokio Authors
+
+Permission is hereby granted, free of charge, to any
+person obtaining a copy of this software and associated
+documentation files (the "Software"), to deal in the
+Software without restriction, including without
+limitation the rights to use, copy, modify, merge,
+publish, distribute, sublicense, and/or sell copies of
+the Software, and to permit persons to whom the Software
+is furnished to do so, subject to the following
+conditions:
+
+The above copyright notice and this permission notice
+shall be included in all copies or substantial portions
+of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF
+ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED
+TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A
+PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT
+SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY
+CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
+OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR
+IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
+DEALINGS IN THE SOFTWARE.
diff --git a/external/vendor/futures-core/README.md b/external/vendor/futures-core/README.md
new file mode 100644
index 0000000..96e0e06
--- /dev/null
+++ b/external/vendor/futures-core/README.md
@@ -0,0 +1,23 @@
+# futures-core
+
+The core traits and types in for the `futures` library.
+
+## Usage
+
+Add this to your `Cargo.toml`:
+
+```toml
+[dependencies]
+futures-core = "0.3"
+```
+
+The current `futures-core` requires Rust 1.36 or later.
+
+## License
+
+Licensed under either of [Apache License, Version 2.0](LICENSE-APACHE) or
+[MIT license](LICENSE-MIT) at your option.
+
+Unless you explicitly state otherwise, any contribution intentionally submitted
+for inclusion in the work by you, as defined in the Apache-2.0 license, shall
+be dual licensed as above, without any additional terms or conditions.
diff --git a/external/vendor/futures-core/src/future.rs b/external/vendor/futures-core/src/future.rs
new file mode 100644
index 0000000..30c0323
--- /dev/null
+++ b/external/vendor/futures-core/src/future.rs
@@ -0,0 +1,113 @@
+//! Futures.
+
+use core::ops::DerefMut;
+use core::pin::Pin;
+use core::task::{Context, Poll};
+
+#[doc(no_inline)]
+pub use core::future::Future;
+
+/// An owned dynamically typed [`Future`] for use in cases where you can't
+/// statically type your result or need to add some indirection.
+///
+/// This type is often created by the [`boxed`] method on [`FutureExt`]. See its documentation for more.
+///
+/// [`boxed`]: https://docs.rs/futures/latest/futures/future/trait.FutureExt.html#method.boxed
+/// [`FutureExt`]: https://docs.rs/futures/latest/futures/future/trait.FutureExt.html
+#[cfg(feature = "alloc")]
+pub type BoxFuture<'a, T> = Pin<alloc::boxed::Box<dyn Future<Output = T> + Send + 'a>>;
+
+/// `BoxFuture`, but without the `Send` requirement.
+///
+/// This type is often created by the [`boxed_local`] method on [`FutureExt`]. See its documentation for more.
+///
+/// [`boxed_local`]: https://docs.rs/futures/latest/futures/future/trait.FutureExt.html#method.boxed_local
+/// [`FutureExt`]: https://docs.rs/futures/latest/futures/future/trait.FutureExt.html
+#[cfg(feature = "alloc")]
+pub type LocalBoxFuture<'a, T> = Pin<alloc::boxed::Box<dyn Future<Output = T> + 'a>>;
+
+/// A future which tracks whether or not the underlying future
+/// should no longer be polled.
+///
+/// `is_terminated` will return `true` if a future should no longer be polled.
+/// Usually, this state occurs after `poll` (or `try_poll`) returned
+/// `Poll::Ready`. However, `is_terminated` may also return `true` if a future
+/// has become inactive and can no longer make progress and should be ignored
+/// or dropped rather than being `poll`ed again.
+pub trait FusedFuture: Future {
+ /// Returns `true` if the underlying future should no longer be polled.
+ fn is_terminated(&self) -> bool;
+}
+
+impl<F: FusedFuture + ?Sized + Unpin> FusedFuture for &mut F {
+ fn is_terminated(&self) -> bool {
+ <F as FusedFuture>::is_terminated(&**self)
+ }
+}
+
+impl<P> FusedFuture for Pin<P>
+where
+ P: DerefMut + Unpin,
+ P::Target: FusedFuture,
+{
+ fn is_terminated(&self) -> bool {
+ <P::Target as FusedFuture>::is_terminated(&**self)
+ }
+}
+
+mod private_try_future {
+ use super::Future;
+
+ pub trait Sealed {}
+
+ impl<F, T, E> Sealed for F where F: ?Sized + Future<Output = Result<T, E>> {}
+}
+
+/// A convenience for futures that return `Result` values that includes
+/// a variety of adapters tailored to such futures.
+pub trait TryFuture: Future + private_try_future::Sealed {
+ /// The type of successful values yielded by this future
+ type Ok;
+
+ /// The type of failures yielded by this future
+ type Error;
+
+ /// Poll this `TryFuture` as if it were a `Future`.
+ ///
+ /// This method is a stopgap for a compiler limitation that prevents us from
+ /// directly inheriting from the `Future` trait; in the future it won't be
+ /// needed.
+ fn try_poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<Self::Ok, Self::Error>>;
+}
+
+impl<F, T, E> TryFuture for F
+where
+ F: ?Sized + Future<Output = Result<T, E>>,
+{
+ type Ok = T;
+ type Error = E;
+
+ #[inline]
+ fn try_poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+ self.poll(cx)
+ }
+}
+
+#[cfg(feature = "alloc")]
+mod if_alloc {
+ use super::*;
+ use alloc::boxed::Box;
+
+ impl<F: FusedFuture + ?Sized + Unpin> FusedFuture for Box<F> {
+ fn is_terminated(&self) -> bool {
+ <F as FusedFuture>::is_terminated(&**self)
+ }
+ }
+
+ #[cfg(feature = "std")]
+ impl<F: FusedFuture> FusedFuture for std::panic::AssertUnwindSafe<F> {
+ fn is_terminated(&self) -> bool {
+ <F as FusedFuture>::is_terminated(&**self)
+ }
+ }
+}
diff --git a/external/vendor/futures-core/src/lib.rs b/external/vendor/futures-core/src/lib.rs
new file mode 100644
index 0000000..6ff6b97
--- /dev/null
+++ b/external/vendor/futures-core/src/lib.rs
@@ -0,0 +1,27 @@
+//! Core traits and types for asynchronous operations in Rust.
+
+#![no_std]
+#![doc(test(
+ no_crate_inject,
+ attr(
+ deny(warnings, rust_2018_idioms, single_use_lifetimes),
+ allow(dead_code, unused_assignments, unused_variables)
+ )
+))]
+#![warn(missing_docs, /* unsafe_op_in_unsafe_fn */)] // unsafe_op_in_unsafe_fn requires Rust 1.52
+
+#[cfg(feature = "alloc")]
+extern crate alloc;
+#[cfg(feature = "std")]
+extern crate std;
+
+pub mod future;
+#[doc(no_inline)]
+pub use self::future::{FusedFuture, Future, TryFuture};
+
+pub mod stream;
+#[doc(no_inline)]
+pub use self::stream::{FusedStream, Stream, TryStream};
+
+#[macro_use]
+pub mod task;
diff --git a/external/vendor/futures-core/src/stream.rs b/external/vendor/futures-core/src/stream.rs
new file mode 100644
index 0000000..dd07d5a
--- /dev/null
+++ b/external/vendor/futures-core/src/stream.rs
@@ -0,0 +1,245 @@
+//! Asynchronous streams.
+
+use core::ops::DerefMut;
+use core::pin::Pin;
+use core::task::{Context, Poll};
+
+/// An owned dynamically typed [`Stream`] for use in cases where you can't
+/// statically type your result or need to add some indirection.
+///
+/// This type is often created by the [`boxed`] method on [`StreamExt`]. See its documentation for more.
+///
+/// [`boxed`]: https://docs.rs/futures/latest/futures/stream/trait.StreamExt.html#method.boxed
+/// [`StreamExt`]: https://docs.rs/futures/latest/futures/stream/trait.StreamExt.html
+#[cfg(feature = "alloc")]
+pub type BoxStream<'a, T> = Pin<alloc::boxed::Box<dyn Stream<Item = T> + Send + 'a>>;
+
+/// `BoxStream`, but without the `Send` requirement.
+///
+/// This type is often created by the [`boxed_local`] method on [`StreamExt`]. See its documentation for more.
+///
+/// [`boxed_local`]: https://docs.rs/futures/latest/futures/stream/trait.StreamExt.html#method.boxed_local
+/// [`StreamExt`]: https://docs.rs/futures/latest/futures/stream/trait.StreamExt.html
+#[cfg(feature = "alloc")]
+pub type LocalBoxStream<'a, T> = Pin<alloc::boxed::Box<dyn Stream<Item = T> + 'a>>;
+
+/// A stream of values produced asynchronously.
+///
+/// If `Future<Output = T>` is an asynchronous version of `T`, then `Stream<Item
+/// = T>` is an asynchronous version of `Iterator<Item = T>`. A stream
+/// represents a sequence of value-producing events that occur asynchronously to
+/// the caller.
+///
+/// The trait is modeled after `Future`, but allows `poll_next` to be called
+/// even after a value has been produced, yielding `None` once the stream has
+/// been fully exhausted.
+#[must_use = "streams do nothing unless polled"]
+pub trait Stream {
+ /// Values yielded by the stream.
+ type Item;
+
+ /// Attempt to pull out the next value of this stream, registering the
+ /// current task for wakeup if the value is not yet available, and returning
+ /// `None` if the stream is exhausted.
+ ///
+ /// # Return value
+ ///
+ /// There are several possible return values, each indicating a distinct
+ /// stream state:
+ ///
+ /// - `Poll::Pending` means that this stream's next value is not ready
+ /// yet. Implementations will ensure that the current task will be notified
+ /// when the next value may be ready.
+ ///
+ /// - `Poll::Ready(Some(val))` means that the stream has successfully
+ /// produced a value, `val`, and may produce further values on subsequent
+ /// `poll_next` calls.
+ ///
+ /// - `Poll::Ready(None)` means that the stream has terminated, and
+ /// `poll_next` should not be invoked again.
+ ///
+ /// # Panics
+ ///
+ /// Once a stream has finished (returned `Ready(None)` from `poll_next`), calling its
+ /// `poll_next` method again may panic, block forever, or cause other kinds of
+ /// problems; the `Stream` trait places no requirements on the effects of
+ /// such a call. However, as the `poll_next` method is not marked `unsafe`,
+ /// Rust's usual rules apply: calls must never cause undefined behavior
+ /// (memory corruption, incorrect use of `unsafe` functions, or the like),
+ /// regardless of the stream's state.
+ ///
+ /// If this is difficult to guard against then the [`fuse`] adapter can be used
+ /// to ensure that `poll_next` always returns `Ready(None)` in subsequent
+ /// calls.
+ ///
+ /// [`fuse`]: https://docs.rs/futures/0.3/futures/stream/trait.StreamExt.html#method.fuse
+ fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>>;
+
+ /// Returns the bounds on the remaining length of the stream.
+ ///
+ /// Specifically, `size_hint()` returns a tuple where the first element
+ /// is the lower bound, and the second element is the upper bound.
+ ///
+ /// The second half of the tuple that is returned is an [`Option`]`<`[`usize`]`>`.
+ /// A [`None`] here means that either there is no known upper bound, or the
+ /// upper bound is larger than [`usize`].
+ ///
+ /// # Implementation notes
+ ///
+ /// It is not enforced that a stream implementation yields the declared
+ /// number of elements. A buggy stream may yield less than the lower bound
+ /// or more than the upper bound of elements.
+ ///
+ /// `size_hint()` is primarily intended to be used for optimizations such as
+ /// reserving space for the elements of the stream, but must not be
+ /// trusted to e.g., omit bounds checks in unsafe code. An incorrect
+ /// implementation of `size_hint()` should not lead to memory safety
+ /// violations.
+ ///
+ /// That said, the implementation should provide a correct estimation,
+ /// because otherwise it would be a violation of the trait's protocol.
+ ///
+ /// The default implementation returns `(0, `[`None`]`)` which is correct for any
+ /// stream.
+ #[inline]
+ fn size_hint(&self) -> (usize, Option<usize>) {
+ (0, None)
+ }
+}
+
+impl<S: ?Sized + Stream + Unpin> Stream for &mut S {
+ type Item = S::Item;
+
+ fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+ S::poll_next(Pin::new(&mut **self), cx)
+ }
+
+ fn size_hint(&self) -> (usize, Option<usize>) {
+ (**self).size_hint()
+ }
+}
+
+impl<P> Stream for Pin<P>
+where
+ P: DerefMut + Unpin,
+ P::Target: Stream,
+{
+ type Item = <P::Target as Stream>::Item;
+
+ fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+ self.get_mut().as_mut().poll_next(cx)
+ }
+
+ fn size_hint(&self) -> (usize, Option<usize>) {
+ (**self).size_hint()
+ }
+}
+
+/// A stream which tracks whether or not the underlying stream
+/// should no longer be polled.
+///
+/// `is_terminated` will return `true` if a future should no longer be polled.
+/// Usually, this state occurs after `poll_next` (or `try_poll_next`) returned
+/// `Poll::Ready(None)`. However, `is_terminated` may also return `true` if a
+/// stream has become inactive and can no longer make progress and should be
+/// ignored or dropped rather than being polled again.
+pub trait FusedStream: Stream {
+ /// Returns `true` if the stream should no longer be polled.
+ fn is_terminated(&self) -> bool;
+}
+
+impl<F: ?Sized + FusedStream + Unpin> FusedStream for &mut F {
+ fn is_terminated(&self) -> bool {
+ <F as FusedStream>::is_terminated(&**self)
+ }
+}
+
+impl<P> FusedStream for Pin<P>
+where
+ P: DerefMut + Unpin,
+ P::Target: FusedStream,
+{
+ fn is_terminated(&self) -> bool {
+ <P::Target as FusedStream>::is_terminated(&**self)
+ }
+}
+
+mod private_try_stream {
+ use super::Stream;
+
+ pub trait Sealed {}
+
+ impl<S, T, E> Sealed for S where S: ?Sized + Stream<Item = Result<T, E>> {}
+}
+
+/// A convenience for streams that return `Result` values that includes
+/// a variety of adapters tailored to such futures.
+pub trait TryStream: Stream + private_try_stream::Sealed {
+ /// The type of successful values yielded by this future
+ type Ok;
+
+ /// The type of failures yielded by this future
+ type Error;
+
+ /// Poll this `TryStream` as if it were a `Stream`.
+ ///
+ /// This method is a stopgap for a compiler limitation that prevents us from
+ /// directly inheriting from the `Stream` trait; in the future it won't be
+ /// needed.
+ fn try_poll_next(
+ self: Pin<&mut Self>,
+ cx: &mut Context<'_>,
+ ) -> Poll<Option<Result<Self::Ok, Self::Error>>>;
+}
+
+impl<S, T, E> TryStream for S
+where
+ S: ?Sized + Stream<Item = Result<T, E>>,
+{
+ type Ok = T;
+ type Error = E;
+
+ fn try_poll_next(
+ self: Pin<&mut Self>,
+ cx: &mut Context<'_>,
+ ) -> Poll<Option<Result<Self::Ok, Self::Error>>> {
+ self.poll_next(cx)
+ }
+}
+
+#[cfg(feature = "alloc")]
+mod if_alloc {
+ use super::*;
+ use alloc::boxed::Box;
+
+ impl<S: ?Sized + Stream + Unpin> Stream for Box<S> {
+ type Item = S::Item;
+
+ fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+ Pin::new(&mut **self).poll_next(cx)
+ }
+
+ fn size_hint(&self) -> (usize, Option<usize>) {
+ (**self).size_hint()
+ }
+ }
+
+ #[cfg(feature = "std")]
+ impl<S: Stream> Stream for std::panic::AssertUnwindSafe<S> {
+ type Item = S::Item;
+
+ fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<S::Item>> {
+ unsafe { self.map_unchecked_mut(|x| &mut x.0) }.poll_next(cx)
+ }
+
+ fn size_hint(&self) -> (usize, Option<usize>) {
+ self.0.size_hint()
+ }
+ }
+
+ impl<S: ?Sized + FusedStream + Unpin> FusedStream for Box<S> {
+ fn is_terminated(&self) -> bool {
+ <S as FusedStream>::is_terminated(&**self)
+ }
+ }
+}
diff --git a/external/vendor/futures-core/src/task/__internal/atomic_waker.rs b/external/vendor/futures-core/src/task/__internal/atomic_waker.rs
new file mode 100644
index 0000000..3b82fb7
--- /dev/null
+++ b/external/vendor/futures-core/src/task/__internal/atomic_waker.rs
@@ -0,0 +1,422 @@
+use core::cell::UnsafeCell;
+use core::fmt;
+use core::task::Waker;
+
+use atomic::AtomicUsize;
+use atomic::Ordering::{AcqRel, Acquire, Release};
+
+#[cfg(feature = "portable-atomic")]
+use portable_atomic as atomic;
+
+#[cfg(not(feature = "portable-atomic"))]
+use core::sync::atomic;
+
+/// A synchronization primitive for task wakeup.
+///
+/// Sometimes the task interested in a given event will change over time.
+/// An `AtomicWaker` can coordinate concurrent notifications with the consumer
+/// potentially "updating" the underlying task to wake up. This is useful in
+/// scenarios where a computation completes in another thread and wants to
+/// notify the consumer, but the consumer is in the process of being migrated to
+/// a new logical task.
+///
+/// Consumers should call `register` before checking the result of a computation
+/// and producers should call `wake` after producing the computation (this
+/// differs from the usual `thread::park` pattern). It is also permitted for
+/// `wake` to be called **before** `register`. This results in a no-op.
+///
+/// A single `AtomicWaker` may be reused for any number of calls to `register` or
+/// `wake`.
+///
+/// # Memory ordering
+///
+/// Calling `register` "acquires" all memory "released" by calls to `wake`
+/// before the call to `register`. Later calls to `wake` will wake the
+/// registered waker (on contention this wake might be triggered in `register`).
+///
+/// For concurrent calls to `register` (should be avoided) the ordering is only
+/// guaranteed for the winning call.
+///
+/// # Examples
+///
+/// Here is a simple example providing a `Flag` that can be signalled manually
+/// when it is ready.
+///
+/// ```
+/// use futures::future::Future;
+/// use futures::task::{Context, Poll, AtomicWaker};
+/// use std::sync::Arc;
+/// use std::sync::atomic::AtomicBool;
+/// use std::sync::atomic::Ordering::Relaxed;
+/// use std::pin::Pin;
+///
+/// struct Inner {
+/// waker: AtomicWaker,
+/// set: AtomicBool,
+/// }
+///
+/// #[derive(Clone)]
+/// struct Flag(Arc<Inner>);
+///
+/// impl Flag {
+/// pub fn new() -> Self {
+/// Self(Arc::new(Inner {
+/// waker: AtomicWaker::new(),
+/// set: AtomicBool::new(false),
+/// }))
+/// }
+///
+/// pub fn signal(&self) {
+/// self.0.set.store(true, Relaxed);
+/// self.0.waker.wake();
+/// }
+/// }
+///
+/// impl Future for Flag {
+/// type Output = ();
+///
+/// fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> {
+/// // quick check to avoid registration if already done.
+/// if self.0.set.load(Relaxed) {
+/// return Poll::Ready(());
+/// }
+///
+/// self.0.waker.register(cx.waker());
+///
+/// // Need to check condition **after** `register` to avoid a race
+/// // condition that would result in lost notifications.
+/// if self.0.set.load(Relaxed) {
+/// Poll::Ready(())
+/// } else {
+/// Poll::Pending
+/// }
+/// }
+/// }
+/// ```
+pub struct AtomicWaker {
+ state: AtomicUsize,
+ waker: UnsafeCell<Option<Waker>>,
+}
+
+// `AtomicWaker` is a multi-consumer, single-producer transfer cell. The cell
+// stores a `Waker` value produced by calls to `register` and many threads can
+// race to take the waker (to wake it) by calling `wake`.
+//
+// If a new `Waker` instance is produced by calling `register` before an
+// existing one is consumed, then the existing one is overwritten.
+//
+// While `AtomicWaker` is single-producer, the implementation ensures memory
+// safety. In the event of concurrent calls to `register`, there will be a
+// single winner whose waker will get stored in the cell. The losers will not
+// have their tasks woken. As such, callers should ensure to add synchronization
+// to calls to `register`.
+//
+// The implementation uses a single `AtomicUsize` value to coordinate access to
+// the `Waker` cell. There are two bits that are operated on independently.
+// These are represented by `REGISTERING` and `WAKING`.
+//
+// The `REGISTERING` bit is set when a producer enters the critical section. The
+// `WAKING` bit is set when a consumer enters the critical section. Neither bit
+// being set is represented by `WAITING`.
+//
+// A thread obtains an exclusive lock on the waker cell by transitioning the
+// state from `WAITING` to `REGISTERING` or `WAKING`, depending on the operation
+// the thread wishes to perform. When this transition is made, it is guaranteed
+// that no other thread will access the waker cell.
+//
+// # Registering
+//
+// On a call to `register`, an attempt to transition the state from WAITING to
+// REGISTERING is made. On success, the caller obtains a lock on the waker cell.
+//
+// If the lock is obtained, then the thread sets the waker cell to the waker
+// provided as an argument. Then it attempts to transition the state back from
+// `REGISTERING` -> `WAITING`.
+//
+// If this transition is successful, then the registering process is complete
+// and the next call to `wake` will observe the waker.
+//
+// If the transition fails, then there was a concurrent call to `wake` that was
+// unable to access the waker cell (due to the registering thread holding the
+// lock). To handle this, the registering thread removes the waker it just set
+// from the cell and calls `wake` on it. This call to wake represents the
+// attempt to wake by the other thread (that set the `WAKING` bit). The state is
+// then transitioned from `REGISTERING | WAKING` back to `WAITING`. This
+// transition must succeed because, at this point, the state cannot be
+// transitioned by another thread.
+//
+// # Waking
+//
+// On a call to `wake`, an attempt to transition the state from `WAITING` to
+// `WAKING` is made. On success, the caller obtains a lock on the waker cell.
+//
+// If the lock is obtained, then the thread takes ownership of the current value
+// in the waker cell, and calls `wake` on it. The state is then transitioned
+// back to `WAITING`. This transition must succeed as, at this point, the state
+// cannot be transitioned by another thread.
+//
+// If the thread is unable to obtain the lock, the `WAKING` bit is still. This
+// is because it has either been set by the current thread but the previous
+// value included the `REGISTERING` bit **or** a concurrent thread is in the
+// `WAKING` critical section. Either way, no action must be taken.
+//
+// If the current thread is the only concurrent call to `wake` and another
+// thread is in the `register` critical section, when the other thread **exits**
+// the `register` critical section, it will observe the `WAKING` bit and handle
+// the wake itself.
+//
+// If another thread is in the `wake` critical section, then it will handle
+// waking the task.
+//
+// # A potential race (is safely handled).
+//
+// Imagine the following situation:
+//
+// * Thread A obtains the `wake` lock and wakes a task.
+//
+// * Before thread A releases the `wake` lock, the woken task is scheduled.
+//
+// * Thread B attempts to wake the task. In theory this should result in the
+// task being woken, but it cannot because thread A still holds the wake lock.
+//
+// This case is handled by requiring users of `AtomicWaker` to call `register`
+// **before** attempting to observe the application state change that resulted
+// in the task being awoken. The wakers also change the application state before
+// calling wake.
+//
+// Because of this, the waker will do one of two things.
+//
+// 1) Observe the application state change that Thread B is woken for. In this
+// case, it is OK for Thread B's wake to be lost.
+//
+// 2) Call register before attempting to observe the application state. Since
+// Thread A still holds the `wake` lock, the call to `register` will result
+// in the task waking itself and get scheduled again.
+
+/// Idle state
+const WAITING: usize = 0;
+
+/// A new waker value is being registered with the `AtomicWaker` cell.
+const REGISTERING: usize = 0b01;
+
+/// The waker currently registered with the `AtomicWaker` cell is being woken.
+const WAKING: usize = 0b10;
+
+impl AtomicWaker {
+ /// Create an `AtomicWaker`.
+ pub const fn new() -> Self {
+ // Make sure that task is Sync
+ #[allow(dead_code)]
+ trait AssertSync: Sync {}
+ impl AssertSync for Waker {}
+
+ Self { state: AtomicUsize::new(WAITING), waker: UnsafeCell::new(None) }
+ }
+
+ /// Registers the waker to be notified on calls to `wake`.
+ ///
+ /// The new task will take place of any previous tasks that were registered
+ /// by previous calls to `register`. Any calls to `wake` that happen after
+ /// a call to `register` (as defined by the memory ordering rules), will
+ /// notify the `register` caller's task and deregister the waker from future
+ /// notifications. Because of this, callers should ensure `register` gets
+ /// invoked with a new `Waker` **each** time they require a wakeup.
+ ///
+ /// It is safe to call `register` with multiple other threads concurrently
+ /// calling `wake`. This will result in the `register` caller's current
+ /// task being notified once.
+ ///
+ /// This function is safe to call concurrently, but this is generally a bad
+ /// idea. Concurrent calls to `register` will attempt to register different
+ /// tasks to be notified. One of the callers will win and have its task set,
+ /// but there is no guarantee as to which caller will succeed.
+ ///
+ /// # Examples
+ ///
+ /// Here is how `register` is used when implementing a flag.
+ ///
+ /// ```
+ /// use futures::future::Future;
+ /// use futures::task::{Context, Poll, AtomicWaker};
+ /// use std::sync::atomic::AtomicBool;
+ /// use std::sync::atomic::Ordering::Relaxed;
+ /// use std::pin::Pin;
+ ///
+ /// struct Flag {
+ /// waker: AtomicWaker,
+ /// set: AtomicBool,
+ /// }
+ ///
+ /// impl Future for Flag {
+ /// type Output = ();
+ ///
+ /// fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> {
+ /// // Register **before** checking `set` to avoid a race condition
+ /// // that would result in lost notifications.
+ /// self.waker.register(cx.waker());
+ ///
+ /// if self.set.load(Relaxed) {
+ /// Poll::Ready(())
+ /// } else {
+ /// Poll::Pending
+ /// }
+ /// }
+ /// }
+ /// ```
+ pub fn register(&self, waker: &Waker) {
+ match self
+ .state
+ .compare_exchange(WAITING, REGISTERING, Acquire, Acquire)
+ .unwrap_or_else(|x| x)
+ {
+ WAITING => {
+ unsafe {
+ // Locked acquired, update the waker cell
+
+ // Avoid cloning the waker if the old waker will awaken the same task.
+ match &*self.waker.get() {
+ Some(old_waker) if old_waker.will_wake(waker) => (),
+ _ => *self.waker.get() = Some(waker.clone()),
+ }
+
+ // Release the lock. If the state transitioned to include
+ // the `WAKING` bit, this means that at least one wake has
+ // been called concurrently.
+ //
+ // Start by assuming that the state is `REGISTERING` as this
+ // is what we just set it to. If this holds, we know that no
+ // other writes were performed in the meantime, so there is
+ // nothing to acquire, only release. In case of concurrent
+ // wakers, we need to acquire their releases, so success needs
+ // to do both.
+ let res = self.state.compare_exchange(REGISTERING, WAITING, AcqRel, Acquire);
+
+ match res {
+ Ok(_) => {
+ // memory ordering: acquired self.state during CAS
+ // - if previous wakes went through it syncs with
+ // their final release (`fetch_and`)
+ // - if there was no previous wake the next wake
+ // will wake us, no sync needed.
+ }
+ Err(actual) => {
+ // This branch can only be reached if at least one
+ // concurrent thread called `wake`. In this
+ // case, `actual` **must** be `REGISTERING |
+ // `WAKING`.
+ debug_assert_eq!(actual, REGISTERING | WAKING);
+
+ // Take the waker to wake once the atomic operation has
+ // completed.
+ let waker = (*self.waker.get()).take().unwrap();
+
+ // We need to return to WAITING state (clear our lock and
+ // concurrent WAKING flag). This needs to acquire all
+ // WAKING fetch_or releases and it needs to release our
+ // update to self.waker, so we need a `swap` operation.
+ self.state.swap(WAITING, AcqRel);
+
+ // memory ordering: we acquired the state for all
+ // concurrent wakes, but future wakes might still
+ // need to wake us in case we can't make progress
+ // from the pending wakes.
+ //
+ // So we simply schedule to come back later (we could
+ // also simply leave the registration in place above).
+ waker.wake();
+ }
+ }
+ }
+ }
+ WAKING => {
+ // Currently in the process of waking the task, i.e.,
+ // `wake` is currently being called on the old task handle.
+ //
+ // memory ordering: we acquired the state for all
+ // concurrent wakes, but future wakes might still
+ // need to wake us in case we can't make progress
+ // from the pending wakes.
+ //
+ // So we simply schedule to come back later (we
+ // could also spin here trying to acquire the lock
+ // to register).
+ waker.wake_by_ref();
+ }
+ state => {
+ // In this case, a concurrent thread is holding the
+ // "registering" lock. This probably indicates a bug in the
+ // caller's code as racing to call `register` doesn't make much
+ // sense.
+ //
+ // memory ordering: don't care. a concurrent register() is going
+ // to succeed and provide proper memory ordering.
+ //
+ // We just want to maintain memory safety. It is ok to drop the
+ // call to `register`.
+ debug_assert!(state == REGISTERING || state == REGISTERING | WAKING);
+ }
+ }
+ }
+
+ /// Calls `wake` on the last `Waker` passed to `register`.
+ ///
+ /// If `register` has not been called yet, then this does nothing.
+ pub fn wake(&self) {
+ if let Some(waker) = self.take() {
+ waker.wake();
+ }
+ }
+
+ /// Returns the last `Waker` passed to `register`, so that the user can wake it.
+ ///
+ ///
+ /// Sometimes, just waking the AtomicWaker is not fine grained enough. This allows the user
+ /// to take the waker and then wake it separately, rather than performing both steps in one
+ /// atomic action.
+ ///
+ /// If a waker has not been registered, this returns `None`.
+ pub fn take(&self) -> Option<Waker> {
+ // AcqRel ordering is used in order to acquire the value of the `task`
+ // cell as well as to establish a `release` ordering with whatever
+ // memory the `AtomicWaker` is associated with.
+ match self.state.fetch_or(WAKING, AcqRel) {
+ WAITING => {
+ // The waking lock has been acquired.
+ let waker = unsafe { (*self.waker.get()).take() };
+
+ // Release the lock
+ self.state.fetch_and(!WAKING, Release);
+
+ waker
+ }
+ state => {
+ // There is a concurrent thread currently updating the
+ // associated task.
+ //
+ // Nothing more to do as the `WAKING` bit has been set. It
+ // doesn't matter if there are concurrent registering threads or
+ // not.
+ //
+ debug_assert!(
+ state == REGISTERING || state == REGISTERING | WAKING || state == WAKING
+ );
+ None
+ }
+ }
+ }
+}
+
+impl Default for AtomicWaker {
+ fn default() -> Self {
+ Self::new()
+ }
+}
+
+impl fmt::Debug for AtomicWaker {
+ fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+ write!(f, "AtomicWaker")
+ }
+}
+
+unsafe impl Send for AtomicWaker {}
+unsafe impl Sync for AtomicWaker {}
diff --git a/external/vendor/futures-core/src/task/__internal/mod.rs b/external/vendor/futures-core/src/task/__internal/mod.rs
new file mode 100644
index 0000000..c248742
--- /dev/null
+++ b/external/vendor/futures-core/src/task/__internal/mod.rs
@@ -0,0 +1,7 @@
+#[cfg_attr(target_os = "none", cfg(any(target_has_atomic = "ptr", feature = "portable-atomic")))]
+mod atomic_waker;
+#[cfg_attr(
+ target_os = "none",
+ cfg(any(target_has_atomic = "ptr", feature = "portable-atomic"))
+)]
+pub use self::atomic_waker::AtomicWaker;
diff --git a/external/vendor/futures-core/src/task/mod.rs b/external/vendor/futures-core/src/task/mod.rs
new file mode 100644
index 0000000..19e4eae
--- /dev/null
+++ b/external/vendor/futures-core/src/task/mod.rs
@@ -0,0 +1,10 @@
+//! Task notification.
+
+#[macro_use]
+mod poll;
+
+#[doc(hidden)]
+pub mod __internal;
+
+#[doc(no_inline)]
+pub use core::task::{Context, Poll, RawWaker, RawWakerVTable, Waker};
diff --git a/external/vendor/futures-core/src/task/poll.rs b/external/vendor/futures-core/src/task/poll.rs
new file mode 100644
index 0000000..607e78e
--- /dev/null
+++ b/external/vendor/futures-core/src/task/poll.rs
@@ -0,0 +1,12 @@
+/// Extracts the successful type of a `Poll<T>`.
+///
+/// This macro bakes in propagation of `Pending` signals by returning early.
+#[macro_export]
+macro_rules! ready {
+ ($e:expr $(,)?) => {
+ match $e {
+ $crate::task::Poll::Ready(t) => t,
+ $crate::task::Poll::Pending => return $crate::task::Poll::Pending,
+ }
+ };
+}
diff --git a/external/vendor/futures-lite/.cargo-checksum.json b/external/vendor/futures-lite/.cargo-checksum.json
new file mode 100644
index 0000000..7cefc3f
--- /dev/null
+++ b/external/vendor/futures-lite/.cargo-checksum.json
@@ -0,0 +1 @@
+{"files":{".cargo_vcs_info.json":"2fab79ec87b12bb8b09541146b57440a1d141b9f4f153b1d0219dffeb6712d27","CHANGELOG.md":"deed0a442b53d578f397083a5a1034a6bc546e948957218efe0f027d32a65769","Cargo.lock":"8eb703a4cb18ec5224283c75bc06d2102f8359a84d4927737b58f6c0f7bfa8c4","Cargo.toml":"43218a6e46a95142ade2400a01937357a4d220dbf467462739b578c43a09848a","Cargo.toml.orig":"fa4ba70dab5fa8a5613b23238524d3a548786b1e27a2fc23925a023a2e54365f","FEATURES.md":"2904922c4e7f09ac4105c99e931594467804c80deb1f5eaec3b296386c332278","LICENSE-APACHE":"a60eea817514531668d7e00765731449fe14d059d3249e0bc93b36de45f759f2","LICENSE-MIT":"23f18e03dc49df91622fe2a76176497404e46ced8a715d9d2b67a7446571cca3","LICENSE-THIRD-PARTY":"6226d0632e2e1a80c23597e964da9812ae193c535fe058154afb034e94167aa5","README.md":"0242ae3d8e434dbafb916997adfc8fea1b7c49fa9b2de25f515a69c523e8bba7","src/future.rs":"23b6f79a4366860dbfafb38308147f923196766facc9c9b1a436f1331b94f8e9","src/io.rs":"fb5301c11bf40b2b9508942043671ec12cb9c0c0e13adbe2283980d0a7cd852e","src/lib.rs":"277b557fe7e7835cceb2d856c8e6adf20be8d6602b65ce0feecea38bd6d384d6","src/prelude.rs":"ab0ec9c549e9104c84ae64dfba03d4078702a0897c5ac3e15770dd692c6b23e3","src/stream.rs":"87f6fb8404dbd7114a8973f20798eaa022a8be3c30a307a0fa33a3bd1de56df5"},"package":"f78e10609fe0e0b3f4157ffab1876319b5b0db102a2c60dc4626306dc46b44ad"}
\ No newline at end of file
diff --git a/external/vendor/futures-lite/.cargo_vcs_info.json b/external/vendor/futures-lite/.cargo_vcs_info.json
new file mode 100644
index 0000000..a13e34c
--- /dev/null
+++ b/external/vendor/futures-lite/.cargo_vcs_info.json
@@ -0,0 +1,6 @@
+{
+ "git": {
+ "sha1": "226ce18976d8714d6bd9700b61dcc81d7200bc9a"
+ },
+ "path_in_vcs": ""
+}
\ No newline at end of file
diff --git a/external/vendor/futures-lite/CHANGELOG.md b/external/vendor/futures-lite/CHANGELOG.md
new file mode 100644
index 0000000..fe00145
--- /dev/null
+++ b/external/vendor/futures-lite/CHANGELOG.md
@@ -0,0 +1,212 @@
+# Version 2.6.1
+
+- Fix docs for `once_future` and `stop_after_future`. (#131)
+
+# Version 2.6.0
+
+- Add `Stream::map_while()` combinator. (#116)
+- Add list of excluded features to crate documentation. (#112)
+- Update docs for `AsyncRead::read_exact` (#121)
+
+# Version 2.5.0
+
+- Remove `Unpin` bound from the `Lines` I/O adapter. (#113)
+
+# Version 2.4.0
+
+- Add a "fuse" method that makes it so a `Future` returns `Poll::Pending`
+ forever after it returns `Poll::Pending` once. (#101)
+- Add a "stop_after_future" function that allows for running a `Stream` until a
+ `Future` completes. (#103)
+- Make it so `Zip`/`TryZip` drop completed futures. (#106)
+
+# Version 2.3.0
+
+- Add `StreamExt::drain` for draining objects from a `Stream` without waiting (#70).
+
+# Version 2.2.0
+
+- Relax `Unpin` bounds on `io::copy`. (#87)
+- Implement `size_hint` for `stream::Filter`. (#88)
+- Relax MSRV to 1.60. (#90)
+
+# Version 2.1.0
+
+- Make it so `read_line` and other futures use a naive implementation of byte
+ searching unless the `memchr` feature is enabled. This prevents needing to
+ compile the `memchr` crate unless it is desired. (#77)
+
+# Version 2.0.1
+
+- Remove dependency on the `waker-fn` crate. (#81)
+
+# Version 2.0.0
+
+- **Breaking:** Expose `future::{ready, pending}` from `core` instead of defining
+ our own. (#73)
+- **Breaking:** The `TryZip` and `Zip` combinators are modified to have a cleaner
+ API, where generic constraints are not necessary on the structure itself at the
+ cost of additional generics. (#74)
+- Add a way to use racey futures on `no_std` by providing your own seed. (#75)
+
+# Version 1.13.0
+
+- Unbind Debug implementations of BufReader and BufWriter. (#49)
+- Add the once_future() combinator. (#59)
+- Add a combinator for temporarily using an AsyncRead/AsyncWrite as Read/Write. (#62)
+- Implement more methods for stream::BlockOn. (#68)
+
+# Version 1.12.0
+
+- Implement `BufRead` for `BlockOn`
+
+# Version 1.11.3
+
+- Update `pin-project-lite`.
+
+# Version 1.11.2
+
+- Improve docs for `ready!`.
+
+# Version 1.11.1
+
+- Fix some typos.
+
+# Version 1.11.0
+
+- Add the new `prelude` module.
+- Deprecate trait re-exports in the root module.
+
+# Version 1.10.1
+
+- Fix compilation errors with Rust 1.42.0 and 1.45.2
+
+# Version 1.10.0
+
+- Add `io::split()`.
+
+# Version 1.9.0
+
+- Add `FutureExt::poll()`.
+- Add `StreamExt::poll_next()`.
+- Add `AsyncBufReadExt::fill_buf()`.
+- Add `AsyncBufReadExt::consume()`.
+
+# Version 1.8.0
+
+- Add `BoxedReader` and `BoxedWriter`.
+
+# Version 1.7.0
+
+- Implement `AsyncRead` for `Bytes`.
+- Add `StreamExt::then()`.
+
+# Version 1.6.0
+
+- Add `FutureExt::catch_unwind()`.
+
+# Version 1.5.0
+
+- Add `stream::race()` and `StreamExt::race()`.
+
+# Version 1.4.0
+
+- Add `alloc` Cargo feature.
+
+# Version 1.3.0
+
+- Add `future::or()`.
+- Add `FutureExt::race()`.
+- Disable `waker-fn` dependency on `#![no_std]` targets.
+
+# Version 1.2.0
+
+- Fix compilation errors on `#![no_std]` systems.
+- Add `StreamExt::try_next()`.
+- Add `StreamExt::partition()`.
+- Add `StreamExt::for_each()`.
+- Add `StreamExt::try_for_each()`.
+- Add `StreamExt::zip()`.
+- Add `StreamExt::unzip()`.
+- Add `StreamExt::nth()`.
+- Add `StreamExt::last()`.
+- Add `StreamExt::find()`.
+- Add `StreamExt::find_map()`.
+- Add `StreamExt::position()`.
+- Add `StreamExt::all()`.
+- Add `StreamExt::any()`.
+- Add `StreamExt::scan()`.
+- Add `StreamExt::flat_map()`.
+- Add `StreamExt::flatten()`.
+- Add `StreamExt::skip()`.
+- Add `StreamExt::skip_while()`.
+
+# Version 1.1.0
+
+- Add `StreamExt::take()`.
+- Add `StreamExt::take_while()`.
+- Add `StreamExt::step_by()`.
+- Add `StreamExt::fuse()`.
+- Add `StreamExt::chain()`.
+- Add `StreamExt::cloned()`.
+- Add `StreamExt::copied()`.
+- Add `StreamExt::cycle()`.
+- Add `StreamExt::enumeraate()`.
+- Add `StreamExt::inspect()`.
+- Parametrize `FutureExt::boxed()` and `FutureExt::boxed_local()` over a lifetime.
+- Parametrize `StreamExt::boxed()` and `StreamExt::boxed_local()` over a lifetime.
+
+# Version 1.0.0
+
+- Add `StreamExt::map()`.
+- Add `StreamExt::count()`.
+- Add `StreamExt::filter()`.
+- Add `StreamExt::filter_map()`.
+- Rename `future::join()` to `future::zip()`.
+- Rename `future::try_join()` to `future::try_zip()`.
+
+# Version 0.1.11
+
+- Update `parking` to v2.0.0
+
+# Version 0.1.10
+
+- Add `AssertAsync`.
+
+# Version 0.1.9
+
+- Add `FutureExt::or()`.
+- Put `#[must_use]` on all futures and streams.
+
+# Version 0.1.8
+
+- Fix lints about unsafe code.
+
+# Version 0.1.7
+
+- Add blocking APIs (`block_on()` and `BlockOn`).
+
+# Version 0.1.6
+
+- Add `boxed()`, `boxed_local()`, `Boxed`, and `BoxedLocal`.
+
+# Version 0.1.5
+
+- Add `fold()` and `try_fold()`.
+
+# Version 0.1.4
+
+- Add `future::race()`.
+- Fix a bug in `BufReader`.
+
+# Version 0.1.3
+
+- Add `future::join()`, `future::try_join()`, and `AsyncWriteExt::close()`.
+
+# Version 0.1.2
+
+- Lots of new APIs.
+
+# Version 0.1.1
+
+- Initial version
diff --git a/external/vendor/futures-lite/Cargo.lock b/external/vendor/futures-lite/Cargo.lock
new file mode 100644
index 0000000..91d6ac4
--- /dev/null
+++ b/external/vendor/futures-lite/Cargo.lock
@@ -0,0 +1,74 @@
+# This file is automatically @generated by Cargo.
+# It is not intended for manual editing.
+version = 3
+
+[[package]]
+name = "fastrand"
+version = "2.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "37909eebbb50d72f9059c3b6d82c0463f2ff062c9e95845c43a6c9c0355411be"
+
+[[package]]
+name = "futures-core"
+version = "0.3.31"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "05f29059c0c2090612e8d742178b0580d2dc940c837851ad723096f87af6663e"
+
+[[package]]
+name = "futures-io"
+version = "0.3.31"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9e5c1b78ca4aae1ac06c48a526a655760685149f0d465d21f37abfe57ce075c6"
+
+[[package]]
+name = "futures-lite"
+version = "2.6.1"
+dependencies = [
+ "fastrand",
+ "futures-core",
+ "futures-io",
+ "memchr",
+ "parking",
+ "pin-project-lite",
+ "spin_on",
+ "waker-fn",
+]
+
+[[package]]
+name = "memchr"
+version = "2.7.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "32a282da65faaf38286cf3be983213fcf1d2e2a58700e808f83f4ea9a4804bc0"
+
+[[package]]
+name = "parking"
+version = "2.2.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f38d5652c16fde515bb1ecef450ab0f6a219d619a7274976324d5e377f7dceba"
+
+[[package]]
+name = "pin-project-lite"
+version = "0.2.16"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "3b3cff922bd51709b605d9ead9aa71031d81447142d828eb4a6eba76fe619f9b"
+
+[[package]]
+name = "pin-utils"
+version = "0.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8b870d8c151b6f2fb93e84a13146138f05d02ed11c7e7c54f8826aaaf7c9f184"
+
+[[package]]
+name = "spin_on"
+version = "0.1.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "076e103ed41b9864aa838287efe5f4e3a7a0362dd00671ae62a212e5e4612da2"
+dependencies = [
+ "pin-utils",
+]
+
+[[package]]
+name = "waker-fn"
+version = "1.2.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "317211a0dc0ceedd78fb2ca9a44aed3d7b9b26f81870d485c07122b4350673b7"
diff --git a/external/vendor/futures-lite/Cargo.toml b/external/vendor/futures-lite/Cargo.toml
new file mode 100644
index 0000000..4461c03
--- /dev/null
+++ b/external/vendor/futures-lite/Cargo.toml
@@ -0,0 +1,90 @@
+# THIS FILE IS AUTOMATICALLY GENERATED BY CARGO
+#
+# When uploading crates to the registry Cargo will automatically
+# "normalize" Cargo.toml files for maximal compatibility
+# with all versions of Cargo and also rewrite `path` dependencies
+# to registry (e.g., crates.io) dependencies.
+#
+# If you are reading this file be aware that the original Cargo.toml
+# will likely look very different (and much more reasonable).
+# See Cargo.toml.orig for the original contents.
+
+[package]
+edition = "2021"
+rust-version = "1.60"
+name = "futures-lite"
+version = "2.6.1"
+authors = [
+ "Stjepan Glavina <stjepang@gmail.com>",
+ "Contributors to futures-rs",
+]
+build = false
+exclude = ["/.*"]
+autolib = false
+autobins = false
+autoexamples = false
+autotests = false
+autobenches = false
+description = "Futures, streams, and async I/O combinators"
+homepage = "https://github.com/smol-rs/futures-lite"
+documentation = "https://docs.rs/futures-lite"
+readme = "README.md"
+keywords = [
+ "asynchronous",
+ "futures",
+ "async",
+]
+categories = [
+ "asynchronous",
+ "concurrency",
+]
+license = "Apache-2.0 OR MIT"
+repository = "https://github.com/smol-rs/futures-lite"
+
+[features]
+alloc = []
+default = [
+ "race",
+ "std",
+]
+race = ["fastrand"]
+std = [
+ "alloc",
+ "fastrand/std",
+ "futures-io",
+ "parking",
+]
+
+[lib]
+name = "futures_lite"
+path = "src/lib.rs"
+
+[dependencies.fastrand]
+version = "2.0.0"
+optional = true
+default-features = false
+
+[dependencies.futures-core]
+version = "0.3.5"
+default-features = false
+
+[dependencies.futures-io]
+version = "0.3.5"
+optional = true
+
+[dependencies.memchr]
+version = "2.3.3"
+optional = true
+
+[dependencies.parking]
+version = "2.2.0"
+optional = true
+
+[dependencies.pin-project-lite]
+version = "0.2.0"
+
+[dev-dependencies.spin_on]
+version = "0.1.0"
+
+[dev-dependencies.waker-fn]
+version = "1.0.0"
diff --git a/external/vendor/futures-lite/Cargo.toml.orig b/external/vendor/futures-lite/Cargo.toml.orig
new file mode 100644
index 0000000..44df93b
--- /dev/null
+++ b/external/vendor/futures-lite/Cargo.toml.orig
@@ -0,0 +1,38 @@
+[package]
+name = "futures-lite"
+# When publishing a new version:
+# - Update CHANGELOG.md
+# - Create "v2.x.y" git tag
+version = "2.6.1"
+authors = [
+ "Stjepan Glavina <stjepang@gmail.com>",
+ "Contributors to futures-rs",
+]
+edition = "2021"
+rust-version = "1.60"
+description = "Futures, streams, and async I/O combinators"
+license = "Apache-2.0 OR MIT"
+repository = "https://github.com/smol-rs/futures-lite"
+homepage = "https://github.com/smol-rs/futures-lite"
+documentation = "https://docs.rs/futures-lite"
+keywords = ["asynchronous", "futures", "async"]
+categories = ["asynchronous", "concurrency"]
+exclude = ["/.*"]
+
+[features]
+default = ["race", "std"]
+std = ["alloc", "fastrand/std", "futures-io", "parking"]
+alloc = []
+race = ["fastrand"]
+
+[dependencies]
+fastrand = { version = "2.0.0", optional = true, default-features = false }
+futures-core = { version = "0.3.5", default-features = false }
+futures-io = { version = "0.3.5", optional = true }
+memchr = { version = "2.3.3", optional = true }
+parking = { version = "2.2.0", optional = true }
+pin-project-lite = "0.2.0"
+
+[dev-dependencies]
+spin_on = "0.1.0"
+waker-fn = "1.0.0" # used in doctests
diff --git a/external/vendor/futures-lite/FEATURES.md b/external/vendor/futures-lite/FEATURES.md
new file mode 100644
index 0000000..a92fc02
--- /dev/null
+++ b/external/vendor/futures-lite/FEATURES.md
@@ -0,0 +1,275 @@
+# Intentional Occlusions from `futures-lite`
+
+[`futures-lite`] has an API that is deliberately smaller than the [`futures`]
+crate. This allows it to compile significantly faster and have fewer
+dependencies.
+
+This fact does not mean that [`futures-lite`] is not open to new feature
+requests. However it does mean that any proposed new features are subject to
+scrutiny to determine whether or not they are truly necessary for this crate.
+In many cases there are much simpler ways to implement these features, or they
+would be a much better fit for an external crate.
+
+This document aims to describe all intentional feature occlusions and provide
+suggestions for how these features can be used in the context of
+[`futures-lite`]. If you have a feature request that you believe does not fall
+under any of the following occlusions, please open an issue on the
+[official `futures-lite` bug tracker](https://github.com/smol-rs/futures-lite/issues).
+
+## Simple Combinators
+
+In general, anything that can be implemented in terms of `async`/`await` syntax
+is not implemented in [`futures-lite`]. This is done to encourage the use of
+modern `async`/`await` syntax rather than [`futures`] v1.0 combinator chaining.
+
+As an example, take the [`map`] method in [`futures`]. It takes a future and
+processes its output through a closure.
+
+```rust
+let my_future = async { 1 };
+
+// Add one to the result of `my_future`.
+let mapped_future = my_future.map(|x| x + 1);
+
+assert_eq!(mapped_future.await, 2);
+```
+
+However, this does not need to be implemented in the form of a combinator. With
+`async`/`await` syntax, you can simply `await` on `my_future` in an `async`
+block, then process its output. The following code is equivalent to the above,
+but doesn't use a combinator.
+
+```rust
+let my_future = async { 1 };
+
+// Add one to the result of `my_future`.
+let mapped_future = async move { my_future.await + 1 };
+
+assert_eq!(mapped_future.await, 2);
+```
+
+By not implementing combinators that can be implemented in terms of `async`,
+[`futures-lite`] has a significantly smaller API that still has roughly the
+same amount of power as [`futures`].
+
+As part of this policy, the [`TryFutureExt`] trait is not implemented. All of
+its methods can be implemented by just using `async`/`await` combined with
+other simpler future combinators. For instance, consider [`and_then`]:
+
+```rust
+let my_future = async { Ok(2) };
+
+let and_then = my_future.and_then(|x| async move {
+ Ok(x + 1)
+});
+
+assert_eq!(and_then.await.unwrap(), 3);
+```
+
+This can be implemented with an `async` block and the normal `and_then`
+combinator.
+
+```rust
+let my_future = async { Ok(2) };
+
+let and_then = async move {
+ let x = my_future.await;
+ x.and_then(|x| x + 1)
+};
+
+assert_eq!(and_then.await.unwrap(), 3);
+```
+
+One drawback of this approach is that `async` blocks are not named types. So
+if a trait (like [`Service`]) requires a named future type it cannot be
+returned.
+
+```rust
+impl Service for MyService {
+ type Future = /* ??? */;
+
+ fn call(&mut self) -> Self::Future {
+ async { 1 + 1 }
+ }
+}
+```
+
+One possible solution is to box the future and return a dynamic dispatch
+object, but in many cases this adds non trivial overhead.
+
+```rust
+impl Service for MyService {
+ type Future = Pin<Box<dyn Future<Output = i32>>>;
+
+ fn call(&mut self) -> Self::Future {
+ async { 1 + 1 }.boxed_local()
+ }
+}
+```
+
+This problem is expected to be resolved in the future, thanks to
+[`async` fn in traits] and [TAIT]. At this point we would rather wait for these
+better solutions than significantly expand [`futures-lite`]'s API. If this is a
+deal breaker for you, [`futures`] is probably better for your use case.
+
+## Asynchronous Closures
+
+As a pattern, most combinators in [`futures-lite`] take regular closures rather
+than `async` closures. For example:
+
+```rust
+// In `futures`, the `all` combinator takes a closure returning a future.
+my_stream.all(|x| async move { x > 5 }).await;
+
+// In `futures-lite`, the `all` combinator just takes a closure.
+my_stream.all(|x| x > 5).await;
+```
+
+This strategy is taken for two primary reasons.
+
+First of all, it is significantly simpler to implement. Since we don't need to
+keep track of whether we are currently `poll`ing a future or not it makes the
+combinators an order of magnitude easier to write.
+
+Second of all it avoids the common [`futures`] wart of needing to pass trivial
+values into `async move { ... }` or `future::ready(...)` for the vast
+majority of operations.
+
+For futures, combinators that would normally require `async` closures can
+usually be implemented in terms of `async`/`await`. See the above section for
+more information on that. For streams, the [`then`] combinator is one of the
+few that actually takes an `async` closure, and can therefore be used to
+implement operations that would normally need `async` closures.
+
+```rust
+// In `futures`.
+my_stream.all(|x| my_async_fn(x)).await;
+
+// In `futures-lite`, use `then` and pass the result to `all`.
+my_stream.then(|x| my_async_fn(x)).all(|pass| pass).await;
+```
+
+## Higher-Order Concurrency
+
+[`futures`] provides a number of primitives and combinators that allow for
+polling a significant number of futures at once. Examples of this include
+[`for_each_concurrent`] and [`FuturesUnordered`].
+
+[`futures-lite`] provides simple primitives like [`race`] and [`zip`]. However
+these don't really scale to handling more than two futures at once. It has
+been proposed in the past to add deeper concurrency primitives to
+[`futures-lite`]. However our current stance is that such primitives would
+represent a significant uptick in complexity and thus is better suited to
+other crates.
+
+[`futures-concurrency`] provides a number of simple APIs for dealing with
+fixed numbers of futures. For example, here is an example for waiting on
+multiple futures to complete.
+
+```rust
+let (a, b, c) = /* assume these are all futures */;
+
+// futures
+let (x, y, z) = join!(a, b, c);
+
+// futures-concurrency
+use futures_concurrency::prelude::*;
+let (x, y, z) = (a, b, c).join().await;
+```
+
+For large or variable numbers of futures it is recommended to use an executor
+instead. [`smol`] provides both an [`Executor`] and a [`LocalExecutor`]
+depending on the flavor of your program.
+
+@notgull has a [blog post](https://notgull.net/futures-concurrency-in-smol/)
+describing this in greater detail.
+
+To explicitly answer a frequently asked question, the popular [`select`] macro
+can be implemented by using simple `async`/`await` and a race combinator.
+
+```rust
+let (a, b, c) = /* assume these are all futures */;
+
+// futures
+let x = select! {
+ a_res = a => a_res + 1,
+ _ = b => 0,
+ c_res = c => c_res + 3,
+};
+
+// futures-concurrency
+let x = (
+ async move { a.await + 1 },
+ async move { b.await; 0 },
+ async move { c.await + 3 }
+).race().await;
+```
+
+## Sink Trait
+
+[`futures`] offers a [`Sink`] trait that is in many ways the opposite of the
+[`Stream`] trait. Rather than asynchronously producing values, the point of the
+[`Sink`] is to asynchronously receive values.
+
+[`futures-lite`] and the rest of [`smol`] intentionally does not support the
+[`Sink`] trait. [`Sink`] is a relic from the old [`futures`] v0.1 days where
+I/O was tied directly into the API. The `Error` subtype is wholly unnecessary
+and makes the API significantly harder to use. In addition the multi-call
+requirement makes the API harder to both use and implement. It increases the
+complexity of any futures that use it significantly, and its API necessitates
+that implementors have an internal buffer for objects.
+
+In short, the ideal [`Sink`] API would be if it was replaced with this trait.
+
+*Sidenote: [`Stream`], [`AsyncRead`] and [`AsyncWrite`] suffer from this same
+problem to an extent. I think they could also be fixed by transforming their
+`fn poll_[X]` functions into `async fn [X]` functions. However their APIs are
+not broken to the point that [`Sink`]'s is.*
+
+In order to avoid relying on a broken API, [`futures-lite`] does not import
+[`Sink`] or expose any APIs that build upon [`Sink`]. Unfortunately some crates
+make their only accessible API the [`Sink`] call. Ideally instead they would
+just have an `async fn send()` function.
+
+## Out-of-scope modules
+
+[`futures`] provides several sets of tools that are out of scope for
+[`futures-lite`]. Usually these are implemented in external crates, some of
+which depend on [`futures-lite`] themselves. Here are examples of these
+primitives:
+
+- **Channels:** [`async-channel`] provides an asynchronous MPMC channel, while
+ [`oneshot`] provides an asynchronous oneshot channel.
+- **Mutex:** [`async-lock`] provides asynchronous mutexes, alongside other
+ locking primitives.
+- **Atomic Wakers:** [`atomic-waker`] provides standalone atomic wakers.
+- **Executors:** [`async-executor`] provides [`Executor`] to replace
+ `ThreadPool` and [`LocalExecutor`] to replace `LocalPool`.
+
+[`smol`]: https://crates.io/crates/smol
+[`futures-lite`]: https://crates.io/crates/futures-lite
+[`futures`]: https://crates.io/crates/futures
+[`map`]: https://docs.rs/futures/latest/futures/future/trait.FutureExt.html#method.map
+[`TryFutureExt`]: https://docs.rs/futures/latest/futures/future/trait.TryFutureExt.html
+[`and_then`]: https://docs.rs/futures/latest/futures/future/trait.TryFutureExt.html#method.and_then
+[`Service`]: https://docs.rs/tower-service/latest/tower_service/trait.Service.html
+[`async` fn in traits]: https://blog.rust-lang.org/2023/12/21/async-fn-rpit-in-traits.html
+[TAIT]: https://rust-lang.github.io/impl-trait-initiative/explainer/tait.html
+[`then`]: https://docs.rs/futures-lite/latest/futures_lite/stream/trait.StreamExt.html#method.then
+[`FuturesUnordered`]: https://docs.rs/futures/latest/futures/stream/struct.FuturesUnordered.html
+[`for_each_concurrent`]: https://docs.rs/futures/latest/futures/stream/trait.StreamExt.html#method.for_each_concurrent
+[`race`]: https://docs.rs/futures-lite/latest/futures_lite/future/fn.race.html
+[`zip`]: https://docs.rs/futures-lite/latest/futures_lite/future/fn.zip.html
+[`futures-concurrency`]: https://docs.rs/futures-concurrency/latest/futures_concurrency/
+[`Executor`]: https://docs.rs/async-executor/latest/async_executor/struct.Executor.html
+[`LocalExecutor`]: https://docs.rs/async-executor/latest/async_executor/struct.LocalExecutor.html
+[`select`]: https://docs.rs/futures/latest/futures/macro.select.html
+[`Sink`]: https://docs.rs/futures/latest/futures/sink/trait.Sink.html
+[`Stream`]: https://docs.rs/futures-core/latest/futures_core/stream/trait.Stream.html
+[`AsyncRead`]: https://docs.rs/futures-io/latest/futures_io/trait.AsyncRead.html
+[`AsyncWrite`]: https://docs.rs/futures-io/latest/futures_io/trait.AsyncWrite.html
+[`async-channel`]: https://crates.io/crates/async-channel
+[`async-lock`]: https://crates.io/crates/async-lock
+[`async-executor`]: https://crates.io/crates/async-executor
+[`oneshot`]: https://crates.io/crates/oneshot
+[`atomic-waker`]: https://crates.io/crates/atomic-waker
diff --git a/external/vendor/futures-lite/LICENSE-APACHE b/external/vendor/futures-lite/LICENSE-APACHE
new file mode 100644
index 0000000..16fe87b
--- /dev/null
+++ b/external/vendor/futures-lite/LICENSE-APACHE
@@ -0,0 +1,201 @@
+ Apache License
+ Version 2.0, January 2004
+ http://www.apache.org/licenses/
+
+TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
+
+1. Definitions.
+
+ "License" shall mean the terms and conditions for use, reproduction,
+ and distribution as defined by Sections 1 through 9 of this document.
+
+ "Licensor" shall mean the copyright owner or entity authorized by
+ the copyright owner that is granting the License.
+
+ "Legal Entity" shall mean the union of the acting entity and all
+ other entities that control, are controlled by, or are under common
+ control with that entity. For the purposes of this definition,
+ "control" means (i) the power, direct or indirect, to cause the
+ direction or management of such entity, whether by contract or
+ otherwise, or (ii) ownership of fifty percent (50%) or more of the
+ outstanding shares, or (iii) beneficial ownership of such entity.
+
+ "You" (or "Your") shall mean an individual or Legal Entity
+ exercising permissions granted by this License.
+
+ "Source" form shall mean the preferred form for making modifications,
+ including but not limited to software source code, documentation
+ source, and configuration files.
+
+ "Object" form shall mean any form resulting from mechanical
+ transformation or translation of a Source form, including but
+ not limited to compiled object code, generated documentation,
+ and conversions to other media types.
+
+ "Work" shall mean the work of authorship, whether in Source or
+ Object form, made available under the License, as indicated by a
+ copyright notice that is included in or attached to the work
+ (an example is provided in the Appendix below).
+
+ "Derivative Works" shall mean any work, whether in Source or Object
+ form, that is based on (or derived from) the Work and for which the
+ editorial revisions, annotations, elaborations, or other modifications
+ represent, as a whole, an original work of authorship. For the purposes
+ of this License, Derivative Works shall not include works that remain
+ separable from, or merely link (or bind by name) to the interfaces of,
+ the Work and Derivative Works thereof.
+
+ "Contribution" shall mean any work of authorship, including
+ the original version of the Work and any modifications or additions
+ to that Work or Derivative Works thereof, that is intentionally
+ submitted to Licensor for inclusion in the Work by the copyright owner
+ or by an individual or Legal Entity authorized to submit on behalf of
+ the copyright owner. For the purposes of this definition, "submitted"
+ means any form of electronic, verbal, or written communication sent
+ to the Licensor or its representatives, including but not limited to
+ communication on electronic mailing lists, source code control systems,
+ and issue tracking systems that are managed by, or on behalf of, the
+ Licensor for the purpose of discussing and improving the Work, but
+ excluding communication that is conspicuously marked or otherwise
+ designated in writing by the copyright owner as "Not a Contribution."
+
+ "Contributor" shall mean Licensor and any individual or Legal Entity
+ on behalf of whom a Contribution has been received by Licensor and
+ subsequently incorporated within the Work.
+
+2. Grant of Copyright License. Subject to the terms and conditions of
+ this License, each Contributor hereby grants to You a perpetual,
+ worldwide, non-exclusive, no-charge, royalty-free, irrevocable
+ copyright license to reproduce, prepare Derivative Works of,
+ publicly display, publicly perform, sublicense, and distribute the
+ Work and such Derivative Works in Source or Object form.
+
+3. Grant of Patent License. Subject to the terms and conditions of
+ this License, each Contributor hereby grants to You a perpetual,
+ worldwide, non-exclusive, no-charge, royalty-free, irrevocable
+ (except as stated in this section) patent license to make, have made,
+ use, offer to sell, sell, import, and otherwise transfer the Work,
+ where such license applies only to those patent claims licensable
+ by such Contributor that are necessarily infringed by their
+ Contribution(s) alone or by combination of their Contribution(s)
+ with the Work to which such Contribution(s) was submitted. If You
+ institute patent litigation against any entity (including a
+ cross-claim or counterclaim in a lawsuit) alleging that the Work
+ or a Contribution incorporated within the Work constitutes direct
+ or contributory patent infringement, then any patent licenses
+ granted to You under this License for that Work shall terminate
+ as of the date such litigation is filed.
+
+4. Redistribution. You may reproduce and distribute copies of the
+ Work or Derivative Works thereof in any medium, with or without
+ modifications, and in Source or Object form, provided that You
+ meet the following conditions:
+
+ (a) You must give any other recipients of the Work or
+ Derivative Works a copy of this License; and
+
+ (b) You must cause any modified files to carry prominent notices
+ stating that You changed the files; and
+
+ (c) You must retain, in the Source form of any Derivative Works
+ that You distribute, all copyright, patent, trademark, and
+ attribution notices from the Source form of the Work,
+ excluding those notices that do not pertain to any part of
+ the Derivative Works; and
+
+ (d) If the Work includes a "NOTICE" text file as part of its
+ distribution, then any Derivative Works that You distribute must
+ include a readable copy of the attribution notices contained
+ within such NOTICE file, excluding those notices that do not
+ pertain to any part of the Derivative Works, in at least one
+ of the following places: within a NOTICE text file distributed
+ as part of the Derivative Works; within the Source form or
+ documentation, if provided along with the Derivative Works; or,
+ within a display generated by the Derivative Works, if and
+ wherever such third-party notices normally appear. The contents
+ of the NOTICE file are for informational purposes only and
+ do not modify the License. You may add Your own attribution
+ notices within Derivative Works that You distribute, alongside
+ or as an addendum to the NOTICE text from the Work, provided
+ that such additional attribution notices cannot be construed
+ as modifying the License.
+
+ You may add Your own copyright statement to Your modifications and
+ may provide additional or different license terms and conditions
+ for use, reproduction, or distribution of Your modifications, or
+ for any such Derivative Works as a whole, provided Your use,
+ reproduction, and distribution of the Work otherwise complies with
+ the conditions stated in this License.
+
+5. Submission of Contributions. Unless You explicitly state otherwise,
+ any Contribution intentionally submitted for inclusion in the Work
+ by You to the Licensor shall be under the terms and conditions of
+ this License, without any additional terms or conditions.
+ Notwithstanding the above, nothing herein shall supersede or modify
+ the terms of any separate license agreement you may have executed
+ with Licensor regarding such Contributions.
+
+6. Trademarks. This License does not grant permission to use the trade
+ names, trademarks, service marks, or product names of the Licensor,
+ except as required for reasonable and customary use in describing the
+ origin of the Work and reproducing the content of the NOTICE file.
+
+7. Disclaimer of Warranty. Unless required by applicable law or
+ agreed to in writing, Licensor provides the Work (and each
+ Contributor provides its Contributions) on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+ implied, including, without limitation, any warranties or conditions
+ of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A
+ PARTICULAR PURPOSE. You are solely responsible for determining the
+ appropriateness of using or redistributing the Work and assume any
+ risks associated with Your exercise of permissions under this License.
+
+8. Limitation of Liability. In no event and under no legal theory,
+ whether in tort (including negligence), contract, or otherwise,
+ unless required by applicable law (such as deliberate and grossly
+ negligent acts) or agreed to in writing, shall any Contributor be
+ liable to You for damages, including any direct, indirect, special,
+ incidental, or consequential damages of any character arising as a
+ result of this License or out of the use or inability to use the
+ Work (including but not limited to damages for loss of goodwill,
+ work stoppage, computer failure or malfunction, or any and all
+ other commercial damages or losses), even if such Contributor
+ has been advised of the possibility of such damages.
+
+9. Accepting Warranty or Additional Liability. While redistributing
+ the Work or Derivative Works thereof, You may choose to offer,
+ and charge a fee for, acceptance of support, warranty, indemnity,
+ or other liability obligations and/or rights consistent with this
+ License. However, in accepting such obligations, You may act only
+ on Your own behalf and on Your sole responsibility, not on behalf
+ of any other Contributor, and only if You agree to indemnify,
+ defend, and hold each Contributor harmless for any liability
+ incurred by, or claims asserted against, such Contributor by reason
+ of your accepting any such warranty or additional liability.
+
+END OF TERMS AND CONDITIONS
+
+APPENDIX: How to apply the Apache License to your work.
+
+ To apply the Apache License to your work, attach the following
+ boilerplate notice, with the fields enclosed by brackets "[]"
+ replaced with your own identifying information. (Don't include
+ the brackets!) The text should be enclosed in the appropriate
+ comment syntax for the file format. We also recommend that a
+ file or class name and description of purpose be included on the
+ same "printed page" as the copyright notice for easier
+ identification within third-party archives.
+
+Copyright [yyyy] [name of copyright owner]
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
diff --git a/external/vendor/futures-lite/LICENSE-MIT b/external/vendor/futures-lite/LICENSE-MIT
new file mode 100644
index 0000000..31aa793
--- /dev/null
+++ b/external/vendor/futures-lite/LICENSE-MIT
@@ -0,0 +1,23 @@
+Permission is hereby granted, free of charge, to any
+person obtaining a copy of this software and associated
+documentation files (the "Software"), to deal in the
+Software without restriction, including without
+limitation the rights to use, copy, modify, merge,
+publish, distribute, sublicense, and/or sell copies of
+the Software, and to permit persons to whom the Software
+is furnished to do so, subject to the following
+conditions:
+
+The above copyright notice and this permission notice
+shall be included in all copies or substantial portions
+of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF
+ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED
+TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A
+PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT
+SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY
+CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
+OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR
+IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
+DEALINGS IN THE SOFTWARE.
diff --git a/external/vendor/futures-lite/LICENSE-THIRD-PARTY b/external/vendor/futures-lite/LICENSE-THIRD-PARTY
new file mode 100644
index 0000000..aa77d25
--- /dev/null
+++ b/external/vendor/futures-lite/LICENSE-THIRD-PARTY
@@ -0,0 +1,45 @@
+===============================================================================
+
+Copyright (c) 2016 Alex Crichton
+Copyright (c) 2017 The Tokio Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+
+===============================================================================
+
+Copyright (c) 2016 Alex Crichton
+Copyright (c) 2017 The Tokio Authors
+
+Permission is hereby granted, free of charge, to any
+person obtaining a copy of this software and associated
+documentation files (the "Software"), to deal in the
+Software without restriction, including without
+limitation the rights to use, copy, modify, merge,
+publish, distribute, sublicense, and/or sell copies of
+the Software, and to permit persons to whom the Software
+is furnished to do so, subject to the following
+conditions:
+
+The above copyright notice and this permission notice
+shall be included in all copies or substantial portions
+of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF
+ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED
+TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A
+PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT
+SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY
+CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
+OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR
+IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
+DEALINGS IN THE SOFTWARE.
diff --git a/external/vendor/futures-lite/README.md b/external/vendor/futures-lite/README.md
new file mode 100644
index 0000000..138be6a
--- /dev/null
+++ b/external/vendor/futures-lite/README.md
@@ -0,0 +1,51 @@
+# futures-lite
+
+[](
+https://github.com/smol-rs/futures-lite/actions)
+[](
+https://github.com/smol-rs/futures-lite)
+[](
+https://crates.io/crates/futures-lite)
+[](
+https://docs.rs/futures-lite)
+
+A lightweight async prelude.
+
+This crate is a subset of [futures] that compiles an order of magnitude faster, fixes minor
+warts in its API, fills in some obvious gaps, and removes almost all unsafe code from it.
+
+In short, this crate aims to be more enjoyable than [futures] but still fully compatible with
+it.
+
+The API for this crate is intentionally constrained. Please consult the
+[features list] for APIs that are occluded from this crate.
+
+[futures]: https://docs.rs/futures
+[features list]: https://github.com/smol-rs/futures-lite/blob/master/FEATURES.md
+
+## Examples
+
+```rust
+use futures_lite::future;
+
+fn main() {
+ future::block_on(async {
+ println!("Hello world!");
+ })
+}
+```
+
+## License
+
+Licensed under either of
+
+ * Apache License, Version 2.0 ([LICENSE-APACHE](LICENSE-APACHE) or http://www.apache.org/licenses/LICENSE-2.0)
+ * MIT license ([LICENSE-MIT](LICENSE-MIT) or http://opensource.org/licenses/MIT)
+
+at your option.
+
+#### Contribution
+
+Unless you explicitly state otherwise, any contribution intentionally submitted
+for inclusion in the work by you, as defined in the Apache-2.0 license, shall be
+dual licensed as above, without any additional terms or conditions.
diff --git a/external/vendor/futures-lite/src/future.rs b/external/vendor/futures-lite/src/future.rs
new file mode 100644
index 0000000..70956f1
--- /dev/null
+++ b/external/vendor/futures-lite/src/future.rs
@@ -0,0 +1,830 @@
+//! Combinators for the [`Future`] trait.
+//!
+//! # Examples
+//!
+//! ```
+//! use futures_lite::future;
+//!
+//! # spin_on::spin_on(async {
+//! for step in 0..3 {
+//! println!("step {}", step);
+//!
+//! // Give other tasks a chance to run.
+//! future::yield_now().await;
+//! }
+//! # });
+//! ```
+
+#[doc(no_inline)]
+pub use core::future::{pending, ready, Future, Pending, Ready};
+
+use core::fmt;
+use core::pin::Pin;
+use core::task::{Context, Poll};
+
+#[cfg(feature = "alloc")]
+use alloc::boxed::Box;
+
+#[cfg(feature = "std")]
+use std::{
+ any::Any,
+ panic::{catch_unwind, AssertUnwindSafe, UnwindSafe},
+ thread_local,
+};
+
+#[cfg(feature = "race")]
+use fastrand::Rng;
+use pin_project_lite::pin_project;
+
+/// Blocks the current thread on a future.
+///
+/// # Examples
+///
+/// ```
+/// use futures_lite::future;
+///
+/// let val = future::block_on(async {
+/// 1 + 2
+/// });
+///
+/// assert_eq!(val, 3);
+/// ```
+#[cfg(feature = "std")]
+pub fn block_on<T>(future: impl Future<Output = T>) -> T {
+ use core::cell::RefCell;
+ use core::task::Waker;
+
+ use parking::Parker;
+
+ // Pin the future on the stack.
+ crate::pin!(future);
+
+ // Creates a parker and an associated waker that unparks it.
+ fn parker_and_waker() -> (Parker, Waker) {
+ let parker = Parker::new();
+ let unparker = parker.unparker();
+ let waker = Waker::from(unparker);
+ (parker, waker)
+ }
+
+ thread_local! {
+ // Cached parker and waker for efficiency.
+ static CACHE: RefCell<(Parker, Waker)> = RefCell::new(parker_and_waker());
+ }
+
+ CACHE.with(|cache| {
+ // Try grabbing the cached parker and waker.
+ let tmp_cached;
+ let tmp_fresh;
+ let (parker, waker) = match cache.try_borrow_mut() {
+ Ok(cache) => {
+ // Use the cached parker and waker.
+ tmp_cached = cache;
+ &*tmp_cached
+ }
+ Err(_) => {
+ // Looks like this is a recursive `block_on()` call.
+ // Create a fresh parker and waker.
+ tmp_fresh = parker_and_waker();
+ &tmp_fresh
+ }
+ };
+
+ let cx = &mut Context::from_waker(waker);
+ // Keep polling until the future is ready.
+ loop {
+ match future.as_mut().poll(cx) {
+ Poll::Ready(output) => return output,
+ Poll::Pending => parker.park(),
+ }
+ }
+ })
+}
+
+/// Polls a future just once and returns an [`Option`] with the result.
+///
+/// # Examples
+///
+/// ```
+/// use futures_lite::future;
+///
+/// # spin_on::spin_on(async {
+/// assert_eq!(future::poll_once(future::pending::<()>()).await, None);
+/// assert_eq!(future::poll_once(future::ready(42)).await, Some(42));
+/// # })
+/// ```
+pub fn poll_once<T, F>(f: F) -> PollOnce<F>
+where
+ F: Future<Output = T>,
+{
+ PollOnce { f }
+}
+
+pin_project! {
+ /// Future for the [`poll_once()`] function.
+ #[must_use = "futures do nothing unless you `.await` or poll them"]
+ pub struct PollOnce<F> {
+ #[pin]
+ f: F,
+ }
+}
+
+impl<F> fmt::Debug for PollOnce<F> {
+ fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+ f.debug_struct("PollOnce").finish()
+ }
+}
+
+impl<T, F> Future for PollOnce<F>
+where
+ F: Future<Output = T>,
+{
+ type Output = Option<T>;
+
+ fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+ match self.project().f.poll(cx) {
+ Poll::Ready(t) => Poll::Ready(Some(t)),
+ Poll::Pending => Poll::Ready(None),
+ }
+ }
+}
+
+/// Creates a future from a function returning [`Poll`].
+///
+/// # Examples
+///
+/// ```
+/// use futures_lite::future;
+/// use std::task::{Context, Poll};
+///
+/// # spin_on::spin_on(async {
+/// fn f(_: &mut Context<'_>) -> Poll<i32> {
+/// Poll::Ready(7)
+/// }
+///
+/// assert_eq!(future::poll_fn(f).await, 7);
+/// # })
+/// ```
+pub fn poll_fn<T, F>(f: F) -> PollFn<F>
+where
+ F: FnMut(&mut Context<'_>) -> Poll<T>,
+{
+ PollFn { f }
+}
+
+pin_project! {
+ /// Future for the [`poll_fn()`] function.
+ #[must_use = "futures do nothing unless you `.await` or poll them"]
+ pub struct PollFn<F> {
+ f: F,
+ }
+}
+
+impl<F> fmt::Debug for PollFn<F> {
+ fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+ f.debug_struct("PollFn").finish()
+ }
+}
+
+impl<T, F> Future for PollFn<F>
+where
+ F: FnMut(&mut Context<'_>) -> Poll<T>,
+{
+ type Output = T;
+
+ fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<T> {
+ let this = self.project();
+ (this.f)(cx)
+ }
+}
+
+/// Wakes the current task and returns [`Poll::Pending`] once.
+///
+/// This function is useful when we want to cooperatively give time to the task scheduler. It is
+/// generally a good idea to yield inside loops because that way we make sure long-running tasks
+/// don't prevent other tasks from running.
+///
+/// # Examples
+///
+/// ```
+/// use futures_lite::future;
+///
+/// # spin_on::spin_on(async {
+/// future::yield_now().await;
+/// # })
+/// ```
+pub fn yield_now() -> YieldNow {
+ YieldNow(false)
+}
+
+/// Future for the [`yield_now()`] function.
+#[derive(Debug)]
+#[must_use = "futures do nothing unless you `.await` or poll them"]
+pub struct YieldNow(bool);
+
+impl Future for YieldNow {
+ type Output = ();
+
+ fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+ if !self.0 {
+ self.0 = true;
+ cx.waker().wake_by_ref();
+ Poll::Pending
+ } else {
+ Poll::Ready(())
+ }
+ }
+}
+
+/// Joins two futures, waiting for both to complete.
+///
+/// # Examples
+///
+/// ```
+/// use futures_lite::future;
+///
+/// # spin_on::spin_on(async {
+/// let a = async { 1 };
+/// let b = async { 2 };
+///
+/// assert_eq!(future::zip(a, b).await, (1, 2));
+/// # })
+/// ```
+pub fn zip<F1, F2>(future1: F1, future2: F2) -> Zip<F1, F2>
+where
+ F1: Future,
+ F2: Future,
+{
+ Zip {
+ future1: Some(future1),
+ future2: Some(future2),
+ output1: None,
+ output2: None,
+ }
+}
+
+pin_project! {
+ /// Future for the [`zip()`] function.
+ #[derive(Debug)]
+ #[must_use = "futures do nothing unless you `.await` or poll them"]
+ pub struct Zip<F1, F2>
+ where
+ F1: Future,
+ F2: Future,
+ {
+ #[pin]
+ future1: Option<F1>,
+ output1: Option<F1::Output>,
+ #[pin]
+ future2: Option<F2>,
+ output2: Option<F2::Output>,
+ }
+}
+
+/// Extracts the contents of two options and zips them, handling `(Some(_), None)` cases
+fn take_zip_from_parts<T1, T2>(o1: &mut Option<T1>, o2: &mut Option<T2>) -> Poll<(T1, T2)> {
+ match (o1.take(), o2.take()) {
+ (Some(t1), Some(t2)) => Poll::Ready((t1, t2)),
+ (o1x, o2x) => {
+ *o1 = o1x;
+ *o2 = o2x;
+ Poll::Pending
+ }
+ }
+}
+
+impl<F1, F2> Future for Zip<F1, F2>
+where
+ F1: Future,
+ F2: Future,
+{
+ type Output = (F1::Output, F2::Output);
+
+ fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+ let mut this = self.project();
+
+ if let Some(future) = this.future1.as_mut().as_pin_mut() {
+ if let Poll::Ready(out) = future.poll(cx) {
+ *this.output1 = Some(out);
+ this.future1.set(None);
+ }
+ }
+
+ if let Some(future) = this.future2.as_mut().as_pin_mut() {
+ if let Poll::Ready(out) = future.poll(cx) {
+ *this.output2 = Some(out);
+ this.future2.set(None);
+ }
+ }
+
+ take_zip_from_parts(this.output1, this.output2)
+ }
+}
+
+/// Joins two fallible futures, waiting for both to complete or one of them to error.
+///
+/// # Examples
+///
+/// ```
+/// use futures_lite::future;
+///
+/// # spin_on::spin_on(async {
+/// let a = async { Ok::<i32, i32>(1) };
+/// let b = async { Err::<i32, i32>(2) };
+///
+/// assert_eq!(future::try_zip(a, b).await, Err(2));
+/// # })
+/// ```
+pub fn try_zip<T1, T2, E, F1, F2>(future1: F1, future2: F2) -> TryZip<F1, T1, F2, T2>
+where
+ F1: Future<Output = Result<T1, E>>,
+ F2: Future<Output = Result<T2, E>>,
+{
+ TryZip {
+ future1: Some(future1),
+ future2: Some(future2),
+ output1: None,
+ output2: None,
+ }
+}
+
+pin_project! {
+ /// Future for the [`try_zip()`] function.
+ #[derive(Debug)]
+ #[must_use = "futures do nothing unless you `.await` or poll them"]
+ pub struct TryZip<F1, T1, F2, T2> {
+ #[pin]
+ future1: Option<F1>,
+ output1: Option<T1>,
+ #[pin]
+ future2: Option<F2>,
+ output2: Option<T2>,
+ }
+}
+
+impl<T1, T2, E, F1, F2> Future for TryZip<F1, T1, F2, T2>
+where
+ F1: Future<Output = Result<T1, E>>,
+ F2: Future<Output = Result<T2, E>>,
+{
+ type Output = Result<(T1, T2), E>;
+
+ fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+ let mut this = self.project();
+
+ if let Some(future) = this.future1.as_mut().as_pin_mut() {
+ if let Poll::Ready(out) = future.poll(cx) {
+ match out {
+ Ok(t) => {
+ *this.output1 = Some(t);
+ this.future1.set(None);
+ }
+ Err(err) => return Poll::Ready(Err(err)),
+ }
+ }
+ }
+
+ if let Some(future) = this.future2.as_mut().as_pin_mut() {
+ if let Poll::Ready(out) = future.poll(cx) {
+ match out {
+ Ok(t) => {
+ *this.output2 = Some(t);
+ this.future2.set(None);
+ }
+ Err(err) => return Poll::Ready(Err(err)),
+ }
+ }
+ }
+
+ take_zip_from_parts(this.output1, this.output2).map(Ok)
+ }
+}
+
+/// Returns the result of the future that completes first, preferring `future1` if both are ready.
+///
+/// If you need to treat the two futures fairly without a preference for either, use the [`race()`]
+/// function or the [`FutureExt::race()`] method.
+///
+/// # Examples
+///
+/// ```
+/// use futures_lite::future::{self, pending, ready};
+///
+/// # spin_on::spin_on(async {
+/// assert_eq!(future::or(ready(1), pending()).await, 1);
+/// assert_eq!(future::or(pending(), ready(2)).await, 2);
+///
+/// // The first future wins.
+/// assert_eq!(future::or(ready(1), ready(2)).await, 1);
+/// # })
+/// ```
+pub fn or<T, F1, F2>(future1: F1, future2: F2) -> Or<F1, F2>
+where
+ F1: Future<Output = T>,
+ F2: Future<Output = T>,
+{
+ Or { future1, future2 }
+}
+
+pin_project! {
+ /// Future for the [`or()`] function and the [`FutureExt::or()`] method.
+ #[derive(Debug)]
+ #[must_use = "futures do nothing unless you `.await` or poll them"]
+ pub struct Or<F1, F2> {
+ #[pin]
+ future1: F1,
+ #[pin]
+ future2: F2,
+ }
+}
+
+impl<T, F1, F2> Future for Or<F1, F2>
+where
+ F1: Future<Output = T>,
+ F2: Future<Output = T>,
+{
+ type Output = T;
+
+ fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+ let this = self.project();
+
+ if let Poll::Ready(t) = this.future1.poll(cx) {
+ return Poll::Ready(t);
+ }
+ if let Poll::Ready(t) = this.future2.poll(cx) {
+ return Poll::Ready(t);
+ }
+ Poll::Pending
+ }
+}
+
+/// Fuse a future such that `poll` will never again be called once it has
+/// completed. This method can be used to turn any `Future` into a
+/// `FusedFuture`.
+///
+/// Normally, once a future has returned `Poll::Ready` from `poll`,
+/// any further calls could exhibit bad behavior such as blocking
+/// forever, panicking, never returning, etc. If it is known that `poll`
+/// may be called too often then this method can be used to ensure that it
+/// has defined semantics.
+///
+/// If a `fuse`d future is `poll`ed after having returned `Poll::Ready`
+/// previously, it will return `Poll::Pending`, from `poll` again (and will
+/// continue to do so for all future calls to `poll`).
+///
+/// This combinator will drop the underlying future as soon as it has been
+/// completed to ensure resources are reclaimed as soon as possible.
+pub fn fuse<F>(future: F) -> Fuse<F>
+where
+ F: Future + Sized,
+{
+ Fuse::new(future)
+}
+
+pin_project! {
+ /// [`Future`] for the [`fuse`] method.
+ #[derive(Debug)]
+ #[must_use = "futures do nothing unless you `.await` or poll them"]
+ pub struct Fuse<Fut> {
+ #[pin]
+ inner: Option<Fut>,
+ }
+}
+
+impl<Fut> Fuse<Fut> {
+ fn new(f: Fut) -> Self {
+ Self { inner: Some(f) }
+ }
+}
+
+impl<Fut: Future> Future for Fuse<Fut> {
+ type Output = Fut::Output;
+
+ fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Fut::Output> {
+ match self
+ .as_mut()
+ .project()
+ .inner
+ .as_pin_mut()
+ .map(|f| f.poll(cx))
+ {
+ Some(Poll::Ready(output)) => {
+ self.project().inner.set(None);
+ Poll::Ready(output)
+ }
+
+ Some(Poll::Pending) | None => Poll::Pending,
+ }
+ }
+}
+
+/// Returns the result of the future that completes first, with no preference if both are ready.
+///
+/// Each time [`Race`] is polled, the two inner futures are polled in random order. Therefore, no
+/// future takes precedence over the other if both can complete at the same time.
+///
+/// If you have preference for one of the futures, use the [`or()`] function or the
+/// [`FutureExt::or()`] method.
+///
+/// # Examples
+///
+/// ```
+/// use futures_lite::future::{self, pending, ready};
+///
+/// # spin_on::spin_on(async {
+/// assert_eq!(future::race(ready(1), pending()).await, 1);
+/// assert_eq!(future::race(pending(), ready(2)).await, 2);
+///
+/// // One of the two futures is randomly chosen as the winner.
+/// let res = future::race(ready(1), ready(2)).await;
+/// # })
+/// ```
+#[cfg(all(feature = "race", feature = "std"))]
+pub fn race<T, F1, F2>(future1: F1, future2: F2) -> Race<F1, F2>
+where
+ F1: Future<Output = T>,
+ F2: Future<Output = T>,
+{
+ Race {
+ future1,
+ future2,
+ rng: Rng::new(),
+ }
+}
+
+/// Race two futures but with a predefined random seed.
+///
+/// This function is identical to [`race`], but instead of using a random seed from a thread-local
+/// RNG, it allows the user to provide a seed. It is useful for when you already have a source of
+/// randomness available, or if you want to use a fixed seed.
+///
+/// See documentation of the [`race`] function for features and caveats.
+///
+/// # Examples
+///
+/// ```
+/// use futures_lite::future::{self, pending, ready};
+///
+/// // A fixed seed is used, so the result is deterministic.
+/// const SEED: u64 = 0x42;
+///
+/// # spin_on::spin_on(async {
+/// assert_eq!(future::race_with_seed(ready(1), pending(), SEED).await, 1);
+/// assert_eq!(future::race_with_seed(pending(), ready(2), SEED).await, 2);
+///
+/// // One of the two futures is randomly chosen as the winner.
+/// let res = future::race_with_seed(ready(1), ready(2), SEED).await;
+/// # })
+/// ```
+#[cfg(feature = "race")]
+pub fn race_with_seed<T, F1, F2>(future1: F1, future2: F2, seed: u64) -> Race<F1, F2>
+where
+ F1: Future<Output = T>,
+ F2: Future<Output = T>,
+{
+ Race {
+ future1,
+ future2,
+ rng: Rng::with_seed(seed),
+ }
+}
+
+#[cfg(feature = "race")]
+pin_project! {
+ /// Future for the [`race()`] function and the [`FutureExt::race()`] method.
+ #[derive(Debug)]
+ #[must_use = "futures do nothing unless you `.await` or poll them"]
+ pub struct Race<F1, F2> {
+ #[pin]
+ future1: F1,
+ #[pin]
+ future2: F2,
+ rng: Rng,
+ }
+}
+
+#[cfg(feature = "race")]
+impl<T, F1, F2> Future for Race<F1, F2>
+where
+ F1: Future<Output = T>,
+ F2: Future<Output = T>,
+{
+ type Output = T;
+
+ fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+ let this = self.project();
+
+ if this.rng.bool() {
+ if let Poll::Ready(t) = this.future1.poll(cx) {
+ return Poll::Ready(t);
+ }
+ if let Poll::Ready(t) = this.future2.poll(cx) {
+ return Poll::Ready(t);
+ }
+ } else {
+ if let Poll::Ready(t) = this.future2.poll(cx) {
+ return Poll::Ready(t);
+ }
+ if let Poll::Ready(t) = this.future1.poll(cx) {
+ return Poll::Ready(t);
+ }
+ }
+ Poll::Pending
+ }
+}
+
+#[cfg(feature = "std")]
+pin_project! {
+ /// Future for the [`FutureExt::catch_unwind()`] method.
+ #[derive(Debug)]
+ #[must_use = "futures do nothing unless you `.await` or poll them"]
+ pub struct CatchUnwind<F> {
+ #[pin]
+ inner: F,
+ }
+}
+
+#[cfg(feature = "std")]
+impl<F: Future + UnwindSafe> Future for CatchUnwind<F> {
+ type Output = Result<F::Output, Box<dyn Any + Send>>;
+
+ fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+ let this = self.project();
+ catch_unwind(AssertUnwindSafe(|| this.inner.poll(cx)))?.map(Ok)
+ }
+}
+
+/// Type alias for `Pin<Box<dyn Future<Output = T> + Send + 'static>>`.
+///
+/// # Examples
+///
+/// ```
+/// use futures_lite::future::{self, FutureExt};
+///
+/// // These two lines are equivalent:
+/// let f1: future::Boxed<i32> = async { 1 + 2 }.boxed();
+/// let f2: future::Boxed<i32> = Box::pin(async { 1 + 2 });
+/// ```
+#[cfg(feature = "alloc")]
+pub type Boxed<T> = Pin<Box<dyn Future<Output = T> + Send + 'static>>;
+
+/// Type alias for `Pin<Box<dyn Future<Output = T> + 'static>>`.
+///
+/// # Examples
+///
+/// ```
+/// use futures_lite::future::{self, FutureExt};
+///
+/// // These two lines are equivalent:
+/// let f1: future::BoxedLocal<i32> = async { 1 + 2 }.boxed_local();
+/// let f2: future::BoxedLocal<i32> = Box::pin(async { 1 + 2 });
+/// ```
+#[cfg(feature = "alloc")]
+pub type BoxedLocal<T> = Pin<Box<dyn Future<Output = T> + 'static>>;
+
+/// Extension trait for [`Future`].
+pub trait FutureExt: Future {
+ /// A convenience for calling [`Future::poll()`] on `!`[`Unpin`] types.
+ fn poll(&mut self, cx: &mut Context<'_>) -> Poll<Self::Output>
+ where
+ Self: Unpin,
+ {
+ Future::poll(Pin::new(self), cx)
+ }
+
+ /// Returns the result of `self` or `other` future, preferring `self` if both are ready.
+ ///
+ /// If you need to treat the two futures fairly without a preference for either, use the
+ /// [`race()`] function or the [`FutureExt::race()`] method.
+ ///
+ /// # Examples
+ ///
+ /// ```
+ /// use futures_lite::future::{pending, ready, FutureExt};
+ ///
+ /// # spin_on::spin_on(async {
+ /// assert_eq!(ready(1).or(pending()).await, 1);
+ /// assert_eq!(pending().or(ready(2)).await, 2);
+ ///
+ /// // The first future wins.
+ /// assert_eq!(ready(1).or(ready(2)).await, 1);
+ /// # })
+ /// ```
+ fn or<F>(self, other: F) -> Or<Self, F>
+ where
+ Self: Sized,
+ F: Future<Output = Self::Output>,
+ {
+ Or {
+ future1: self,
+ future2: other,
+ }
+ }
+
+ /// Returns the result of `self` or `other` future, with no preference if both are ready.
+ ///
+ /// Each time [`Race`] is polled, the two inner futures are polled in random order. Therefore,
+ /// no future takes precedence over the other if both can complete at the same time.
+ ///
+ /// If you have preference for one of the futures, use the [`or()`] function or the
+ /// [`FutureExt::or()`] method.
+ ///
+ /// # Examples
+ ///
+ /// ```
+ /// use futures_lite::future::{pending, ready, FutureExt};
+ ///
+ /// # spin_on::spin_on(async {
+ /// assert_eq!(ready(1).race(pending()).await, 1);
+ /// assert_eq!(pending().race(ready(2)).await, 2);
+ ///
+ /// // One of the two futures is randomly chosen as the winner.
+ /// let res = ready(1).race(ready(2)).await;
+ /// # })
+ /// ```
+ #[cfg(all(feature = "std", feature = "race"))]
+ fn race<F>(self, other: F) -> Race<Self, F>
+ where
+ Self: Sized,
+ F: Future<Output = Self::Output>,
+ {
+ Race {
+ future1: self,
+ future2: other,
+ rng: Rng::new(),
+ }
+ }
+
+ /// Catches panics while polling the future.
+ ///
+ /// # Examples
+ ///
+ /// ```
+ /// use futures_lite::future::FutureExt;
+ ///
+ /// # spin_on::spin_on(async {
+ /// let fut1 = async {}.catch_unwind();
+ /// let fut2 = async { panic!() }.catch_unwind();
+ ///
+ /// assert!(fut1.await.is_ok());
+ /// assert!(fut2.await.is_err());
+ /// # })
+ /// ```
+ #[cfg(feature = "std")]
+ fn catch_unwind(self) -> CatchUnwind<Self>
+ where
+ Self: Sized + UnwindSafe,
+ {
+ CatchUnwind { inner: self }
+ }
+
+ /// Boxes the future and changes its type to `dyn Future + Send + 'a`.
+ ///
+ /// # Examples
+ ///
+ /// ```
+ /// use futures_lite::future::{self, FutureExt};
+ ///
+ /// # spin_on::spin_on(async {
+ /// let a = future::ready('a');
+ /// let b = future::pending();
+ ///
+ /// // Futures of different types can be stored in
+ /// // the same collection when they are boxed:
+ /// let futures = vec![a.boxed(), b.boxed()];
+ /// # })
+ /// ```
+ #[cfg(feature = "alloc")]
+ fn boxed<'a>(self) -> Pin<Box<dyn Future<Output = Self::Output> + Send + 'a>>
+ where
+ Self: Sized + Send + 'a,
+ {
+ Box::pin(self)
+ }
+
+ /// Boxes the future and changes its type to `dyn Future + 'a`.
+ ///
+ /// # Examples
+ ///
+ /// ```
+ /// use futures_lite::future::{self, FutureExt};
+ ///
+ /// # spin_on::spin_on(async {
+ /// let a = future::ready('a');
+ /// let b = future::pending();
+ ///
+ /// // Futures of different types can be stored in
+ /// // the same collection when they are boxed:
+ /// let futures = vec![a.boxed_local(), b.boxed_local()];
+ /// # })
+ /// ```
+ #[cfg(feature = "alloc")]
+ fn boxed_local<'a>(self) -> Pin<Box<dyn Future<Output = Self::Output> + 'a>>
+ where
+ Self: Sized + 'a,
+ {
+ Box::pin(self)
+ }
+}
+
+impl<F: Future + ?Sized> FutureExt for F {}
diff --git a/external/vendor/futures-lite/src/io.rs b/external/vendor/futures-lite/src/io.rs
new file mode 100644
index 0000000..2a8bc87
--- /dev/null
+++ b/external/vendor/futures-lite/src/io.rs
@@ -0,0 +1,3102 @@
+//! Tools and combinators for I/O.
+//!
+//! # Examples
+//!
+//! ```
+//! use futures_lite::io::{self, AsyncReadExt};
+//!
+//! # spin_on::spin_on(async {
+//! let input: &[u8] = b"hello";
+//! let mut reader = io::BufReader::new(input);
+//!
+//! let mut contents = String::new();
+//! reader.read_to_string(&mut contents).await?;
+//! # std::io::Result::Ok(()) });
+//! ```
+
+#[doc(no_inline)]
+pub use std::io::{Error, ErrorKind, Result, SeekFrom};
+
+#[doc(no_inline)]
+pub use futures_io::{AsyncBufRead, AsyncRead, AsyncSeek, AsyncWrite};
+
+use std::borrow::{Borrow, BorrowMut};
+use std::boxed::Box;
+use std::cmp;
+use std::fmt;
+use std::future::Future;
+use std::io::{IoSlice, IoSliceMut};
+use std::mem;
+use std::pin::Pin;
+use std::string::String;
+use std::sync::{Arc, Mutex};
+use std::task::{Context, Poll};
+use std::vec;
+use std::vec::Vec;
+
+use futures_core::stream::Stream;
+use pin_project_lite::pin_project;
+
+use crate::future;
+use crate::ready;
+
+const DEFAULT_BUF_SIZE: usize = 8 * 1024;
+
+/// Copies the entire contents of a reader into a writer.
+///
+/// This function will read data from `reader` and write it into `writer` in a streaming fashion
+/// until `reader` returns EOF.
+///
+/// On success, returns the total number of bytes copied.
+///
+/// # Examples
+///
+/// ```
+/// use futures_lite::io::{self, BufReader, BufWriter};
+///
+/// # spin_on::spin_on(async {
+/// let input: &[u8] = b"hello";
+/// let reader = BufReader::new(input);
+///
+/// let mut output = Vec::new();
+/// let writer = BufWriter::new(&mut output);
+///
+/// io::copy(reader, writer).await?;
+/// # std::io::Result::Ok(()) });
+/// ```
+pub async fn copy<R, W>(reader: R, writer: W) -> Result<u64>
+where
+ R: AsyncRead,
+ W: AsyncWrite,
+{
+ pin_project! {
+ struct CopyFuture<R, W> {
+ #[pin]
+ reader: R,
+ #[pin]
+ writer: W,
+ amt: u64,
+ }
+ }
+
+ impl<R, W> Future for CopyFuture<R, W>
+ where
+ R: AsyncBufRead,
+ W: AsyncWrite,
+ {
+ type Output = Result<u64>;
+
+ fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+ let mut this = self.project();
+ loop {
+ let buffer = ready!(this.reader.as_mut().poll_fill_buf(cx))?;
+ if buffer.is_empty() {
+ ready!(this.writer.as_mut().poll_flush(cx))?;
+ return Poll::Ready(Ok(*this.amt));
+ }
+
+ let i = ready!(this.writer.as_mut().poll_write(cx, buffer))?;
+ if i == 0 {
+ return Poll::Ready(Err(ErrorKind::WriteZero.into()));
+ }
+ *this.amt += i as u64;
+ this.reader.as_mut().consume(i);
+ }
+ }
+ }
+
+ let future = CopyFuture {
+ reader: BufReader::new(reader),
+ writer,
+ amt: 0,
+ };
+ future.await
+}
+
+/// Asserts that a type implementing [`std::io`] traits can be used as an async type.
+///
+/// The underlying I/O handle should never block nor return the [`ErrorKind::WouldBlock`] error.
+/// This is usually the case for in-memory buffered I/O.
+///
+/// # Examples
+///
+/// ```
+/// use futures_lite::io::{AssertAsync, AsyncReadExt};
+///
+/// let reader: &[u8] = b"hello";
+///
+/// # spin_on::spin_on(async {
+/// let mut async_reader = AssertAsync::new(reader);
+/// let mut contents = String::new();
+///
+/// // This line works in async manner - note that there is await:
+/// async_reader.read_to_string(&mut contents).await?;
+/// # std::io::Result::Ok(()) });
+/// ```
+#[derive(Debug, Copy, Clone, Eq, PartialEq, Ord, PartialOrd, Hash)]
+pub struct AssertAsync<T>(T);
+
+impl<T> Unpin for AssertAsync<T> {}
+
+impl<T> AssertAsync<T> {
+ /// Wraps an I/O handle implementing [`std::io`] traits.
+ ///
+ /// # Examples
+ ///
+ /// ```
+ /// use futures_lite::io::AssertAsync;
+ ///
+ /// let reader: &[u8] = b"hello";
+ ///
+ /// let async_reader = AssertAsync::new(reader);
+ /// ```
+ #[inline(always)]
+ pub fn new(io: T) -> Self {
+ AssertAsync(io)
+ }
+
+ /// Gets a reference to the inner I/O handle.
+ ///
+ /// # Examples
+ ///
+ /// ```
+ /// use futures_lite::io::AssertAsync;
+ ///
+ /// let reader: &[u8] = b"hello";
+ ///
+ /// let async_reader = AssertAsync::new(reader);
+ /// let r = async_reader.get_ref();
+ /// ```
+ #[inline(always)]
+ pub fn get_ref(&self) -> &T {
+ &self.0
+ }
+
+ /// Gets a mutable reference to the inner I/O handle.
+ ///
+ /// # Examples
+ ///
+ /// ```
+ /// use futures_lite::io::AssertAsync;
+ ///
+ /// let reader: &[u8] = b"hello";
+ ///
+ /// let mut async_reader = AssertAsync::new(reader);
+ /// let r = async_reader.get_mut();
+ /// ```
+ #[inline(always)]
+ pub fn get_mut(&mut self) -> &mut T {
+ &mut self.0
+ }
+
+ /// Extracts the inner I/O handle.
+ ///
+ /// # Examples
+ ///
+ /// ```
+ /// use futures_lite::io::AssertAsync;
+ ///
+ /// let reader: &[u8] = b"hello";
+ ///
+ /// let async_reader = AssertAsync::new(reader);
+ /// let inner = async_reader.into_inner();
+ /// ```
+ #[inline(always)]
+ pub fn into_inner(self) -> T {
+ self.0
+ }
+}
+
+fn assert_async_wrapio<F, T>(mut f: F) -> Poll<std::io::Result<T>>
+where
+ F: FnMut() -> std::io::Result<T>,
+{
+ loop {
+ match f() {
+ Err(err) if err.kind() == ErrorKind::Interrupted => {}
+ res => return Poll::Ready(res),
+ }
+ }
+}
+
+impl<T: std::io::Read> AsyncRead for AssertAsync<T> {
+ #[inline]
+ fn poll_read(
+ mut self: Pin<&mut Self>,
+ _: &mut Context<'_>,
+ buf: &mut [u8],
+ ) -> Poll<Result<usize>> {
+ assert_async_wrapio(move || self.0.read(buf))
+ }
+
+ #[inline]
+ fn poll_read_vectored(
+ mut self: Pin<&mut Self>,
+ _: &mut Context<'_>,
+ bufs: &mut [IoSliceMut<'_>],
+ ) -> Poll<Result<usize>> {
+ assert_async_wrapio(move || self.0.read_vectored(bufs))
+ }
+}
+
+impl<T: std::io::Write> AsyncWrite for AssertAsync<T> {
+ #[inline]
+ fn poll_write(
+ mut self: Pin<&mut Self>,
+ _: &mut Context<'_>,
+ buf: &[u8],
+ ) -> Poll<Result<usize>> {
+ assert_async_wrapio(move || self.0.write(buf))
+ }
+
+ #[inline]
+ fn poll_write_vectored(
+ mut self: Pin<&mut Self>,
+ _: &mut Context<'_>,
+ bufs: &[IoSlice<'_>],
+ ) -> Poll<Result<usize>> {
+ assert_async_wrapio(move || self.0.write_vectored(bufs))
+ }
+
+ #[inline]
+ fn poll_flush(mut self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<Result<()>> {
+ assert_async_wrapio(move || self.0.flush())
+ }
+
+ #[inline]
+ fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<()>> {
+ self.poll_flush(cx)
+ }
+}
+
+impl<T: std::io::Seek> AsyncSeek for AssertAsync<T> {
+ #[inline]
+ fn poll_seek(
+ mut self: Pin<&mut Self>,
+ _: &mut Context<'_>,
+ pos: SeekFrom,
+ ) -> Poll<Result<u64>> {
+ assert_async_wrapio(move || self.0.seek(pos))
+ }
+}
+
+/// A wrapper around a type that implements `AsyncRead` or `AsyncWrite` that converts `Pending`
+/// polls to `WouldBlock` errors.
+///
+/// This wrapper can be used as a compatibility layer between `AsyncRead` and `Read`, for types
+/// that take `Read` as a parameter.
+///
+/// # Examples
+///
+/// ```
+/// use std::io::Read;
+/// use std::task::{Poll, Context};
+///
+/// fn poll_for_io(cx: &mut Context<'_>) -> Poll<usize> {
+/// // Assume we have a library that's built around `Read` and `Write` traits.
+/// use cooltls::Session;
+///
+/// // We want to use it with our writer that implements `AsyncWrite`.
+/// let writer = Stream::new();
+///
+/// // First, we wrap our `Writer` with `AsyncAsSync` to convert `Pending` polls to `WouldBlock`.
+/// use futures_lite::io::AsyncAsSync;
+/// let writer = AsyncAsSync::new(cx, writer);
+///
+/// // Now, we can use it with `cooltls`.
+/// let mut session = Session::new(writer);
+///
+/// // Match on the result of `read()` and translate it to poll.
+/// match session.read(&mut [0; 1024]) {
+/// Ok(n) => Poll::Ready(n),
+/// Err(err) if err.kind() == std::io::ErrorKind::WouldBlock => Poll::Pending,
+/// Err(err) => panic!("unexpected error: {}", err),
+/// }
+/// }
+///
+/// // Usually, poll-based functions are best wrapped using `poll_fn`.
+/// use futures_lite::future::poll_fn;
+/// # futures_lite::future::block_on(async {
+/// poll_fn(|cx| poll_for_io(cx)).await;
+/// # });
+/// # struct Stream;
+/// # impl Stream {
+/// # fn new() -> Stream {
+/// # Stream
+/// # }
+/// # }
+/// # impl futures_lite::io::AsyncRead for Stream {
+/// # fn poll_read(self: std::pin::Pin<&mut Self>, _: &mut Context<'_>, _: &mut [u8]) -> Poll<std::io::Result<usize>> {
+/// # Poll::Ready(Ok(0))
+/// # }
+/// # }
+/// # mod cooltls {
+/// # pub struct Session<W> {
+/// # reader: W,
+/// # }
+/// # impl<W> Session<W> {
+/// # pub fn new(reader: W) -> Session<W> {
+/// # Session { reader }
+/// # }
+/// # }
+/// # impl<W: std::io::Read> std::io::Read for Session<W> {
+/// # fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
+/// # self.reader.read(buf)
+/// # }
+/// # }
+/// # }
+/// ```
+#[derive(Debug)]
+pub struct AsyncAsSync<'r, 'ctx, T> {
+ /// The context we are using to poll the future.
+ pub context: &'r mut Context<'ctx>,
+
+ /// The actual reader/writer we are wrapping.
+ pub inner: T,
+}
+
+impl<'r, 'ctx, T> AsyncAsSync<'r, 'ctx, T> {
+ /// Wraps an I/O handle implementing [`AsyncRead`] or [`AsyncWrite`] traits.
+ ///
+ /// # Examples
+ ///
+ /// ```
+ /// use futures_lite::io::AsyncAsSync;
+ /// use std::task::Context;
+ /// use waker_fn::waker_fn;
+ ///
+ /// let reader: &[u8] = b"hello";
+ /// let waker = waker_fn(|| {});
+ /// let mut context = Context::from_waker(&waker);
+ ///
+ /// let async_reader = AsyncAsSync::new(&mut context, reader);
+ /// ```
+ #[inline]
+ pub fn new(context: &'r mut Context<'ctx>, inner: T) -> Self {
+ AsyncAsSync { context, inner }
+ }
+
+ /// Attempt to shutdown the I/O handle.
+ ///
+ /// # Examples
+ ///
+ /// ```
+ /// use futures_lite::io::AsyncAsSync;
+ /// use std::task::Context;
+ /// use waker_fn::waker_fn;
+ ///
+ /// let reader: Vec<u8> = b"hello".to_vec();
+ /// let waker = waker_fn(|| {});
+ /// let mut context = Context::from_waker(&waker);
+ ///
+ /// let mut async_reader = AsyncAsSync::new(&mut context, reader);
+ /// async_reader.close().unwrap();
+ /// ```
+ #[inline]
+ pub fn close(&mut self) -> Result<()>
+ where
+ T: AsyncWrite + Unpin,
+ {
+ self.poll_with(|io, cx| io.poll_close(cx))
+ }
+
+ /// Poll this `AsyncAsSync` for some function.
+ ///
+ /// # Examples
+ ///
+ /// ```
+ /// use futures_lite::io::{AsyncAsSync, AsyncRead};
+ /// use std::task::Context;
+ /// use waker_fn::waker_fn;
+ ///
+ /// let reader: &[u8] = b"hello";
+ /// let waker = waker_fn(|| {});
+ /// let mut context = Context::from_waker(&waker);
+ ///
+ /// let mut async_reader = AsyncAsSync::new(&mut context, reader);
+ /// let r = async_reader.poll_with(|io, cx| io.poll_read(cx, &mut [0; 1024]));
+ /// assert_eq!(r.unwrap(), 5);
+ /// ```
+ #[inline]
+ pub fn poll_with<R>(
+ &mut self,
+ f: impl FnOnce(Pin<&mut T>, &mut Context<'_>) -> Poll<Result<R>>,
+ ) -> Result<R>
+ where
+ T: Unpin,
+ {
+ match f(Pin::new(&mut self.inner), self.context) {
+ Poll::Ready(res) => res,
+ Poll::Pending => Err(ErrorKind::WouldBlock.into()),
+ }
+ }
+}
+
+impl<T: AsyncRead + Unpin> std::io::Read for AsyncAsSync<'_, '_, T> {
+ #[inline]
+ fn read(&mut self, buf: &mut [u8]) -> Result<usize> {
+ self.poll_with(|io, cx| io.poll_read(cx, buf))
+ }
+
+ #[inline]
+ fn read_vectored(&mut self, bufs: &mut [IoSliceMut<'_>]) -> Result<usize> {
+ self.poll_with(|io, cx| io.poll_read_vectored(cx, bufs))
+ }
+}
+
+impl<T: AsyncWrite + Unpin> std::io::Write for AsyncAsSync<'_, '_, T> {
+ #[inline]
+ fn write(&mut self, buf: &[u8]) -> Result<usize> {
+ self.poll_with(|io, cx| io.poll_write(cx, buf))
+ }
+
+ #[inline]
+ fn write_vectored(&mut self, bufs: &[IoSlice<'_>]) -> Result<usize> {
+ self.poll_with(|io, cx| io.poll_write_vectored(cx, bufs))
+ }
+
+ #[inline]
+ fn flush(&mut self) -> Result<()> {
+ self.poll_with(|io, cx| io.poll_flush(cx))
+ }
+}
+
+impl<T: AsyncSeek + Unpin> std::io::Seek for AsyncAsSync<'_, '_, T> {
+ #[inline]
+ fn seek(&mut self, pos: SeekFrom) -> Result<u64> {
+ self.poll_with(|io, cx| io.poll_seek(cx, pos))
+ }
+}
+
+impl<T> AsRef<T> for AsyncAsSync<'_, '_, T> {
+ #[inline]
+ fn as_ref(&self) -> &T {
+ &self.inner
+ }
+}
+
+impl<T> AsMut<T> for AsyncAsSync<'_, '_, T> {
+ #[inline]
+ fn as_mut(&mut self) -> &mut T {
+ &mut self.inner
+ }
+}
+
+impl<T> Borrow<T> for AsyncAsSync<'_, '_, T> {
+ #[inline]
+ fn borrow(&self) -> &T {
+ &self.inner
+ }
+}
+
+impl<T> BorrowMut<T> for AsyncAsSync<'_, '_, T> {
+ #[inline]
+ fn borrow_mut(&mut self) -> &mut T {
+ &mut self.inner
+ }
+}
+
+/// Blocks on all async I/O operations and implements [`std::io`] traits.
+///
+/// Sometimes async I/O needs to be used in a blocking manner. If calling [`future::block_on()`]
+/// manually all the time becomes too tedious, use this type for more convenient blocking on async
+/// I/O operations.
+///
+/// This type implements traits [`Read`][`std::io::Read`], [`Write`][`std::io::Write`], or
+/// [`Seek`][`std::io::Seek`] if the inner type implements [`AsyncRead`], [`AsyncWrite`], or
+/// [`AsyncSeek`], respectively.
+///
+/// If writing data through the [`Write`][`std::io::Write`] trait, make sure to flush before
+/// dropping the [`BlockOn`] handle or some buffered data might get lost.
+///
+/// # Examples
+///
+/// ```
+/// use futures_lite::io::BlockOn;
+/// use futures_lite::pin;
+/// use std::io::Read;
+///
+/// let reader: &[u8] = b"hello";
+/// pin!(reader);
+///
+/// let mut blocking_reader = BlockOn::new(reader);
+/// let mut contents = String::new();
+///
+/// // This line blocks - note that there is no await:
+/// blocking_reader.read_to_string(&mut contents)?;
+/// # std::io::Result::Ok(())
+/// ```
+#[derive(Debug)]
+pub struct BlockOn<T>(T);
+
+impl<T> BlockOn<T> {
+ /// Wraps an async I/O handle into a blocking interface.
+ ///
+ /// # Examples
+ ///
+ /// ```
+ /// use futures_lite::io::BlockOn;
+ /// use futures_lite::pin;
+ ///
+ /// let reader: &[u8] = b"hello";
+ /// pin!(reader);
+ ///
+ /// let blocking_reader = BlockOn::new(reader);
+ /// ```
+ pub fn new(io: T) -> BlockOn<T> {
+ BlockOn(io)
+ }
+
+ /// Gets a reference to the async I/O handle.
+ ///
+ /// # Examples
+ ///
+ /// ```
+ /// use futures_lite::io::BlockOn;
+ /// use futures_lite::pin;
+ ///
+ /// let reader: &[u8] = b"hello";
+ /// pin!(reader);
+ ///
+ /// let blocking_reader = BlockOn::new(reader);
+ /// let r = blocking_reader.get_ref();
+ /// ```
+ pub fn get_ref(&self) -> &T {
+ &self.0
+ }
+
+ /// Gets a mutable reference to the async I/O handle.
+ ///
+ /// # Examples
+ ///
+ /// ```
+ /// use futures_lite::io::BlockOn;
+ /// use futures_lite::pin;
+ ///
+ /// let reader: &[u8] = b"hello";
+ /// pin!(reader);
+ ///
+ /// let mut blocking_reader = BlockOn::new(reader);
+ /// let r = blocking_reader.get_mut();
+ /// ```
+ pub fn get_mut(&mut self) -> &mut T {
+ &mut self.0
+ }
+
+ /// Extracts the inner async I/O handle.
+ ///
+ /// # Examples
+ ///
+ /// ```
+ /// use futures_lite::io::BlockOn;
+ /// use futures_lite::pin;
+ ///
+ /// let reader: &[u8] = b"hello";
+ /// pin!(reader);
+ ///
+ /// let blocking_reader = BlockOn::new(reader);
+ /// let inner = blocking_reader.into_inner();
+ /// ```
+ pub fn into_inner(self) -> T {
+ self.0
+ }
+}
+
+impl<T: AsyncRead + Unpin> std::io::Read for BlockOn<T> {
+ fn read(&mut self, buf: &mut [u8]) -> Result<usize> {
+ future::block_on(self.0.read(buf))
+ }
+}
+
+impl<T: AsyncBufRead + Unpin> std::io::BufRead for BlockOn<T> {
+ fn fill_buf(&mut self) -> Result<&[u8]> {
+ future::block_on(self.0.fill_buf())
+ }
+
+ fn consume(&mut self, amt: usize) {
+ Pin::new(&mut self.0).consume(amt)
+ }
+}
+
+impl<T: AsyncWrite + Unpin> std::io::Write for BlockOn<T> {
+ fn write(&mut self, buf: &[u8]) -> Result<usize> {
+ future::block_on(self.0.write(buf))
+ }
+
+ fn flush(&mut self) -> Result<()> {
+ future::block_on(self.0.flush())
+ }
+}
+
+impl<T: AsyncSeek + Unpin> std::io::Seek for BlockOn<T> {
+ fn seek(&mut self, pos: SeekFrom) -> Result<u64> {
+ future::block_on(self.0.seek(pos))
+ }
+}
+
+pin_project! {
+ /// Adds buffering to a reader.
+ ///
+ /// It can be excessively inefficient to work directly with an [`AsyncRead`] instance. A
+ /// [`BufReader`] performs large, infrequent reads on the underlying [`AsyncRead`] and
+ /// maintains an in-memory buffer of the incoming byte stream.
+ ///
+ /// [`BufReader`] can improve the speed of programs that make *small* and *repeated* reads to
+ /// the same file or networking socket. It does not help when reading very large amounts at
+ /// once, or reading just once or a few times. It also provides no advantage when reading from
+ /// a source that is already in memory, like a `Vec<u8>`.
+ ///
+ /// When a [`BufReader`] is dropped, the contents of its buffer are discarded. Creating
+ /// multiple instances of [`BufReader`] on the same reader can cause data loss.
+ ///
+ /// # Examples
+ ///
+ /// ```
+ /// use futures_lite::io::{AsyncBufReadExt, BufReader};
+ ///
+ /// # spin_on::spin_on(async {
+ /// let input: &[u8] = b"hello";
+ /// let mut reader = BufReader::new(input);
+ ///
+ /// let mut line = String::new();
+ /// reader.read_line(&mut line).await?;
+ /// # std::io::Result::Ok(()) });
+ /// ```
+ pub struct BufReader<R> {
+ #[pin]
+ inner: R,
+ buf: Box<[u8]>,
+ pos: usize,
+ cap: usize,
+ }
+}
+
+impl<R: AsyncRead> BufReader<R> {
+ /// Creates a buffered reader with the default buffer capacity.
+ ///
+ /// The default capacity is currently 8 KB, but that may change in the future.
+ ///
+ /// # Examples
+ ///
+ /// ```
+ /// use futures_lite::io::BufReader;
+ ///
+ /// let input: &[u8] = b"hello";
+ /// let reader = BufReader::new(input);
+ /// ```
+ pub fn new(inner: R) -> BufReader<R> {
+ BufReader::with_capacity(DEFAULT_BUF_SIZE, inner)
+ }
+
+ /// Creates a buffered reader with the specified capacity.
+ ///
+ /// # Examples
+ ///
+ /// ```
+ /// use futures_lite::io::BufReader;
+ ///
+ /// let input: &[u8] = b"hello";
+ /// let reader = BufReader::with_capacity(1024, input);
+ /// ```
+ pub fn with_capacity(capacity: usize, inner: R) -> BufReader<R> {
+ BufReader {
+ inner,
+ buf: vec![0; capacity].into_boxed_slice(),
+ pos: 0,
+ cap: 0,
+ }
+ }
+}
+
+impl<R> BufReader<R> {
+ /// Gets a reference to the underlying reader.
+ ///
+ /// It is not advisable to directly read from the underlying reader.
+ ///
+ /// # Examples
+ ///
+ /// ```
+ /// use futures_lite::io::BufReader;
+ ///
+ /// let input: &[u8] = b"hello";
+ /// let reader = BufReader::new(input);
+ ///
+ /// let r = reader.get_ref();
+ /// ```
+ pub fn get_ref(&self) -> &R {
+ &self.inner
+ }
+
+ /// Gets a mutable reference to the underlying reader.
+ ///
+ /// It is not advisable to directly read from the underlying reader.
+ ///
+ /// # Examples
+ ///
+ /// ```
+ /// use futures_lite::io::BufReader;
+ ///
+ /// let input: &[u8] = b"hello";
+ /// let mut reader = BufReader::new(input);
+ ///
+ /// let r = reader.get_mut();
+ /// ```
+ pub fn get_mut(&mut self) -> &mut R {
+ &mut self.inner
+ }
+
+ /// Gets a pinned mutable reference to the underlying reader.
+ ///
+ /// It is not advisable to directly read from the underlying reader.
+ fn get_pin_mut(self: Pin<&mut Self>) -> Pin<&mut R> {
+ self.project().inner
+ }
+
+ /// Returns a reference to the internal buffer.
+ ///
+ /// This method will not attempt to fill the buffer if it is empty.
+ ///
+ /// # Examples
+ ///
+ /// ```
+ /// use futures_lite::io::BufReader;
+ ///
+ /// let input: &[u8] = b"hello";
+ /// let reader = BufReader::new(input);
+ ///
+ /// // The internal buffer is empty until the first read request.
+ /// assert_eq!(reader.buffer(), &[]);
+ /// ```
+ pub fn buffer(&self) -> &[u8] {
+ &self.buf[self.pos..self.cap]
+ }
+
+ /// Unwraps the buffered reader, returning the underlying reader.
+ ///
+ /// Note that any leftover data in the internal buffer will be lost.
+ ///
+ /// # Examples
+ ///
+ /// ```
+ /// use futures_lite::io::BufReader;
+ ///
+ /// let input: &[u8] = b"hello";
+ /// let reader = BufReader::new(input);
+ ///
+ /// assert_eq!(reader.into_inner(), input);
+ /// ```
+ pub fn into_inner(self) -> R {
+ self.inner
+ }
+
+ /// Invalidates all data in the internal buffer.
+ #[inline]
+ fn discard_buffer(self: Pin<&mut Self>) {
+ let this = self.project();
+ *this.pos = 0;
+ *this.cap = 0;
+ }
+}
+
+impl<R: AsyncRead> AsyncRead for BufReader<R> {
+ fn poll_read(
+ mut self: Pin<&mut Self>,
+ cx: &mut Context<'_>,
+ buf: &mut [u8],
+ ) -> Poll<Result<usize>> {
+ // If we don't have any buffered data and we're doing a massive read
+ // (larger than our internal buffer), bypass our internal buffer
+ // entirely.
+ if self.pos == self.cap && buf.len() >= self.buf.len() {
+ let res = ready!(self.as_mut().get_pin_mut().poll_read(cx, buf));
+ self.discard_buffer();
+ return Poll::Ready(res);
+ }
+ let mut rem = ready!(self.as_mut().poll_fill_buf(cx))?;
+ let nread = std::io::Read::read(&mut rem, buf)?;
+ self.consume(nread);
+ Poll::Ready(Ok(nread))
+ }
+
+ fn poll_read_vectored(
+ mut self: Pin<&mut Self>,
+ cx: &mut Context<'_>,
+ bufs: &mut [IoSliceMut<'_>],
+ ) -> Poll<Result<usize>> {
+ let total_len = bufs.iter().map(|b| b.len()).sum::<usize>();
+ if self.pos == self.cap && total_len >= self.buf.len() {
+ let res = ready!(self.as_mut().get_pin_mut().poll_read_vectored(cx, bufs));
+ self.discard_buffer();
+ return Poll::Ready(res);
+ }
+ let mut rem = ready!(self.as_mut().poll_fill_buf(cx))?;
+ let nread = std::io::Read::read_vectored(&mut rem, bufs)?;
+ self.consume(nread);
+ Poll::Ready(Ok(nread))
+ }
+}
+
+impl<R: AsyncRead> AsyncBufRead for BufReader<R> {
+ fn poll_fill_buf<'a>(self: Pin<&'a mut Self>, cx: &mut Context<'_>) -> Poll<Result<&'a [u8]>> {
+ let mut this = self.project();
+
+ // If we've reached the end of our internal buffer then we need to fetch
+ // some more data from the underlying reader.
+ // Branch using `>=` instead of the more correct `==`
+ // to tell the compiler that the pos..cap slice is always valid.
+ if *this.pos >= *this.cap {
+ debug_assert!(*this.pos == *this.cap);
+ *this.cap = ready!(this.inner.as_mut().poll_read(cx, this.buf))?;
+ *this.pos = 0;
+ }
+ Poll::Ready(Ok(&this.buf[*this.pos..*this.cap]))
+ }
+
+ fn consume(self: Pin<&mut Self>, amt: usize) {
+ let this = self.project();
+ *this.pos = cmp::min(*this.pos + amt, *this.cap);
+ }
+}
+
+impl<R: fmt::Debug> fmt::Debug for BufReader<R> {
+ fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+ f.debug_struct("BufReader")
+ .field("reader", &self.inner)
+ .field(
+ "buffer",
+ &format_args!("{}/{}", self.cap - self.pos, self.buf.len()),
+ )
+ .finish()
+ }
+}
+
+impl<R: AsyncSeek> AsyncSeek for BufReader<R> {
+ /// Seeks to an offset, in bytes, in the underlying reader.
+ ///
+ /// The position used for seeking with [`SeekFrom::Current`] is the position the underlying
+ /// reader would be at if the [`BufReader`] had no internal buffer.
+ ///
+ /// Seeking always discards the internal buffer, even if the seek position would otherwise fall
+ /// within it. This guarantees that calling [`into_inner()`][`BufReader::into_inner()`]
+ /// immediately after a seek yields the underlying reader at the same position.
+ ///
+ /// See [`AsyncSeek`] for more details.
+ ///
+ /// Note: In the edge case where you're seeking with `SeekFrom::Current(n)` where `n` minus the
+ /// internal buffer length overflows an `i64`, two seeks will be performed instead of one. If
+ /// the second seek returns `Err`, the underlying reader will be left at the same position it
+ /// would have if you called [`seek()`][`AsyncSeekExt::seek()`] with `SeekFrom::Current(0)`.
+ fn poll_seek(
+ mut self: Pin<&mut Self>,
+ cx: &mut Context<'_>,
+ pos: SeekFrom,
+ ) -> Poll<Result<u64>> {
+ let result: u64;
+ if let SeekFrom::Current(n) = pos {
+ let remainder = (self.cap - self.pos) as i64;
+ // it should be safe to assume that remainder fits within an i64 as the alternative
+ // means we managed to allocate 8 exbibytes and that's absurd.
+ // But it's not out of the realm of possibility for some weird underlying reader to
+ // support seeking by i64::min_value() so we need to handle underflow when subtracting
+ // remainder.
+ if let Some(offset) = n.checked_sub(remainder) {
+ result = ready!(self
+ .as_mut()
+ .get_pin_mut()
+ .poll_seek(cx, SeekFrom::Current(offset)))?;
+ } else {
+ // seek backwards by our remainder, and then by the offset
+ ready!(self
+ .as_mut()
+ .get_pin_mut()
+ .poll_seek(cx, SeekFrom::Current(-remainder)))?;
+ self.as_mut().discard_buffer();
+ result = ready!(self
+ .as_mut()
+ .get_pin_mut()
+ .poll_seek(cx, SeekFrom::Current(n)))?;
+ }
+ } else {
+ // Seeking with Start/End doesn't care about our buffer length.
+ result = ready!(self.as_mut().get_pin_mut().poll_seek(cx, pos))?;
+ }
+ self.discard_buffer();
+ Poll::Ready(Ok(result))
+ }
+}
+
+impl<R: AsyncWrite> AsyncWrite for BufReader<R> {
+ fn poll_write(
+ mut self: Pin<&mut Self>,
+ cx: &mut Context<'_>,
+ buf: &[u8],
+ ) -> Poll<Result<usize>> {
+ self.as_mut().get_pin_mut().poll_write(cx, buf)
+ }
+
+ fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<()>> {
+ self.as_mut().get_pin_mut().poll_flush(cx)
+ }
+
+ fn poll_close(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<()>> {
+ self.as_mut().get_pin_mut().poll_close(cx)
+ }
+}
+
+pin_project! {
+ /// Adds buffering to a writer.
+ ///
+ /// It can be excessively inefficient to work directly with something that implements
+ /// [`AsyncWrite`]. For example, every call to [`write()`][`AsyncWriteExt::write()`] on a TCP
+ /// stream results in a system call. A [`BufWriter`] keeps an in-memory buffer of data and
+ /// writes it to the underlying writer in large, infrequent batches.
+ ///
+ /// [`BufWriter`] can improve the speed of programs that make *small* and *repeated* writes to
+ /// the same file or networking socket. It does not help when writing very large amounts at
+ /// once, or writing just once or a few times. It also provides no advantage when writing to a
+ /// destination that is in memory, like a `Vec<u8>`.
+ ///
+ /// Unlike [`std::io::BufWriter`], this type does not write out the contents of its buffer when
+ /// it is dropped. Therefore, it is important that users explicitly flush the buffer before
+ /// dropping the [`BufWriter`].
+ ///
+ /// # Examples
+ ///
+ /// ```
+ /// use futures_lite::io::{AsyncWriteExt, BufWriter};
+ ///
+ /// # spin_on::spin_on(async {
+ /// let mut output = Vec::new();
+ /// let mut writer = BufWriter::new(&mut output);
+ ///
+ /// writer.write_all(b"hello").await?;
+ /// writer.flush().await?;
+ /// # std::io::Result::Ok(()) });
+ /// ```
+ pub struct BufWriter<W> {
+ #[pin]
+ inner: W,
+ buf: Vec<u8>,
+ written: usize,
+ }
+}
+
+impl<W: AsyncWrite> BufWriter<W> {
+ /// Creates a buffered writer with the default buffer capacity.
+ ///
+ /// The default capacity is currently 8 KB, but that may change in the future.
+ ///
+ /// # Examples
+ ///
+ /// ```
+ /// use futures_lite::io::BufWriter;
+ ///
+ /// let mut output = Vec::new();
+ /// let writer = BufWriter::new(&mut output);
+ /// ```
+ pub fn new(inner: W) -> BufWriter<W> {
+ BufWriter::with_capacity(DEFAULT_BUF_SIZE, inner)
+ }
+
+ /// Creates a buffered writer with the specified buffer capacity.
+ ///
+ /// # Examples
+ ///
+ /// ```
+ /// use futures_lite::io::BufWriter;
+ ///
+ /// let mut output = Vec::new();
+ /// let writer = BufWriter::with_capacity(100, &mut output);
+ /// ```
+ pub fn with_capacity(capacity: usize, inner: W) -> BufWriter<W> {
+ BufWriter {
+ inner,
+ buf: Vec::with_capacity(capacity),
+ written: 0,
+ }
+ }
+
+ /// Gets a reference to the underlying writer.
+ ///
+ /// # Examples
+ ///
+ /// ```
+ /// use futures_lite::io::BufWriter;
+ ///
+ /// let mut output = Vec::new();
+ /// let writer = BufWriter::new(&mut output);
+ ///
+ /// let r = writer.get_ref();
+ /// ```
+ pub fn get_ref(&self) -> &W {
+ &self.inner
+ }
+
+ /// Gets a mutable reference to the underlying writer.
+ ///
+ /// It is not advisable to directly write to the underlying writer.
+ ///
+ /// # Examples
+ ///
+ /// ```
+ /// use futures_lite::io::BufWriter;
+ ///
+ /// let mut output = Vec::new();
+ /// let mut writer = BufWriter::new(&mut output);
+ ///
+ /// let r = writer.get_mut();
+ /// ```
+ pub fn get_mut(&mut self) -> &mut W {
+ &mut self.inner
+ }
+
+ /// Gets a pinned mutable reference to the underlying writer.
+ ///
+ /// It is not not advisable to directly write to the underlying writer.
+ fn get_pin_mut(self: Pin<&mut Self>) -> Pin<&mut W> {
+ self.project().inner
+ }
+
+ /// Unwraps the buffered writer, returning the underlying writer.
+ ///
+ /// Note that any leftover data in the internal buffer will be lost. If you don't want to lose
+ /// that data, flush the buffered writer before unwrapping it.
+ ///
+ /// # Examples
+ ///
+ /// ```
+ /// use futures_lite::io::{AsyncWriteExt, BufWriter};
+ ///
+ /// # spin_on::spin_on(async {
+ /// let mut output = vec![1, 2, 3];
+ /// let mut writer = BufWriter::new(&mut output);
+ ///
+ /// writer.write_all(&[4]).await?;
+ /// writer.flush().await?;
+ /// assert_eq!(writer.into_inner(), &[1, 2, 3, 4]);
+ /// # std::io::Result::Ok(()) });
+ /// ```
+ pub fn into_inner(self) -> W {
+ self.inner
+ }
+
+ /// Returns a reference to the internal buffer.
+ ///
+ /// # Examples
+ ///
+ /// ```
+ /// use futures_lite::io::BufWriter;
+ ///
+ /// let mut output = Vec::new();
+ /// let writer = BufWriter::new(&mut output);
+ ///
+ /// // The internal buffer is empty until the first write request.
+ /// assert_eq!(writer.buffer(), &[]);
+ /// ```
+ pub fn buffer(&self) -> &[u8] {
+ &self.buf
+ }
+
+ /// Flush the buffer.
+ fn poll_flush_buf(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<()>> {
+ let mut this = self.project();
+ let len = this.buf.len();
+ let mut ret = Ok(());
+
+ while *this.written < len {
+ match this
+ .inner
+ .as_mut()
+ .poll_write(cx, &this.buf[*this.written..])
+ {
+ Poll::Ready(Ok(0)) => {
+ ret = Err(Error::new(
+ ErrorKind::WriteZero,
+ "Failed to write buffered data",
+ ));
+ break;
+ }
+ Poll::Ready(Ok(n)) => *this.written += n,
+ Poll::Ready(Err(ref e)) if e.kind() == ErrorKind::Interrupted => {}
+ Poll::Ready(Err(e)) => {
+ ret = Err(e);
+ break;
+ }
+ Poll::Pending => return Poll::Pending,
+ }
+ }
+
+ if *this.written > 0 {
+ this.buf.drain(..*this.written);
+ }
+ *this.written = 0;
+
+ Poll::Ready(ret)
+ }
+}
+
+impl<W: fmt::Debug> fmt::Debug for BufWriter<W> {
+ fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+ f.debug_struct("BufWriter")
+ .field("writer", &self.inner)
+ .field("buf", &self.buf)
+ .finish()
+ }
+}
+
+impl<W: AsyncWrite> AsyncWrite for BufWriter<W> {
+ fn poll_write(
+ mut self: Pin<&mut Self>,
+ cx: &mut Context<'_>,
+ buf: &[u8],
+ ) -> Poll<Result<usize>> {
+ if self.buf.len() + buf.len() > self.buf.capacity() {
+ ready!(self.as_mut().poll_flush_buf(cx))?;
+ }
+ if buf.len() >= self.buf.capacity() {
+ self.get_pin_mut().poll_write(cx, buf)
+ } else {
+ Pin::new(&mut *self.project().buf).poll_write(cx, buf)
+ }
+ }
+
+ fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<()>> {
+ ready!(self.as_mut().poll_flush_buf(cx))?;
+ self.get_pin_mut().poll_flush(cx)
+ }
+
+ fn poll_close(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<()>> {
+ ready!(self.as_mut().poll_flush_buf(cx))?;
+ self.get_pin_mut().poll_close(cx)
+ }
+}
+
+impl<W: AsyncWrite + AsyncSeek> AsyncSeek for BufWriter<W> {
+ /// Seek to the offset, in bytes, in the underlying writer.
+ ///
+ /// Seeking always writes out the internal buffer before seeking.
+ fn poll_seek(
+ mut self: Pin<&mut Self>,
+ cx: &mut Context<'_>,
+ pos: SeekFrom,
+ ) -> Poll<Result<u64>> {
+ ready!(self.as_mut().poll_flush_buf(cx))?;
+ self.get_pin_mut().poll_seek(cx, pos)
+ }
+}
+
+/// Gives an in-memory buffer a cursor for reading and writing.
+///
+/// # Examples
+///
+/// ```
+/// use futures_lite::io::{AsyncReadExt, AsyncSeekExt, AsyncWriteExt, Cursor, SeekFrom};
+///
+/// # spin_on::spin_on(async {
+/// let mut bytes = b"hello".to_vec();
+/// let mut cursor = Cursor::new(&mut bytes);
+///
+/// // Overwrite 'h' with 'H'.
+/// cursor.write_all(b"H").await?;
+///
+/// // Move the cursor one byte forward.
+/// cursor.seek(SeekFrom::Current(1)).await?;
+///
+/// // Read a byte.
+/// let mut byte = [0];
+/// cursor.read_exact(&mut byte).await?;
+/// assert_eq!(&byte, b"l");
+///
+/// // Check the final buffer.
+/// assert_eq!(bytes, b"Hello");
+/// # std::io::Result::Ok(()) });
+/// ```
+#[derive(Clone, Debug, Default)]
+pub struct Cursor<T> {
+ inner: std::io::Cursor<T>,
+}
+
+impl<T> Cursor<T> {
+ /// Creates a cursor for an in-memory buffer.
+ ///
+ /// Cursor's initial position is 0 even if the underlying buffer is not empty. Writing using
+ /// [`Cursor`] will overwrite the existing contents unless the cursor is moved to the end of
+ /// the buffer using [`set_position()`][Cursor::set_position()`] or
+ /// [`seek()`][`AsyncSeekExt::seek()`].
+ ///
+ /// # Examples
+ ///
+ /// ```
+ /// use futures_lite::io::Cursor;
+ ///
+ /// let cursor = Cursor::new(Vec::<u8>::new());
+ /// ```
+ pub fn new(inner: T) -> Cursor<T> {
+ Cursor {
+ inner: std::io::Cursor::new(inner),
+ }
+ }
+
+ /// Gets a reference to the underlying buffer.
+ ///
+ /// # Examples
+ ///
+ /// ```
+ /// use futures_lite::io::Cursor;
+ ///
+ /// let cursor = Cursor::new(Vec::<u8>::new());
+ /// let r = cursor.get_ref();
+ /// ```
+ pub fn get_ref(&self) -> &T {
+ self.inner.get_ref()
+ }
+
+ /// Gets a mutable reference to the underlying buffer.
+ ///
+ /// # Examples
+ ///
+ /// ```
+ /// use futures_lite::io::Cursor;
+ ///
+ /// let mut cursor = Cursor::new(Vec::<u8>::new());
+ /// let r = cursor.get_mut();
+ /// ```
+ pub fn get_mut(&mut self) -> &mut T {
+ self.inner.get_mut()
+ }
+
+ /// Unwraps the cursor, returning the underlying buffer.
+ ///
+ /// # Examples
+ ///
+ /// ```
+ /// use futures_lite::io::Cursor;
+ ///
+ /// let cursor = Cursor::new(vec![1, 2, 3]);
+ /// assert_eq!(cursor.into_inner(), [1, 2, 3]);
+ /// ```
+ pub fn into_inner(self) -> T {
+ self.inner.into_inner()
+ }
+
+ /// Returns the current position of this cursor.
+ ///
+ /// # Examples
+ ///
+ /// ```
+ /// use futures_lite::io::{AsyncSeekExt, Cursor, SeekFrom};
+ ///
+ /// # spin_on::spin_on(async {
+ /// let mut cursor = Cursor::new(b"hello");
+ /// assert_eq!(cursor.position(), 0);
+ ///
+ /// cursor.seek(SeekFrom::Start(2)).await?;
+ /// assert_eq!(cursor.position(), 2);
+ /// # std::io::Result::Ok(()) });
+ /// ```
+ pub fn position(&self) -> u64 {
+ self.inner.position()
+ }
+
+ /// Sets the position of this cursor.
+ ///
+ /// # Examples
+ ///
+ /// ```
+ /// use futures_lite::io::Cursor;
+ ///
+ /// let mut cursor = Cursor::new(b"hello");
+ /// assert_eq!(cursor.position(), 0);
+ ///
+ /// cursor.set_position(2);
+ /// assert_eq!(cursor.position(), 2);
+ /// ```
+ pub fn set_position(&mut self, pos: u64) {
+ self.inner.set_position(pos)
+ }
+}
+
+impl<T> AsyncSeek for Cursor<T>
+where
+ T: AsRef<[u8]> + Unpin,
+{
+ fn poll_seek(
+ mut self: Pin<&mut Self>,
+ _: &mut Context<'_>,
+ pos: SeekFrom,
+ ) -> Poll<Result<u64>> {
+ Poll::Ready(std::io::Seek::seek(&mut self.inner, pos))
+ }
+}
+
+impl<T> AsyncRead for Cursor<T>
+where
+ T: AsRef<[u8]> + Unpin,
+{
+ fn poll_read(
+ mut self: Pin<&mut Self>,
+ _cx: &mut Context<'_>,
+ buf: &mut [u8],
+ ) -> Poll<Result<usize>> {
+ Poll::Ready(std::io::Read::read(&mut self.inner, buf))
+ }
+
+ fn poll_read_vectored(
+ mut self: Pin<&mut Self>,
+ _: &mut Context<'_>,
+ bufs: &mut [IoSliceMut<'_>],
+ ) -> Poll<Result<usize>> {
+ Poll::Ready(std::io::Read::read_vectored(&mut self.inner, bufs))
+ }
+}
+
+impl<T> AsyncBufRead for Cursor<T>
+where
+ T: AsRef<[u8]> + Unpin,
+{
+ fn poll_fill_buf(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<Result<&[u8]>> {
+ Poll::Ready(std::io::BufRead::fill_buf(&mut self.get_mut().inner))
+ }
+
+ fn consume(mut self: Pin<&mut Self>, amt: usize) {
+ std::io::BufRead::consume(&mut self.inner, amt)
+ }
+}
+
+impl AsyncWrite for Cursor<&mut [u8]> {
+ fn poll_write(
+ mut self: Pin<&mut Self>,
+ _: &mut Context<'_>,
+ buf: &[u8],
+ ) -> Poll<Result<usize>> {
+ Poll::Ready(std::io::Write::write(&mut self.inner, buf))
+ }
+
+ fn poll_write_vectored(
+ mut self: Pin<&mut Self>,
+ _: &mut Context<'_>,
+ bufs: &[IoSlice<'_>],
+ ) -> Poll<Result<usize>> {
+ Poll::Ready(std::io::Write::write_vectored(&mut self.inner, bufs))
+ }
+
+ fn poll_flush(mut self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<Result<()>> {
+ Poll::Ready(std::io::Write::flush(&mut self.inner))
+ }
+
+ fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<()>> {
+ self.poll_flush(cx)
+ }
+}
+
+impl AsyncWrite for Cursor<&mut Vec<u8>> {
+ fn poll_write(
+ mut self: Pin<&mut Self>,
+ _: &mut Context<'_>,
+ buf: &[u8],
+ ) -> Poll<Result<usize>> {
+ Poll::Ready(std::io::Write::write(&mut self.inner, buf))
+ }
+
+ fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<()>> {
+ self.poll_flush(cx)
+ }
+
+ fn poll_flush(mut self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<Result<()>> {
+ Poll::Ready(std::io::Write::flush(&mut self.inner))
+ }
+}
+
+impl AsyncWrite for Cursor<Vec<u8>> {
+ fn poll_write(
+ mut self: Pin<&mut Self>,
+ _: &mut Context<'_>,
+ buf: &[u8],
+ ) -> Poll<Result<usize>> {
+ Poll::Ready(std::io::Write::write(&mut self.inner, buf))
+ }
+
+ fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<()>> {
+ self.poll_flush(cx)
+ }
+
+ fn poll_flush(mut self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<Result<()>> {
+ Poll::Ready(std::io::Write::flush(&mut self.inner))
+ }
+}
+
+/// Creates an empty reader.
+///
+/// # Examples
+///
+/// ```
+/// use futures_lite::io::{self, AsyncReadExt};
+///
+/// # spin_on::spin_on(async {
+/// let mut reader = io::empty();
+///
+/// let mut contents = Vec::new();
+/// reader.read_to_end(&mut contents).await?;
+/// assert!(contents.is_empty());
+/// # std::io::Result::Ok(()) });
+/// ```
+pub fn empty() -> Empty {
+ Empty { _private: () }
+}
+
+/// Reader for the [`empty()`] function.
+pub struct Empty {
+ _private: (),
+}
+
+impl fmt::Debug for Empty {
+ fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+ f.pad("Empty { .. }")
+ }
+}
+
+impl AsyncRead for Empty {
+ #[inline]
+ fn poll_read(self: Pin<&mut Self>, _: &mut Context<'_>, _: &mut [u8]) -> Poll<Result<usize>> {
+ Poll::Ready(Ok(0))
+ }
+}
+
+impl AsyncBufRead for Empty {
+ #[inline]
+ fn poll_fill_buf<'a>(self: Pin<&'a mut Self>, _: &mut Context<'_>) -> Poll<Result<&'a [u8]>> {
+ Poll::Ready(Ok(&[]))
+ }
+
+ #[inline]
+ fn consume(self: Pin<&mut Self>, _: usize) {}
+}
+
+/// Creates an infinite reader that reads the same byte repeatedly.
+///
+/// # Examples
+///
+/// ```
+/// use futures_lite::io::{self, AsyncReadExt};
+///
+/// # spin_on::spin_on(async {
+/// let mut reader = io::repeat(b'a');
+///
+/// let mut contents = vec![0; 5];
+/// reader.read_exact(&mut contents).await?;
+/// assert_eq!(contents, b"aaaaa");
+/// # std::io::Result::Ok(()) });
+/// ```
+pub fn repeat(byte: u8) -> Repeat {
+ Repeat { byte }
+}
+
+/// Reader for the [`repeat()`] function.
+#[derive(Debug)]
+pub struct Repeat {
+ byte: u8,
+}
+
+impl AsyncRead for Repeat {
+ #[inline]
+ fn poll_read(self: Pin<&mut Self>, _: &mut Context<'_>, buf: &mut [u8]) -> Poll<Result<usize>> {
+ for b in &mut *buf {
+ *b = self.byte;
+ }
+ Poll::Ready(Ok(buf.len()))
+ }
+}
+
+/// Creates a writer that consumes and drops all data.
+///
+/// # Examples
+///
+/// ```
+/// use futures_lite::io::{self, AsyncWriteExt};
+///
+/// # spin_on::spin_on(async {
+/// let mut writer = io::sink();
+/// writer.write_all(b"hello").await?;
+/// # std::io::Result::Ok(()) });
+/// ```
+pub fn sink() -> Sink {
+ Sink { _private: () }
+}
+
+/// Writer for the [`sink()`] function.
+#[derive(Debug)]
+pub struct Sink {
+ _private: (),
+}
+
+impl AsyncWrite for Sink {
+ #[inline]
+ fn poll_write(self: Pin<&mut Self>, _: &mut Context<'_>, buf: &[u8]) -> Poll<Result<usize>> {
+ Poll::Ready(Ok(buf.len()))
+ }
+
+ #[inline]
+ fn poll_flush(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<Result<()>> {
+ Poll::Ready(Ok(()))
+ }
+
+ #[inline]
+ fn poll_close(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<Result<()>> {
+ Poll::Ready(Ok(()))
+ }
+}
+
+/// Extension trait for [`AsyncBufRead`].
+pub trait AsyncBufReadExt: AsyncBufRead {
+ /// Returns the contents of the internal buffer, filling it with more data if empty.
+ ///
+ /// If the stream has reached EOF, an empty buffer will be returned.
+ ///
+ /// # Examples
+ ///
+ /// ```
+ /// use futures_lite::io::{AsyncBufReadExt, BufReader};
+ /// use std::pin::Pin;
+ ///
+ /// # spin_on::spin_on(async {
+ /// let input: &[u8] = b"hello world";
+ /// let mut reader = BufReader::with_capacity(5, input);
+ ///
+ /// assert_eq!(reader.fill_buf().await?, b"hello");
+ /// reader.consume(2);
+ /// assert_eq!(reader.fill_buf().await?, b"llo");
+ /// reader.consume(3);
+ /// assert_eq!(reader.fill_buf().await?, b" worl");
+ /// # std::io::Result::Ok(()) });
+ /// ```
+ fn fill_buf(&mut self) -> FillBuf<'_, Self>
+ where
+ Self: Unpin,
+ {
+ FillBuf { reader: Some(self) }
+ }
+
+ /// Consumes `amt` buffered bytes.
+ ///
+ /// This method does not perform any I/O, it simply consumes some amount of bytes from the
+ /// internal buffer.
+ ///
+ /// The `amt` must be <= the number of bytes in the buffer returned by
+ /// [`fill_buf()`][`AsyncBufReadExt::fill_buf()`].
+ ///
+ /// # Examples
+ ///
+ /// ```
+ /// use futures_lite::io::{AsyncBufReadExt, BufReader};
+ /// use std::pin::Pin;
+ ///
+ /// # spin_on::spin_on(async {
+ /// let input: &[u8] = b"hello";
+ /// let mut reader = BufReader::with_capacity(4, input);
+ ///
+ /// assert_eq!(reader.fill_buf().await?, b"hell");
+ /// reader.consume(2);
+ /// assert_eq!(reader.fill_buf().await?, b"ll");
+ /// # std::io::Result::Ok(()) });
+ /// ```
+ fn consume(&mut self, amt: usize)
+ where
+ Self: Unpin,
+ {
+ AsyncBufRead::consume(Pin::new(self), amt);
+ }
+
+ /// Reads all bytes and appends them into `buf` until the delimiter `byte` or EOF is found.
+ ///
+ /// This method will read bytes from the underlying stream until the delimiter or EOF is
+ /// found. All bytes up to and including the delimiter (if found) will be appended to `buf`.
+ ///
+ /// If successful, returns the total number of bytes read.
+ ///
+ /// # Examples
+ ///
+ /// ```
+ /// use futures_lite::io::{AsyncBufReadExt, BufReader};
+ ///
+ /// # spin_on::spin_on(async {
+ /// let input: &[u8] = b"hello";
+ /// let mut reader = BufReader::new(input);
+ ///
+ /// let mut buf = Vec::new();
+ /// let n = reader.read_until(b'\n', &mut buf).await?;
+ /// # std::io::Result::Ok(()) });
+ /// ```
+ fn read_until<'a>(&'a mut self, byte: u8, buf: &'a mut Vec<u8>) -> ReadUntilFuture<'a, Self>
+ where
+ Self: Unpin,
+ {
+ ReadUntilFuture {
+ reader: self,
+ byte,
+ buf,
+ read: 0,
+ }
+ }
+
+ /// Reads all bytes and appends them into `buf` until a newline (the 0xA byte) or EOF is found.
+ ///
+ /// This method will read bytes from the underlying stream until the newline delimiter (the
+ /// 0xA byte) or EOF is found. All bytes up to, and including, the newline delimiter (if found)
+ /// will be appended to `buf`.
+ ///
+ /// If successful, returns the total number of bytes read.
+ ///
+ /// # Examples
+ ///
+ /// ```
+ /// use futures_lite::io::{AsyncBufReadExt, BufReader};
+ ///
+ /// # spin_on::spin_on(async {
+ /// let input: &[u8] = b"hello";
+ /// let mut reader = BufReader::new(input);
+ ///
+ /// let mut line = String::new();
+ /// let n = reader.read_line(&mut line).await?;
+ /// # std::io::Result::Ok(()) });
+ /// ```
+ fn read_line<'a>(&'a mut self, buf: &'a mut String) -> ReadLineFuture<'a, Self>
+ where
+ Self: Unpin,
+ {
+ ReadLineFuture {
+ reader: self,
+ buf,
+ bytes: Vec::new(),
+ read: 0,
+ }
+ }
+
+ /// Returns a stream over the lines of this byte stream.
+ ///
+ /// The stream returned from this method yields items of type
+ /// [`io::Result`][`super::io::Result`]`<`[`String`]`>`.
+ /// Each string returned will *not* have a newline byte (the 0xA byte) or CRLF (0xD, 0xA bytes)
+ /// at the end.
+ ///
+ /// # Examples
+ ///
+ /// ```
+ /// use futures_lite::io::{AsyncBufReadExt, BufReader};
+ /// use futures_lite::stream::StreamExt;
+ ///
+ /// # spin_on::spin_on(async {
+ /// let input: &[u8] = b"hello\nworld\n";
+ /// let mut reader = BufReader::new(input);
+ /// let mut lines = reader.lines();
+ ///
+ /// while let Some(line) = lines.next().await {
+ /// println!("{}", line?);
+ /// }
+ /// # std::io::Result::Ok(()) });
+ /// ```
+ fn lines(self) -> Lines<Self>
+ where
+ Self: Sized,
+ {
+ Lines {
+ reader: self,
+ buf: String::new(),
+ bytes: Vec::new(),
+ read: 0,
+ }
+ }
+
+ /// Returns a stream over the contents of this reader split on the specified `byte`.
+ ///
+ /// The stream returned from this method yields items of type
+ /// [`io::Result`][`super::io::Result`]`<`[`Vec<u8>`][`Vec`]`>`.
+ /// Each vector returned will *not* have the delimiter byte at the end.
+ ///
+ /// # Examples
+ ///
+ /// ```
+ /// use futures_lite::io::{AsyncBufReadExt, Cursor};
+ /// use futures_lite::stream::StreamExt;
+ ///
+ /// # spin_on::spin_on(async {
+ /// let cursor = Cursor::new(b"lorem-ipsum-dolor");
+ /// let items: Vec<Vec<u8>> = cursor.split(b'-').try_collect().await?;
+ ///
+ /// assert_eq!(items[0], b"lorem");
+ /// assert_eq!(items[1], b"ipsum");
+ /// assert_eq!(items[2], b"dolor");
+ /// # std::io::Result::Ok(()) });
+ /// ```
+ fn split(self, byte: u8) -> Split<Self>
+ where
+ Self: Sized,
+ {
+ Split {
+ reader: self,
+ buf: Vec::new(),
+ delim: byte,
+ read: 0,
+ }
+ }
+}
+
+impl<R: AsyncBufRead + ?Sized> AsyncBufReadExt for R {}
+
+/// Future for the [`AsyncBufReadExt::fill_buf()`] method.
+#[derive(Debug)]
+#[must_use = "futures do nothing unless you `.await` or poll them"]
+pub struct FillBuf<'a, R: ?Sized> {
+ reader: Option<&'a mut R>,
+}
+
+impl<R: ?Sized> Unpin for FillBuf<'_, R> {}
+
+impl<'a, R> Future for FillBuf<'a, R>
+where
+ R: AsyncBufRead + Unpin + ?Sized,
+{
+ type Output = Result<&'a [u8]>;
+
+ fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+ let this = &mut *self;
+ let reader = this
+ .reader
+ .take()
+ .expect("polled `FillBuf` after completion");
+
+ match Pin::new(&mut *reader).poll_fill_buf(cx) {
+ Poll::Ready(Ok(_)) => match Pin::new(reader).poll_fill_buf(cx) {
+ Poll::Ready(Ok(slice)) => Poll::Ready(Ok(slice)),
+ poll => panic!("`poll_fill_buf()` was ready but now it isn't: {:?}", poll),
+ },
+ Poll::Ready(Err(err)) => Poll::Ready(Err(err)),
+ Poll::Pending => {
+ this.reader = Some(reader);
+ Poll::Pending
+ }
+ }
+ }
+}
+
+/// Future for the [`AsyncBufReadExt::read_until()`] method.
+#[derive(Debug)]
+#[must_use = "futures do nothing unless you `.await` or poll them"]
+pub struct ReadUntilFuture<'a, R: Unpin + ?Sized> {
+ reader: &'a mut R,
+ byte: u8,
+ buf: &'a mut Vec<u8>,
+ read: usize,
+}
+
+impl<R: Unpin + ?Sized> Unpin for ReadUntilFuture<'_, R> {}
+
+impl<R: AsyncBufRead + Unpin + ?Sized> Future for ReadUntilFuture<'_, R> {
+ type Output = Result<usize>;
+
+ fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+ let Self {
+ reader,
+ byte,
+ buf,
+ read,
+ } = &mut *self;
+ read_until_internal(Pin::new(reader), cx, *byte, buf, read)
+ }
+}
+
+fn read_until_internal<R: AsyncBufReadExt + ?Sized>(
+ mut reader: Pin<&mut R>,
+ cx: &mut Context<'_>,
+ byte: u8,
+ buf: &mut Vec<u8>,
+ read: &mut usize,
+) -> Poll<Result<usize>> {
+ loop {
+ let (done, used) = {
+ let available = ready!(reader.as_mut().poll_fill_buf(cx))?;
+
+ if let Some(i) = memchr(byte, available) {
+ buf.extend_from_slice(&available[..=i]);
+ (true, i + 1)
+ } else {
+ buf.extend_from_slice(available);
+ (false, available.len())
+ }
+ };
+
+ reader.as_mut().consume(used);
+ *read += used;
+
+ if done || used == 0 {
+ return Poll::Ready(Ok(mem::replace(read, 0)));
+ }
+ }
+}
+
+/// Future for the [`AsyncBufReadExt::read_line()`] method.
+#[derive(Debug)]
+#[must_use = "futures do nothing unless you `.await` or poll them"]
+pub struct ReadLineFuture<'a, R: Unpin + ?Sized> {
+ reader: &'a mut R,
+ buf: &'a mut String,
+ bytes: Vec<u8>,
+ read: usize,
+}
+
+impl<R: Unpin + ?Sized> Unpin for ReadLineFuture<'_, R> {}
+
+impl<R: AsyncBufRead + Unpin + ?Sized> Future for ReadLineFuture<'_, R> {
+ type Output = Result<usize>;
+
+ fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+ let Self {
+ reader,
+ buf,
+ bytes,
+ read,
+ } = &mut *self;
+ read_line_internal(Pin::new(reader), cx, buf, bytes, read)
+ }
+}
+
+pin_project! {
+ /// Stream for the [`AsyncBufReadExt::lines()`] method.
+ #[derive(Debug)]
+ #[must_use = "streams do nothing unless polled"]
+ pub struct Lines<R> {
+ #[pin]
+ reader: R,
+ buf: String,
+ bytes: Vec<u8>,
+ read: usize,
+ }
+}
+
+impl<R: AsyncBufRead> Stream for Lines<R> {
+ type Item = Result<String>;
+
+ fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+ let this = self.project();
+
+ let n = ready!(read_line_internal(
+ this.reader,
+ cx,
+ this.buf,
+ this.bytes,
+ this.read
+ ))?;
+ if n == 0 && this.buf.is_empty() {
+ return Poll::Ready(None);
+ }
+
+ if this.buf.ends_with('\n') {
+ this.buf.pop();
+ if this.buf.ends_with('\r') {
+ this.buf.pop();
+ }
+ }
+ Poll::Ready(Some(Ok(mem::take(this.buf))))
+ }
+}
+
+fn read_line_internal<R: AsyncBufRead + ?Sized>(
+ reader: Pin<&mut R>,
+ cx: &mut Context<'_>,
+ buf: &mut String,
+ bytes: &mut Vec<u8>,
+ read: &mut usize,
+) -> Poll<Result<usize>> {
+ let ret = ready!(read_until_internal(reader, cx, b'\n', bytes, read));
+
+ match String::from_utf8(mem::take(bytes)) {
+ Ok(s) => {
+ debug_assert!(buf.is_empty());
+ debug_assert_eq!(*read, 0);
+ *buf = s;
+ Poll::Ready(ret)
+ }
+ Err(_) => Poll::Ready(ret.and_then(|_| {
+ Err(Error::new(
+ ErrorKind::InvalidData,
+ "stream did not contain valid UTF-8",
+ ))
+ })),
+ }
+}
+
+pin_project! {
+ /// Stream for the [`AsyncBufReadExt::split()`] method.
+ #[derive(Debug)]
+ #[must_use = "streams do nothing unless polled"]
+ pub struct Split<R> {
+ #[pin]
+ reader: R,
+ buf: Vec<u8>,
+ read: usize,
+ delim: u8,
+ }
+}
+
+impl<R: AsyncBufRead> Stream for Split<R> {
+ type Item = Result<Vec<u8>>;
+
+ fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+ let this = self.project();
+
+ let n = ready!(read_until_internal(
+ this.reader,
+ cx,
+ *this.delim,
+ this.buf,
+ this.read
+ ))?;
+ if n == 0 && this.buf.is_empty() {
+ return Poll::Ready(None);
+ }
+
+ if this.buf[this.buf.len() - 1] == *this.delim {
+ this.buf.pop();
+ }
+ Poll::Ready(Some(Ok(mem::take(this.buf))))
+ }
+}
+
+/// Extension trait for [`AsyncRead`].
+pub trait AsyncReadExt: AsyncRead {
+ /// Reads some bytes from the byte stream.
+ ///
+ /// On success, returns the total number of bytes read.
+ ///
+ /// If the return value is `Ok(n)`, then it must be guaranteed that
+ /// `0 <= n <= buf.len()`. A nonzero `n` value indicates that the buffer has been
+ /// filled with `n` bytes of data. If `n` is `0`, then it can indicate one of two
+ /// scenarios:
+ ///
+ /// 1. This reader has reached its "end of file" and will likely no longer be able to
+ /// produce bytes. Note that this does not mean that the reader will always no
+ /// longer be able to produce bytes.
+ /// 2. The buffer specified was 0 bytes in length.
+ ///
+ /// # Examples
+ ///
+ /// ```
+ /// use futures_lite::io::{AsyncReadExt, BufReader};
+ ///
+ /// # spin_on::spin_on(async {
+ /// let input: &[u8] = b"hello";
+ /// let mut reader = BufReader::new(input);
+ ///
+ /// let mut buf = vec![0; 1024];
+ /// let n = reader.read(&mut buf).await?;
+ /// # std::io::Result::Ok(()) });
+ /// ```
+ fn read<'a>(&'a mut self, buf: &'a mut [u8]) -> ReadFuture<'a, Self>
+ where
+ Self: Unpin,
+ {
+ ReadFuture { reader: self, buf }
+ }
+
+ /// Like [`read()`][`AsyncReadExt::read()`], except it reads into a slice of buffers.
+ ///
+ /// Data is copied to fill each buffer in order, with the final buffer possibly being
+ /// only partially filled. This method must behave same as a single call to
+ /// [`read()`][`AsyncReadExt::read()`] with the buffers concatenated would.
+ fn read_vectored<'a>(
+ &'a mut self,
+ bufs: &'a mut [IoSliceMut<'a>],
+ ) -> ReadVectoredFuture<'a, Self>
+ where
+ Self: Unpin,
+ {
+ ReadVectoredFuture { reader: self, bufs }
+ }
+
+ /// Reads the entire contents and appends them to a [`Vec`].
+ ///
+ /// On success, returns the total number of bytes read.
+ ///
+ /// # Examples
+ ///
+ /// ```
+ /// use futures_lite::io::{AsyncReadExt, Cursor};
+ ///
+ /// # spin_on::spin_on(async {
+ /// let mut reader = Cursor::new(vec![1, 2, 3]);
+ /// let mut contents = Vec::new();
+ ///
+ /// let n = reader.read_to_end(&mut contents).await?;
+ /// assert_eq!(n, 3);
+ /// assert_eq!(contents, [1, 2, 3]);
+ /// # std::io::Result::Ok(()) });
+ /// ```
+ fn read_to_end<'a>(&'a mut self, buf: &'a mut Vec<u8>) -> ReadToEndFuture<'a, Self>
+ where
+ Self: Unpin,
+ {
+ let start_len = buf.len();
+ ReadToEndFuture {
+ reader: self,
+ buf,
+ start_len,
+ }
+ }
+
+ /// Reads the entire contents and appends them to a [`String`].
+ ///
+ /// On success, returns the total number of bytes read.
+ ///
+ /// # Examples
+ ///
+ /// ```
+ /// use futures_lite::io::{AsyncReadExt, Cursor};
+ ///
+ /// # spin_on::spin_on(async {
+ /// let mut reader = Cursor::new(&b"hello");
+ /// let mut contents = String::new();
+ ///
+ /// let n = reader.read_to_string(&mut contents).await?;
+ /// assert_eq!(n, 5);
+ /// assert_eq!(contents, "hello");
+ /// # std::io::Result::Ok(()) });
+ /// ```
+ fn read_to_string<'a>(&'a mut self, buf: &'a mut String) -> ReadToStringFuture<'a, Self>
+ where
+ Self: Unpin,
+ {
+ ReadToStringFuture {
+ reader: self,
+ buf,
+ bytes: Vec::new(),
+ start_len: 0,
+ }
+ }
+
+ /// Reads the exact number of bytes required to fill `buf`.
+ ///
+ /// # Examples
+ ///
+ /// ```
+ /// use futures_lite::io::{AsyncReadExt, Cursor};
+ ///
+ /// # spin_on::spin_on(async {
+ /// let mut reader = Cursor::new(&b"hello");
+ /// let mut contents = vec![0; 3];
+ ///
+ /// reader.read_exact(&mut contents).await?;
+ /// assert_eq!(contents, b"hel");
+ /// # std::io::Result::Ok(()) });
+ /// ```
+ fn read_exact<'a>(&'a mut self, buf: &'a mut [u8]) -> ReadExactFuture<'a, Self>
+ where
+ Self: Unpin,
+ {
+ ReadExactFuture { reader: self, buf }
+ }
+
+ /// Creates an adapter which will read at most `limit` bytes from it.
+ ///
+ /// This method returns a new instance of [`AsyncRead`] which will read at most
+ /// `limit` bytes, after which it will always return `Ok(0)` indicating EOF.
+ ///
+ /// # Examples
+ ///
+ /// ```
+ /// use futures_lite::io::{AsyncReadExt, Cursor};
+ ///
+ /// # spin_on::spin_on(async {
+ /// let mut reader = Cursor::new(&b"hello");
+ /// let mut contents = String::new();
+ ///
+ /// let n = reader.take(3).read_to_string(&mut contents).await?;
+ /// assert_eq!(n, 3);
+ /// assert_eq!(contents, "hel");
+ /// # std::io::Result::Ok(()) });
+ /// ```
+ fn take(self, limit: u64) -> Take<Self>
+ where
+ Self: Sized,
+ {
+ Take { inner: self, limit }
+ }
+
+ /// Converts this [`AsyncRead`] into a [`Stream`] of bytes.
+ ///
+ /// The returned type implements [`Stream`] where `Item` is `io::Result<u8>`.
+ ///
+ /// ```
+ /// use futures_lite::io::{AsyncReadExt, Cursor};
+ /// use futures_lite::stream::StreamExt;
+ ///
+ /// # spin_on::spin_on(async {
+ /// let reader = Cursor::new(&b"hello");
+ /// let mut bytes = reader.bytes();
+ ///
+ /// while let Some(byte) = bytes.next().await {
+ /// println!("byte: {}", byte?);
+ /// }
+ /// # std::io::Result::Ok(()) });
+ /// ```
+ fn bytes(self) -> Bytes<Self>
+ where
+ Self: Sized,
+ {
+ Bytes { inner: self }
+ }
+
+ /// Creates an adapter which will chain this stream with another.
+ ///
+ /// The returned [`AsyncRead`] instance will first read all bytes from this reader
+ /// until EOF is found, and then continue with `next`.
+ ///
+ /// # Examples
+ ///
+ /// ```
+ /// use futures_lite::io::{AsyncReadExt, Cursor};
+ ///
+ /// # spin_on::spin_on(async {
+ /// let r1 = Cursor::new(&b"hello");
+ /// let r2 = Cursor::new(&b"world");
+ /// let mut reader = r1.chain(r2);
+ ///
+ /// let mut contents = String::new();
+ /// reader.read_to_string(&mut contents).await?;
+ /// assert_eq!(contents, "helloworld");
+ /// # std::io::Result::Ok(()) });
+ /// ```
+ fn chain<R: AsyncRead>(self, next: R) -> Chain<Self, R>
+ where
+ Self: Sized,
+ {
+ Chain {
+ first: self,
+ second: next,
+ done_first: false,
+ }
+ }
+
+ /// Boxes the reader and changes its type to `dyn AsyncRead + Send + 'a`.
+ ///
+ /// # Examples
+ ///
+ /// ```
+ /// use futures_lite::io::AsyncReadExt;
+ ///
+ /// let reader = [1, 2, 3].boxed_reader();
+ /// ```
+ #[cfg(feature = "alloc")]
+ fn boxed_reader<'a>(self) -> Pin<Box<dyn AsyncRead + Send + 'a>>
+ where
+ Self: Sized + Send + 'a,
+ {
+ Box::pin(self)
+ }
+}
+
+impl<R: AsyncRead + ?Sized> AsyncReadExt for R {}
+
+/// Future for the [`AsyncReadExt::read()`] method.
+#[derive(Debug)]
+#[must_use = "futures do nothing unless you `.await` or poll them"]
+pub struct ReadFuture<'a, R: Unpin + ?Sized> {
+ reader: &'a mut R,
+ buf: &'a mut [u8],
+}
+
+impl<R: Unpin + ?Sized> Unpin for ReadFuture<'_, R> {}
+
+impl<R: AsyncRead + Unpin + ?Sized> Future for ReadFuture<'_, R> {
+ type Output = Result<usize>;
+
+ fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+ let Self { reader, buf } = &mut *self;
+ Pin::new(reader).poll_read(cx, buf)
+ }
+}
+
+/// Future for the [`AsyncReadExt::read_vectored()`] method.
+#[derive(Debug)]
+#[must_use = "futures do nothing unless you `.await` or poll them"]
+pub struct ReadVectoredFuture<'a, R: Unpin + ?Sized> {
+ reader: &'a mut R,
+ bufs: &'a mut [IoSliceMut<'a>],
+}
+
+impl<R: Unpin + ?Sized> Unpin for ReadVectoredFuture<'_, R> {}
+
+impl<R: AsyncRead + Unpin + ?Sized> Future for ReadVectoredFuture<'_, R> {
+ type Output = Result<usize>;
+
+ fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+ let Self { reader, bufs } = &mut *self;
+ Pin::new(reader).poll_read_vectored(cx, bufs)
+ }
+}
+
+/// Future for the [`AsyncReadExt::read_to_end()`] method.
+#[derive(Debug)]
+#[must_use = "futures do nothing unless you `.await` or poll them"]
+pub struct ReadToEndFuture<'a, R: Unpin + ?Sized> {
+ reader: &'a mut R,
+ buf: &'a mut Vec<u8>,
+ start_len: usize,
+}
+
+impl<R: Unpin + ?Sized> Unpin for ReadToEndFuture<'_, R> {}
+
+impl<R: AsyncRead + Unpin + ?Sized> Future for ReadToEndFuture<'_, R> {
+ type Output = Result<usize>;
+
+ fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+ let Self {
+ reader,
+ buf,
+ start_len,
+ } = &mut *self;
+ read_to_end_internal(Pin::new(reader), cx, buf, *start_len)
+ }
+}
+
+/// Future for the [`AsyncReadExt::read_to_string()`] method.
+#[derive(Debug)]
+#[must_use = "futures do nothing unless you `.await` or poll them"]
+pub struct ReadToStringFuture<'a, R: Unpin + ?Sized> {
+ reader: &'a mut R,
+ buf: &'a mut String,
+ bytes: Vec<u8>,
+ start_len: usize,
+}
+
+impl<R: Unpin + ?Sized> Unpin for ReadToStringFuture<'_, R> {}
+
+impl<R: AsyncRead + Unpin + ?Sized> Future for ReadToStringFuture<'_, R> {
+ type Output = Result<usize>;
+
+ fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+ let Self {
+ reader,
+ buf,
+ bytes,
+ start_len,
+ } = &mut *self;
+ let reader = Pin::new(reader);
+
+ let ret = ready!(read_to_end_internal(reader, cx, bytes, *start_len));
+
+ match String::from_utf8(mem::take(bytes)) {
+ Ok(s) => {
+ debug_assert!(buf.is_empty());
+ **buf = s;
+ Poll::Ready(ret)
+ }
+ Err(_) => Poll::Ready(ret.and_then(|_| {
+ Err(Error::new(
+ ErrorKind::InvalidData,
+ "stream did not contain valid UTF-8",
+ ))
+ })),
+ }
+ }
+}
+
+// This uses an adaptive system to extend the vector when it fills. We want to
+// avoid paying to allocate and zero a huge chunk of memory if the reader only
+// has 4 bytes while still making large reads if the reader does have a ton
+// of data to return. Simply tacking on an extra DEFAULT_BUF_SIZE space every
+// time is 4,500 times (!) slower than this if the reader has a very small
+// amount of data to return.
+//
+// Because we're extending the buffer with uninitialized data for trusted
+// readers, we need to make sure to truncate that if any of this panics.
+fn read_to_end_internal<R: AsyncRead + ?Sized>(
+ mut rd: Pin<&mut R>,
+ cx: &mut Context<'_>,
+ buf: &mut Vec<u8>,
+ start_len: usize,
+) -> Poll<Result<usize>> {
+ struct Guard<'a> {
+ buf: &'a mut Vec<u8>,
+ len: usize,
+ }
+
+ impl Drop for Guard<'_> {
+ fn drop(&mut self) {
+ self.buf.resize(self.len, 0);
+ }
+ }
+
+ let mut g = Guard {
+ len: buf.len(),
+ buf,
+ };
+ let ret;
+ loop {
+ if g.len == g.buf.len() {
+ g.buf.reserve(32);
+ let capacity = g.buf.capacity();
+ g.buf.resize(capacity, 0);
+ }
+
+ match ready!(rd.as_mut().poll_read(cx, &mut g.buf[g.len..])) {
+ Ok(0) => {
+ ret = Poll::Ready(Ok(g.len - start_len));
+ break;
+ }
+ Ok(n) => g.len += n,
+ Err(e) => {
+ ret = Poll::Ready(Err(e));
+ break;
+ }
+ }
+ }
+
+ ret
+}
+
+/// Future for the [`AsyncReadExt::read_exact()`] method.
+#[derive(Debug)]
+#[must_use = "futures do nothing unless you `.await` or poll them"]
+pub struct ReadExactFuture<'a, R: Unpin + ?Sized> {
+ reader: &'a mut R,
+ buf: &'a mut [u8],
+}
+
+impl<R: Unpin + ?Sized> Unpin for ReadExactFuture<'_, R> {}
+
+impl<R: AsyncRead + Unpin + ?Sized> Future for ReadExactFuture<'_, R> {
+ type Output = Result<()>;
+
+ fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+ let Self { reader, buf } = &mut *self;
+
+ while !buf.is_empty() {
+ let n = ready!(Pin::new(&mut *reader).poll_read(cx, buf))?;
+ let (_, rest) = mem::take(buf).split_at_mut(n);
+ *buf = rest;
+
+ if n == 0 {
+ return Poll::Ready(Err(ErrorKind::UnexpectedEof.into()));
+ }
+ }
+
+ Poll::Ready(Ok(()))
+ }
+}
+
+pin_project! {
+ /// Reader for the [`AsyncReadExt::take()`] method.
+ #[derive(Debug)]
+ pub struct Take<R> {
+ #[pin]
+ inner: R,
+ limit: u64,
+ }
+}
+
+impl<R> Take<R> {
+ /// Returns the number of bytes before this adapter will return EOF.
+ ///
+ /// Note that EOF may be reached sooner if the underlying reader is shorter than the limit.
+ ///
+ /// # Examples
+ ///
+ /// ```
+ /// use futures_lite::io::{AsyncReadExt, Cursor};
+ ///
+ /// let reader = Cursor::new("hello");
+ ///
+ /// let reader = reader.take(3);
+ /// assert_eq!(reader.limit(), 3);
+ /// ```
+ pub fn limit(&self) -> u64 {
+ self.limit
+ }
+
+ /// Puts a limit on the number of bytes.
+ ///
+ /// Changing the limit is equivalent to creating a new adapter with [`AsyncReadExt::take()`].
+ ///
+ /// # Examples
+ ///
+ /// ```
+ /// use futures_lite::io::{AsyncReadExt, Cursor};
+ ///
+ /// let reader = Cursor::new("hello");
+ ///
+ /// let mut reader = reader.take(10);
+ /// assert_eq!(reader.limit(), 10);
+ ///
+ /// reader.set_limit(3);
+ /// assert_eq!(reader.limit(), 3);
+ /// ```
+ pub fn set_limit(&mut self, limit: u64) {
+ self.limit = limit;
+ }
+
+ /// Gets a reference to the underlying reader.
+ ///
+ /// # Examples
+ ///
+ /// ```
+ /// use futures_lite::io::{AsyncReadExt, Cursor};
+ ///
+ /// let reader = Cursor::new("hello");
+ ///
+ /// let reader = reader.take(3);
+ /// let r = reader.get_ref();
+ /// ```
+ pub fn get_ref(&self) -> &R {
+ &self.inner
+ }
+
+ /// Gets a mutable reference to the underlying reader.
+ ///
+ /// # Examples
+ ///
+ /// ```
+ /// use futures_lite::io::{AsyncReadExt, Cursor};
+ ///
+ /// let reader = Cursor::new("hello");
+ ///
+ /// let mut reader = reader.take(3);
+ /// let r = reader.get_mut();
+ /// ```
+ pub fn get_mut(&mut self) -> &mut R {
+ &mut self.inner
+ }
+
+ /// Unwraps the adapter, returning the underlying reader.
+ ///
+ /// # Examples
+ ///
+ /// ```
+ /// use futures_lite::io::{AsyncReadExt, Cursor};
+ ///
+ /// let reader = Cursor::new("hello");
+ ///
+ /// let reader = reader.take(3);
+ /// let reader = reader.into_inner();
+ /// ```
+ pub fn into_inner(self) -> R {
+ self.inner
+ }
+}
+
+impl<R: AsyncRead> AsyncRead for Take<R> {
+ fn poll_read(
+ self: Pin<&mut Self>,
+ cx: &mut Context<'_>,
+ buf: &mut [u8],
+ ) -> Poll<Result<usize>> {
+ let this = self.project();
+ take_read_internal(this.inner, cx, buf, this.limit)
+ }
+}
+
+fn take_read_internal<R: AsyncRead + ?Sized>(
+ mut rd: Pin<&mut R>,
+ cx: &mut Context<'_>,
+ buf: &mut [u8],
+ limit: &mut u64,
+) -> Poll<Result<usize>> {
+ // Don't call into inner reader at all at EOF because it may still block
+ if *limit == 0 {
+ return Poll::Ready(Ok(0));
+ }
+
+ let max = cmp::min(buf.len() as u64, *limit) as usize;
+
+ match ready!(rd.as_mut().poll_read(cx, &mut buf[..max])) {
+ Ok(n) => {
+ *limit -= n as u64;
+ Poll::Ready(Ok(n))
+ }
+ Err(e) => Poll::Ready(Err(e)),
+ }
+}
+
+impl<R: AsyncBufRead> AsyncBufRead for Take<R> {
+ fn poll_fill_buf(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<&[u8]>> {
+ let this = self.project();
+
+ if *this.limit == 0 {
+ return Poll::Ready(Ok(&[]));
+ }
+
+ match ready!(this.inner.poll_fill_buf(cx)) {
+ Ok(buf) => {
+ let cap = cmp::min(buf.len() as u64, *this.limit) as usize;
+ Poll::Ready(Ok(&buf[..cap]))
+ }
+ Err(e) => Poll::Ready(Err(e)),
+ }
+ }
+
+ fn consume(self: Pin<&mut Self>, amt: usize) {
+ let this = self.project();
+ // Don't let callers reset the limit by passing an overlarge value
+ let amt = cmp::min(amt as u64, *this.limit) as usize;
+ *this.limit -= amt as u64;
+
+ this.inner.consume(amt);
+ }
+}
+
+pin_project! {
+ /// Reader for the [`AsyncReadExt::bytes()`] method.
+ #[derive(Debug)]
+ pub struct Bytes<R> {
+ #[pin]
+ inner: R,
+ }
+}
+
+impl<R: AsyncRead + Unpin> Stream for Bytes<R> {
+ type Item = Result<u8>;
+
+ fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+ let mut byte = 0;
+
+ let rd = Pin::new(&mut self.inner);
+
+ match ready!(rd.poll_read(cx, std::slice::from_mut(&mut byte))) {
+ Ok(0) => Poll::Ready(None),
+ Ok(..) => Poll::Ready(Some(Ok(byte))),
+ Err(ref e) if e.kind() == ErrorKind::Interrupted => Poll::Pending,
+ Err(e) => Poll::Ready(Some(Err(e))),
+ }
+ }
+}
+
+impl<R: AsyncRead> AsyncRead for Bytes<R> {
+ fn poll_read(
+ self: Pin<&mut Self>,
+ cx: &mut Context<'_>,
+ buf: &mut [u8],
+ ) -> Poll<Result<usize>> {
+ self.project().inner.poll_read(cx, buf)
+ }
+
+ fn poll_read_vectored(
+ self: Pin<&mut Self>,
+ cx: &mut Context<'_>,
+ bufs: &mut [IoSliceMut<'_>],
+ ) -> Poll<Result<usize>> {
+ self.project().inner.poll_read_vectored(cx, bufs)
+ }
+}
+
+pin_project! {
+ /// Reader for the [`AsyncReadExt::chain()`] method.
+ pub struct Chain<R1, R2> {
+ #[pin]
+ first: R1,
+ #[pin]
+ second: R2,
+ done_first: bool,
+ }
+}
+
+impl<R1, R2> Chain<R1, R2> {
+ /// Gets references to the underlying readers.
+ ///
+ /// # Examples
+ ///
+ /// ```
+ /// use futures_lite::io::{AsyncReadExt, Cursor};
+ ///
+ /// let r1 = Cursor::new(b"hello");
+ /// let r2 = Cursor::new(b"world");
+ ///
+ /// let reader = r1.chain(r2);
+ /// let (r1, r2) = reader.get_ref();
+ /// ```
+ pub fn get_ref(&self) -> (&R1, &R2) {
+ (&self.first, &self.second)
+ }
+
+ /// Gets mutable references to the underlying readers.
+ ///
+ /// # Examples
+ ///
+ /// ```
+ /// use futures_lite::io::{AsyncReadExt, Cursor};
+ ///
+ /// let r1 = Cursor::new(b"hello");
+ /// let r2 = Cursor::new(b"world");
+ ///
+ /// let mut reader = r1.chain(r2);
+ /// let (r1, r2) = reader.get_mut();
+ /// ```
+ pub fn get_mut(&mut self) -> (&mut R1, &mut R2) {
+ (&mut self.first, &mut self.second)
+ }
+
+ /// Unwraps the adapter, returning the underlying readers.
+ ///
+ /// # Examples
+ ///
+ /// ```
+ /// use futures_lite::io::{AsyncReadExt, Cursor};
+ ///
+ /// let r1 = Cursor::new(b"hello");
+ /// let r2 = Cursor::new(b"world");
+ ///
+ /// let reader = r1.chain(r2);
+ /// let (r1, r2) = reader.into_inner();
+ /// ```
+ pub fn into_inner(self) -> (R1, R2) {
+ (self.first, self.second)
+ }
+}
+
+impl<R1: fmt::Debug, R2: fmt::Debug> fmt::Debug for Chain<R1, R2> {
+ fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+ f.debug_struct("Chain")
+ .field("r1", &self.first)
+ .field("r2", &self.second)
+ .finish()
+ }
+}
+
+impl<R1: AsyncRead, R2: AsyncRead> AsyncRead for Chain<R1, R2> {
+ fn poll_read(
+ self: Pin<&mut Self>,
+ cx: &mut Context<'_>,
+ buf: &mut [u8],
+ ) -> Poll<Result<usize>> {
+ let this = self.project();
+ if !*this.done_first {
+ match ready!(this.first.poll_read(cx, buf)) {
+ Ok(0) if !buf.is_empty() => *this.done_first = true,
+ Ok(n) => return Poll::Ready(Ok(n)),
+ Err(err) => return Poll::Ready(Err(err)),
+ }
+ }
+
+ this.second.poll_read(cx, buf)
+ }
+
+ fn poll_read_vectored(
+ self: Pin<&mut Self>,
+ cx: &mut Context<'_>,
+ bufs: &mut [IoSliceMut<'_>],
+ ) -> Poll<Result<usize>> {
+ let this = self.project();
+ if !*this.done_first {
+ match ready!(this.first.poll_read_vectored(cx, bufs)) {
+ Ok(0) if !bufs.is_empty() => *this.done_first = true,
+ Ok(n) => return Poll::Ready(Ok(n)),
+ Err(err) => return Poll::Ready(Err(err)),
+ }
+ }
+
+ this.second.poll_read_vectored(cx, bufs)
+ }
+}
+
+impl<R1: AsyncBufRead, R2: AsyncBufRead> AsyncBufRead for Chain<R1, R2> {
+ fn poll_fill_buf(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<&[u8]>> {
+ let this = self.project();
+ if !*this.done_first {
+ match ready!(this.first.poll_fill_buf(cx)) {
+ Ok([]) => *this.done_first = true,
+ Ok(buf) => return Poll::Ready(Ok(buf)),
+ Err(err) => return Poll::Ready(Err(err)),
+ }
+ }
+
+ this.second.poll_fill_buf(cx)
+ }
+
+ fn consume(self: Pin<&mut Self>, amt: usize) {
+ let this = self.project();
+ if !*this.done_first {
+ this.first.consume(amt)
+ } else {
+ this.second.consume(amt)
+ }
+ }
+}
+
+/// Extension trait for [`AsyncSeek`].
+pub trait AsyncSeekExt: AsyncSeek {
+ /// Seeks to a new position in a byte stream.
+ ///
+ /// Returns the new position in the byte stream.
+ ///
+ /// A seek beyond the end of stream is allowed, but behavior is defined by the implementation.
+ ///
+ /// # Examples
+ ///
+ /// ```
+ /// use futures_lite::io::{AsyncSeekExt, Cursor, SeekFrom};
+ ///
+ /// # spin_on::spin_on(async {
+ /// let mut cursor = Cursor::new("hello");
+ ///
+ /// // Move the cursor to the end.
+ /// cursor.seek(SeekFrom::End(0)).await?;
+ ///
+ /// // Check the current position.
+ /// assert_eq!(cursor.seek(SeekFrom::Current(0)).await?, 5);
+ /// # std::io::Result::Ok(()) });
+ /// ```
+ fn seek(&mut self, pos: SeekFrom) -> SeekFuture<'_, Self>
+ where
+ Self: Unpin,
+ {
+ SeekFuture { seeker: self, pos }
+ }
+}
+
+impl<S: AsyncSeek + ?Sized> AsyncSeekExt for S {}
+
+/// Future for the [`AsyncSeekExt::seek()`] method.
+#[derive(Debug)]
+#[must_use = "futures do nothing unless you `.await` or poll them"]
+pub struct SeekFuture<'a, S: Unpin + ?Sized> {
+ seeker: &'a mut S,
+ pos: SeekFrom,
+}
+
+impl<S: Unpin + ?Sized> Unpin for SeekFuture<'_, S> {}
+
+impl<S: AsyncSeek + Unpin + ?Sized> Future for SeekFuture<'_, S> {
+ type Output = Result<u64>;
+
+ fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+ let pos = self.pos;
+ Pin::new(&mut *self.seeker).poll_seek(cx, pos)
+ }
+}
+
+/// Extension trait for [`AsyncWrite`].
+pub trait AsyncWriteExt: AsyncWrite {
+ /// Writes some bytes into the byte stream.
+ ///
+ /// Returns the number of bytes written from the start of the buffer.
+ ///
+ /// If the return value is `Ok(n)` then it must be guaranteed that
+ /// `0 <= n <= buf.len()`. A return value of `0` typically means that the underlying
+ /// object is no longer able to accept bytes and will likely not be able to in the
+ /// future as well, or that the provided buffer is empty.
+ ///
+ /// # Examples
+ ///
+ /// ```
+ /// use futures_lite::io::{AsyncWriteExt, BufWriter};
+ ///
+ /// # spin_on::spin_on(async {
+ /// let mut output = Vec::new();
+ /// let mut writer = BufWriter::new(&mut output);
+ ///
+ /// let n = writer.write(b"hello").await?;
+ /// # std::io::Result::Ok(()) });
+ /// ```
+ fn write<'a>(&'a mut self, buf: &'a [u8]) -> WriteFuture<'a, Self>
+ where
+ Self: Unpin,
+ {
+ WriteFuture { writer: self, buf }
+ }
+
+ /// Like [`write()`][`AsyncWriteExt::write()`], except that it writes a slice of buffers.
+ ///
+ /// Data is copied from each buffer in order, with the final buffer possibly being only
+ /// partially consumed. This method must behave same as a call to
+ /// [`write()`][`AsyncWriteExt::write()`] with the buffers concatenated would.
+ fn write_vectored<'a>(&'a mut self, bufs: &'a [IoSlice<'a>]) -> WriteVectoredFuture<'a, Self>
+ where
+ Self: Unpin,
+ {
+ WriteVectoredFuture { writer: self, bufs }
+ }
+
+ /// Writes an entire buffer into the byte stream.
+ ///
+ /// This method will keep calling [`write()`][`AsyncWriteExt::write()`] until there is no more
+ /// data to be written or an error occurs. It will not return before the entire buffer is
+ /// successfully written or an error occurs.
+ ///
+ /// # Examples
+ ///
+ /// ```
+ /// use futures_lite::io::{AsyncWriteExt, BufWriter};
+ ///
+ /// # spin_on::spin_on(async {
+ /// let mut output = Vec::new();
+ /// let mut writer = BufWriter::new(&mut output);
+ ///
+ /// let n = writer.write_all(b"hello").await?;
+ /// # std::io::Result::Ok(()) });
+ /// ```
+ fn write_all<'a>(&'a mut self, buf: &'a [u8]) -> WriteAllFuture<'a, Self>
+ where
+ Self: Unpin,
+ {
+ WriteAllFuture { writer: self, buf }
+ }
+
+ /// Flushes the stream to ensure that all buffered contents reach their destination.
+ ///
+ /// # Examples
+ ///
+ /// ```
+ /// use futures_lite::io::{AsyncWriteExt, BufWriter};
+ ///
+ /// # spin_on::spin_on(async {
+ /// let mut output = Vec::new();
+ /// let mut writer = BufWriter::new(&mut output);
+ ///
+ /// writer.write_all(b"hello").await?;
+ /// writer.flush().await?;
+ /// # std::io::Result::Ok(()) });
+ /// ```
+ fn flush(&mut self) -> FlushFuture<'_, Self>
+ where
+ Self: Unpin,
+ {
+ FlushFuture { writer: self }
+ }
+
+ /// Closes the writer.
+ ///
+ /// # Examples
+ ///
+ /// ```
+ /// use futures_lite::io::{AsyncWriteExt, BufWriter};
+ ///
+ /// # spin_on::spin_on(async {
+ /// let mut output = Vec::new();
+ /// let mut writer = BufWriter::new(&mut output);
+ ///
+ /// writer.close().await?;
+ /// # std::io::Result::Ok(()) });
+ /// ```
+ fn close(&mut self) -> CloseFuture<'_, Self>
+ where
+ Self: Unpin,
+ {
+ CloseFuture { writer: self }
+ }
+
+ /// Boxes the writer and changes its type to `dyn AsyncWrite + Send + 'a`.
+ ///
+ /// # Examples
+ ///
+ /// ```
+ /// use futures_lite::io::AsyncWriteExt;
+ ///
+ /// let writer = Vec::<u8>::new().boxed_writer();
+ /// ```
+ #[cfg(feature = "alloc")]
+ fn boxed_writer<'a>(self) -> Pin<Box<dyn AsyncWrite + Send + 'a>>
+ where
+ Self: Sized + Send + 'a,
+ {
+ Box::pin(self)
+ }
+}
+
+impl<W: AsyncWrite + ?Sized> AsyncWriteExt for W {}
+
+/// Future for the [`AsyncWriteExt::write()`] method.
+#[derive(Debug)]
+#[must_use = "futures do nothing unless you `.await` or poll them"]
+pub struct WriteFuture<'a, W: Unpin + ?Sized> {
+ writer: &'a mut W,
+ buf: &'a [u8],
+}
+
+impl<W: Unpin + ?Sized> Unpin for WriteFuture<'_, W> {}
+
+impl<W: AsyncWrite + Unpin + ?Sized> Future for WriteFuture<'_, W> {
+ type Output = Result<usize>;
+
+ fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+ let buf = self.buf;
+ Pin::new(&mut *self.writer).poll_write(cx, buf)
+ }
+}
+
+/// Future for the [`AsyncWriteExt::write_vectored()`] method.
+#[derive(Debug)]
+#[must_use = "futures do nothing unless you `.await` or poll them"]
+pub struct WriteVectoredFuture<'a, W: Unpin + ?Sized> {
+ writer: &'a mut W,
+ bufs: &'a [IoSlice<'a>],
+}
+
+impl<W: Unpin + ?Sized> Unpin for WriteVectoredFuture<'_, W> {}
+
+impl<W: AsyncWrite + Unpin + ?Sized> Future for WriteVectoredFuture<'_, W> {
+ type Output = Result<usize>;
+
+ fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+ let bufs = self.bufs;
+ Pin::new(&mut *self.writer).poll_write_vectored(cx, bufs)
+ }
+}
+
+/// Future for the [`AsyncWriteExt::write_all()`] method.
+#[derive(Debug)]
+#[must_use = "futures do nothing unless you `.await` or poll them"]
+pub struct WriteAllFuture<'a, W: Unpin + ?Sized> {
+ writer: &'a mut W,
+ buf: &'a [u8],
+}
+
+impl<W: Unpin + ?Sized> Unpin for WriteAllFuture<'_, W> {}
+
+impl<W: AsyncWrite + Unpin + ?Sized> Future for WriteAllFuture<'_, W> {
+ type Output = Result<()>;
+
+ fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+ let Self { writer, buf } = &mut *self;
+
+ while !buf.is_empty() {
+ let n = ready!(Pin::new(&mut **writer).poll_write(cx, buf))?;
+ let (_, rest) = mem::take(buf).split_at(n);
+ *buf = rest;
+
+ if n == 0 {
+ return Poll::Ready(Err(ErrorKind::WriteZero.into()));
+ }
+ }
+
+ Poll::Ready(Ok(()))
+ }
+}
+
+/// Future for the [`AsyncWriteExt::flush()`] method.
+#[derive(Debug)]
+#[must_use = "futures do nothing unless you `.await` or poll them"]
+pub struct FlushFuture<'a, W: Unpin + ?Sized> {
+ writer: &'a mut W,
+}
+
+impl<W: Unpin + ?Sized> Unpin for FlushFuture<'_, W> {}
+
+impl<W: AsyncWrite + Unpin + ?Sized> Future for FlushFuture<'_, W> {
+ type Output = Result<()>;
+
+ fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+ Pin::new(&mut *self.writer).poll_flush(cx)
+ }
+}
+
+/// Future for the [`AsyncWriteExt::close()`] method.
+#[derive(Debug)]
+#[must_use = "futures do nothing unless you `.await` or poll them"]
+pub struct CloseFuture<'a, W: Unpin + ?Sized> {
+ writer: &'a mut W,
+}
+
+impl<W: Unpin + ?Sized> Unpin for CloseFuture<'_, W> {}
+
+impl<W: AsyncWrite + Unpin + ?Sized> Future for CloseFuture<'_, W> {
+ type Output = Result<()>;
+
+ fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+ Pin::new(&mut *self.writer).poll_close(cx)
+ }
+}
+
+/// Type alias for `Pin<Box<dyn AsyncRead + Send + 'static>>`.
+///
+/// # Examples
+///
+/// ```
+/// use futures_lite::io::AsyncReadExt;
+///
+/// let reader = [1, 2, 3].boxed_reader();
+/// ```
+#[cfg(feature = "alloc")]
+pub type BoxedReader = Pin<Box<dyn AsyncRead + Send + 'static>>;
+
+/// Type alias for `Pin<Box<dyn AsyncWrite + Send + 'static>>`.
+///
+/// # Examples
+///
+/// ```
+/// use futures_lite::io::AsyncWriteExt;
+///
+/// let writer = Vec::<u8>::new().boxed_writer();
+/// ```
+#[cfg(feature = "alloc")]
+pub type BoxedWriter = Pin<Box<dyn AsyncWrite + Send + 'static>>;
+
+/// Splits a stream into [`AsyncRead`] and [`AsyncWrite`] halves.
+///
+/// # Examples
+///
+/// ```
+/// use futures_lite::io::{self, Cursor};
+///
+/// # spin_on::spin_on(async {
+/// let stream = Cursor::new(vec![]);
+/// let (mut reader, mut writer) = io::split(stream);
+/// # std::io::Result::Ok(()) });
+/// ```
+pub fn split<T>(stream: T) -> (ReadHalf<T>, WriteHalf<T>)
+where
+ T: AsyncRead + AsyncWrite + Unpin,
+{
+ let inner = Arc::new(Mutex::new(stream));
+ (ReadHalf(inner.clone()), WriteHalf(inner))
+}
+
+/// The read half returned by [`split()`].
+#[derive(Debug)]
+pub struct ReadHalf<T>(Arc<Mutex<T>>);
+
+/// The write half returned by [`split()`].
+#[derive(Debug)]
+pub struct WriteHalf<T>(Arc<Mutex<T>>);
+
+impl<T: AsyncRead + Unpin> AsyncRead for ReadHalf<T> {
+ fn poll_read(
+ self: Pin<&mut Self>,
+ cx: &mut Context<'_>,
+ buf: &mut [u8],
+ ) -> Poll<Result<usize>> {
+ let mut inner = self.0.lock().unwrap();
+ Pin::new(&mut *inner).poll_read(cx, buf)
+ }
+
+ fn poll_read_vectored(
+ self: Pin<&mut Self>,
+ cx: &mut Context<'_>,
+ bufs: &mut [IoSliceMut<'_>],
+ ) -> Poll<Result<usize>> {
+ let mut inner = self.0.lock().unwrap();
+ Pin::new(&mut *inner).poll_read_vectored(cx, bufs)
+ }
+}
+
+impl<T: AsyncWrite + Unpin> AsyncWrite for WriteHalf<T> {
+ fn poll_write(self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &[u8]) -> Poll<Result<usize>> {
+ let mut inner = self.0.lock().unwrap();
+ Pin::new(&mut *inner).poll_write(cx, buf)
+ }
+
+ fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<()>> {
+ let mut inner = self.0.lock().unwrap();
+ Pin::new(&mut *inner).poll_flush(cx)
+ }
+
+ fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<()>> {
+ let mut inner = self.0.lock().unwrap();
+ Pin::new(&mut *inner).poll_close(cx)
+ }
+}
+
+#[cfg(feature = "memchr")]
+use memchr::memchr;
+
+/// Unoptimized memchr fallback.
+#[cfg(not(feature = "memchr"))]
+fn memchr(needle: u8, haystack: &[u8]) -> Option<usize> {
+ haystack.iter().position(|&b| b == needle)
+}
diff --git a/external/vendor/futures-lite/src/lib.rs b/external/vendor/futures-lite/src/lib.rs
new file mode 100644
index 0000000..507a271
--- /dev/null
+++ b/external/vendor/futures-lite/src/lib.rs
@@ -0,0 +1,147 @@
+//! Futures, streams, and async I/O combinators.
+//!
+//! This crate is a subset of [futures] that compiles an order of magnitude faster, fixes minor
+//! warts in its API, fills in some obvious gaps, and removes almost all unsafe code from it.
+//!
+//! In short, this crate aims to be more enjoyable than [futures] but still fully compatible with
+//! it.
+//!
+//! The API for this crate is intentionally constrained. Please consult the [features list] for
+//! APIs that are occluded from this crate.
+//!
+//! [futures]: https://docs.rs/futures
+//! [features list]: https://github.com/smol-rs/futures-lite/blob/master/FEATURES.md
+//!
+//! # Examples
+//!
+#![cfg_attr(feature = "std", doc = "```no_run")]
+#![cfg_attr(not(feature = "std"), doc = "```ignore")]
+//! use futures_lite::future;
+//!
+//! fn main() {
+//! future::block_on(async {
+//! println!("Hello world!");
+//! })
+//! }
+//! ```
+
+#![no_std]
+#![warn(missing_docs, missing_debug_implementations, rust_2018_idioms)]
+#![allow(clippy::needless_borrow)] // suggest code that doesn't work on MSRV
+#![doc(
+ html_favicon_url = "https://raw.githubusercontent.com/smol-rs/smol/master/assets/images/logo_fullsize_transparent.png"
+)]
+#![doc(
+ html_logo_url = "https://raw.githubusercontent.com/smol-rs/smol/master/assets/images/logo_fullsize_transparent.png"
+)]
+#![cfg_attr(docsrs, feature(doc_auto_cfg))]
+
+#[cfg(feature = "alloc")]
+extern crate alloc;
+
+#[cfg(feature = "std")]
+extern crate std;
+
+#[cfg(feature = "std")]
+#[doc(no_inline)]
+pub use crate::io::{
+ AsyncBufRead, AsyncBufReadExt, AsyncRead, AsyncReadExt, AsyncSeek, AsyncSeekExt, AsyncWrite,
+ AsyncWriteExt,
+};
+#[doc(no_inline)]
+pub use crate::{
+ future::{Future, FutureExt},
+ stream::{Stream, StreamExt},
+};
+
+pub mod future;
+pub mod prelude;
+pub mod stream;
+
+#[cfg(feature = "std")]
+pub mod io;
+
+/// Unwraps `Poll<T>` or returns [`Pending`][`core::task::Poll::Pending`].
+///
+/// # Examples
+///
+/// ```
+/// use futures_lite::{future, prelude::*, ready};
+/// use std::pin::Pin;
+/// use std::task::{Context, Poll};
+///
+/// fn do_poll(cx: &mut Context<'_>) -> Poll<()> {
+/// let mut fut = future::ready(42);
+/// let fut = Pin::new(&mut fut);
+///
+/// let num = ready!(fut.poll(cx));
+/// # drop(num);
+/// // ... use num
+///
+/// Poll::Ready(())
+/// }
+/// ```
+///
+/// The `ready!` call expands to:
+///
+/// ```
+/// # use futures_lite::{future, prelude::*, ready};
+/// # use std::pin::Pin;
+/// # use std::task::{Context, Poll};
+/// #
+/// # fn do_poll(cx: &mut Context<'_>) -> Poll<()> {
+/// # let mut fut = future::ready(42);
+/// # let fut = Pin::new(&mut fut);
+/// #
+/// let num = match fut.poll(cx) {
+/// Poll::Ready(t) => t,
+/// Poll::Pending => return Poll::Pending,
+/// };
+/// # drop(num);
+/// # // ... use num
+/// #
+/// # Poll::Ready(())
+/// # }
+/// ```
+#[macro_export]
+macro_rules! ready {
+ ($e:expr $(,)?) => {
+ match $e {
+ core::task::Poll::Ready(t) => t,
+ core::task::Poll::Pending => return core::task::Poll::Pending,
+ }
+ };
+}
+
+/// Pins a variable of type `T` on the stack and rebinds it as `Pin<&mut T>`.
+///
+/// ```
+/// use futures_lite::{future, pin};
+/// use std::fmt::Debug;
+/// use std::future::Future;
+/// use std::pin::Pin;
+/// use std::time::Instant;
+///
+/// // Inspects each invocation of `Future::poll()`.
+/// async fn inspect<T: Debug>(f: impl Future<Output = T>) -> T {
+/// pin!(f);
+/// future::poll_fn(|cx| dbg!(f.as_mut().poll(cx))).await
+/// }
+///
+/// # spin_on::spin_on(async {
+/// let f = async { 1 + 2 };
+/// inspect(f).await;
+/// # })
+/// ```
+#[macro_export]
+macro_rules! pin {
+ ($($x:ident),* $(,)?) => {
+ $(
+ let mut $x = $x;
+ #[allow(unused_mut)]
+ let mut $x = unsafe {
+ core::pin::Pin::new_unchecked(&mut $x)
+ };
+ )*
+ }
+}
diff --git a/external/vendor/futures-lite/src/prelude.rs b/external/vendor/futures-lite/src/prelude.rs
new file mode 100644
index 0000000..48c6cc9
--- /dev/null
+++ b/external/vendor/futures-lite/src/prelude.rs
@@ -0,0 +1,23 @@
+//! Traits [`Future`], [`Stream`], [`AsyncRead`], [`AsyncWrite`], [`AsyncBufRead`],
+//! [`AsyncSeek`], and their extensions.
+//!
+//! # Examples
+//!
+//! ```
+//! use futures_lite::prelude::*;
+//! ```
+
+#[doc(no_inline)]
+pub use crate::{
+ future::{Future, FutureExt as _},
+ stream::{Stream, StreamExt as _},
+};
+
+#[cfg(feature = "std")]
+#[doc(no_inline)]
+pub use crate::{
+ io::{AsyncBufRead, AsyncBufReadExt as _},
+ io::{AsyncRead, AsyncReadExt as _},
+ io::{AsyncSeek, AsyncSeekExt as _},
+ io::{AsyncWrite, AsyncWriteExt as _},
+};
diff --git a/external/vendor/futures-lite/src/stream.rs b/external/vendor/futures-lite/src/stream.rs
new file mode 100644
index 0000000..4e0e307
--- /dev/null
+++ b/external/vendor/futures-lite/src/stream.rs
@@ -0,0 +1,3542 @@
+//! Combinators for the [`Stream`] trait.
+//!
+//! # Examples
+//!
+//! ```
+//! use futures_lite::stream::{self, StreamExt};
+//!
+//! # spin_on::spin_on(async {
+//! let mut s = stream::iter(vec![1, 2, 3]);
+//!
+//! assert_eq!(s.next().await, Some(1));
+//! assert_eq!(s.next().await, Some(2));
+//! assert_eq!(s.next().await, Some(3));
+//! assert_eq!(s.next().await, None);
+//! # });
+//! ```
+
+#[doc(no_inline)]
+pub use futures_core::stream::Stream;
+
+#[cfg(feature = "alloc")]
+use alloc::boxed::Box;
+
+use core::fmt;
+use core::future::Future;
+use core::marker::PhantomData;
+use core::mem;
+use core::pin::Pin;
+use core::task::{Context, Poll};
+
+#[cfg(feature = "race")]
+use fastrand::Rng;
+
+use pin_project_lite::pin_project;
+
+use crate::ready;
+
+/// Converts a stream into a blocking iterator.
+///
+/// # Examples
+///
+/// ```
+/// use futures_lite::{pin, stream};
+///
+/// let stream = stream::once(7);
+/// pin!(stream);
+///
+/// let mut iter = stream::block_on(stream);
+/// assert_eq!(iter.next(), Some(7));
+/// assert_eq!(iter.next(), None);
+/// ```
+#[cfg(feature = "std")]
+pub fn block_on<S: Stream + Unpin>(stream: S) -> BlockOn<S> {
+ BlockOn(stream)
+}
+
+/// Iterator for the [`block_on()`] function.
+#[derive(Debug)]
+pub struct BlockOn<S>(S);
+
+#[cfg(feature = "std")]
+impl<S: Stream + Unpin> Iterator for BlockOn<S> {
+ type Item = S::Item;
+
+ fn next(&mut self) -> Option<Self::Item> {
+ crate::future::block_on(self.0.next())
+ }
+
+ fn size_hint(&self) -> (usize, Option<usize>) {
+ self.0.size_hint()
+ }
+
+ fn count(self) -> usize {
+ crate::future::block_on(self.0.count())
+ }
+
+ fn last(self) -> Option<Self::Item> {
+ crate::future::block_on(self.0.last())
+ }
+
+ fn nth(&mut self, n: usize) -> Option<Self::Item> {
+ crate::future::block_on(self.0.nth(n))
+ }
+
+ fn fold<B, F>(self, init: B, f: F) -> B
+ where
+ F: FnMut(B, Self::Item) -> B,
+ {
+ crate::future::block_on(self.0.fold(init, f))
+ }
+
+ fn for_each<F>(self, f: F) -> F::Output
+ where
+ F: FnMut(Self::Item),
+ {
+ crate::future::block_on(self.0.for_each(f))
+ }
+
+ fn all<F>(&mut self, f: F) -> bool
+ where
+ F: FnMut(Self::Item) -> bool,
+ {
+ crate::future::block_on(self.0.all(f))
+ }
+
+ fn any<F>(&mut self, f: F) -> bool
+ where
+ F: FnMut(Self::Item) -> bool,
+ {
+ crate::future::block_on(self.0.any(f))
+ }
+
+ fn find<P>(&mut self, predicate: P) -> Option<Self::Item>
+ where
+ P: FnMut(&Self::Item) -> bool,
+ {
+ crate::future::block_on(self.0.find(predicate))
+ }
+
+ fn find_map<B, F>(&mut self, f: F) -> Option<B>
+ where
+ F: FnMut(Self::Item) -> Option<B>,
+ {
+ crate::future::block_on(self.0.find_map(f))
+ }
+
+ fn position<P>(&mut self, predicate: P) -> Option<usize>
+ where
+ P: FnMut(Self::Item) -> bool,
+ {
+ crate::future::block_on(self.0.position(predicate))
+ }
+}
+
+/// Creates an empty stream.
+///
+/// # Examples
+///
+/// ```
+/// use futures_lite::stream::{self, StreamExt};
+///
+/// # spin_on::spin_on(async {
+/// let mut s = stream::empty::<i32>();
+/// assert_eq!(s.next().await, None);
+/// # })
+/// ```
+pub fn empty<T>() -> Empty<T> {
+ Empty {
+ _marker: PhantomData,
+ }
+}
+
+/// Stream for the [`empty()`] function.
+#[derive(Clone, Debug)]
+#[must_use = "streams do nothing unless polled"]
+pub struct Empty<T> {
+ _marker: PhantomData<T>,
+}
+
+impl<T> Unpin for Empty<T> {}
+
+impl<T> Stream for Empty<T> {
+ type Item = T;
+
+ fn poll_next(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+ Poll::Ready(None)
+ }
+
+ fn size_hint(&self) -> (usize, Option<usize>) {
+ (0, Some(0))
+ }
+}
+
+/// Creates a stream from an iterator.
+///
+/// # Examples
+///
+/// ```
+/// use futures_lite::stream::{self, StreamExt};
+///
+/// # spin_on::spin_on(async {
+/// let mut s = stream::iter(vec![1, 2]);
+///
+/// assert_eq!(s.next().await, Some(1));
+/// assert_eq!(s.next().await, Some(2));
+/// assert_eq!(s.next().await, None);
+/// # })
+/// ```
+pub fn iter<I: IntoIterator>(iter: I) -> Iter<I::IntoIter> {
+ Iter {
+ iter: iter.into_iter(),
+ }
+}
+
+/// Stream for the [`iter()`] function.
+#[derive(Clone, Debug)]
+#[must_use = "streams do nothing unless polled"]
+pub struct Iter<I> {
+ iter: I,
+}
+
+impl<I> Unpin for Iter<I> {}
+
+impl<I: Iterator> Stream for Iter<I> {
+ type Item = I::Item;
+
+ fn poll_next(mut self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+ Poll::Ready(self.iter.next())
+ }
+
+ fn size_hint(&self) -> (usize, Option<usize>) {
+ self.iter.size_hint()
+ }
+}
+
+/// Creates a stream that yields a single item.
+///
+/// # Examples
+///
+/// ```
+/// use futures_lite::stream::{self, StreamExt};
+///
+/// # spin_on::spin_on(async {
+/// let mut s = stream::once(7);
+///
+/// assert_eq!(s.next().await, Some(7));
+/// assert_eq!(s.next().await, None);
+/// # })
+/// ```
+pub fn once<T>(t: T) -> Once<T> {
+ Once { value: Some(t) }
+}
+
+pin_project! {
+ /// Stream for the [`once()`] function.
+ #[derive(Clone, Debug)]
+ #[must_use = "streams do nothing unless polled"]
+ pub struct Once<T> {
+ value: Option<T>,
+ }
+}
+
+impl<T> Stream for Once<T> {
+ type Item = T;
+
+ fn poll_next(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<Option<T>> {
+ Poll::Ready(self.project().value.take())
+ }
+
+ fn size_hint(&self) -> (usize, Option<usize>) {
+ if self.value.is_some() {
+ (1, Some(1))
+ } else {
+ (0, Some(0))
+ }
+ }
+}
+
+/// Creates a stream that is always pending.
+///
+/// # Examples
+///
+/// ```no_run
+/// use futures_lite::stream::{self, StreamExt};
+///
+/// # spin_on::spin_on(async {
+/// let mut s = stream::pending::<i32>();
+/// s.next().await;
+/// unreachable!();
+/// # })
+/// ```
+pub fn pending<T>() -> Pending<T> {
+ Pending {
+ _marker: PhantomData,
+ }
+}
+
+/// Stream for the [`pending()`] function.
+#[derive(Clone, Debug)]
+#[must_use = "streams do nothing unless polled"]
+pub struct Pending<T> {
+ _marker: PhantomData<T>,
+}
+
+impl<T> Unpin for Pending<T> {}
+
+impl<T> Stream for Pending<T> {
+ type Item = T;
+
+ fn poll_next(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<Option<T>> {
+ Poll::Pending
+ }
+
+ fn size_hint(&self) -> (usize, Option<usize>) {
+ (0, Some(0))
+ }
+}
+
+/// Creates a stream from a function returning [`Poll`].
+///
+/// # Examples
+///
+/// ```
+/// use futures_lite::stream::{self, StreamExt};
+/// use std::task::{Context, Poll};
+///
+/// # spin_on::spin_on(async {
+/// fn f(_: &mut Context<'_>) -> Poll<Option<i32>> {
+/// Poll::Ready(Some(7))
+/// }
+///
+/// assert_eq!(stream::poll_fn(f).next().await, Some(7));
+/// # })
+/// ```
+pub fn poll_fn<T, F>(f: F) -> PollFn<F>
+where
+ F: FnMut(&mut Context<'_>) -> Poll<Option<T>>,
+{
+ PollFn { f }
+}
+
+/// Stream for the [`poll_fn()`] function.
+#[derive(Clone)]
+#[must_use = "streams do nothing unless polled"]
+pub struct PollFn<F> {
+ f: F,
+}
+
+impl<F> Unpin for PollFn<F> {}
+
+impl<F> fmt::Debug for PollFn<F> {
+ fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+ f.debug_struct("PollFn").finish()
+ }
+}
+
+impl<T, F> Stream for PollFn<F>
+where
+ F: FnMut(&mut Context<'_>) -> Poll<Option<T>>,
+{
+ type Item = T;
+
+ fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<T>> {
+ (&mut self.f)(cx)
+ }
+}
+
+/// Creates an infinite stream that yields the same item repeatedly.
+///
+/// # Examples
+///
+/// ```
+/// use futures_lite::stream::{self, StreamExt};
+///
+/// # spin_on::spin_on(async {
+/// let mut s = stream::repeat(7);
+///
+/// assert_eq!(s.next().await, Some(7));
+/// assert_eq!(s.next().await, Some(7));
+/// # })
+/// ```
+pub fn repeat<T: Clone>(item: T) -> Repeat<T> {
+ Repeat { item }
+}
+
+/// Stream for the [`repeat()`] function.
+#[derive(Clone, Debug)]
+#[must_use = "streams do nothing unless polled"]
+pub struct Repeat<T> {
+ item: T,
+}
+
+impl<T> Unpin for Repeat<T> {}
+
+impl<T: Clone> Stream for Repeat<T> {
+ type Item = T;
+
+ fn poll_next(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+ Poll::Ready(Some(self.item.clone()))
+ }
+
+ fn size_hint(&self) -> (usize, Option<usize>) {
+ (usize::MAX, None)
+ }
+}
+
+/// Creates an infinite stream from a closure that generates items.
+///
+/// # Examples
+///
+/// ```
+/// use futures_lite::stream::{self, StreamExt};
+///
+/// # spin_on::spin_on(async {
+/// let mut s = stream::repeat_with(|| 7);
+///
+/// assert_eq!(s.next().await, Some(7));
+/// assert_eq!(s.next().await, Some(7));
+/// # })
+/// ```
+pub fn repeat_with<T, F>(repeater: F) -> RepeatWith<F>
+where
+ F: FnMut() -> T,
+{
+ RepeatWith { f: repeater }
+}
+
+/// Stream for the [`repeat_with()`] function.
+#[derive(Clone, Debug)]
+#[must_use = "streams do nothing unless polled"]
+pub struct RepeatWith<F> {
+ f: F,
+}
+
+impl<F> Unpin for RepeatWith<F> {}
+
+impl<T, F> Stream for RepeatWith<F>
+where
+ F: FnMut() -> T,
+{
+ type Item = T;
+
+ fn poll_next(mut self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+ let item = (&mut self.f)();
+ Poll::Ready(Some(item))
+ }
+
+ fn size_hint(&self) -> (usize, Option<usize>) {
+ (usize::MAX, None)
+ }
+}
+
+/// Creates a stream from a seed value and an async closure operating on it.
+///
+/// # Examples
+///
+/// ```
+/// use futures_lite::stream::{self, StreamExt};
+///
+/// # spin_on::spin_on(async {
+/// let s = stream::unfold(0, |mut n| async move {
+/// if n < 2 {
+/// let m = n + 1;
+/// Some((n, m))
+/// } else {
+/// None
+/// }
+/// });
+///
+/// let v: Vec<i32> = s.collect().await;
+/// assert_eq!(v, [0, 1]);
+/// # })
+/// ```
+pub fn unfold<T, F, Fut, Item>(seed: T, f: F) -> Unfold<T, F, Fut>
+where
+ F: FnMut(T) -> Fut,
+ Fut: Future<Output = Option<(Item, T)>>,
+{
+ Unfold {
+ f,
+ state: Some(seed),
+ fut: None,
+ }
+}
+
+pin_project! {
+ /// Stream for the [`unfold()`] function.
+ #[derive(Clone)]
+ #[must_use = "streams do nothing unless polled"]
+ pub struct Unfold<T, F, Fut> {
+ f: F,
+ state: Option<T>,
+ #[pin]
+ fut: Option<Fut>,
+ }
+}
+
+impl<T, F, Fut> fmt::Debug for Unfold<T, F, Fut>
+where
+ T: fmt::Debug,
+ Fut: fmt::Debug,
+{
+ fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+ f.debug_struct("Unfold")
+ .field("state", &self.state)
+ .field("fut", &self.fut)
+ .finish()
+ }
+}
+
+impl<T, F, Fut, Item> Stream for Unfold<T, F, Fut>
+where
+ F: FnMut(T) -> Fut,
+ Fut: Future<Output = Option<(Item, T)>>,
+{
+ type Item = Item;
+
+ fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+ let mut this = self.project();
+
+ if let Some(state) = this.state.take() {
+ this.fut.set(Some((this.f)(state)));
+ }
+
+ let step = ready!(this
+ .fut
+ .as_mut()
+ .as_pin_mut()
+ .expect("`Unfold` must not be polled after it returned `Poll::Ready(None)`")
+ .poll(cx));
+ this.fut.set(None);
+
+ if let Some((item, next_state)) = step {
+ *this.state = Some(next_state);
+ Poll::Ready(Some(item))
+ } else {
+ Poll::Ready(None)
+ }
+ }
+}
+
+/// Creates a stream from a seed value and a fallible async closure operating on it.
+///
+/// # Examples
+///
+/// ```
+/// use futures_lite::stream::{self, StreamExt};
+///
+/// # spin_on::spin_on(async {
+/// let s = stream::try_unfold(0, |mut n| async move {
+/// if n < 2 {
+/// let m = n + 1;
+/// Ok(Some((n, m)))
+/// } else {
+/// std::io::Result::Ok(None)
+/// }
+/// });
+///
+/// let v: Vec<i32> = s.try_collect().await?;
+/// assert_eq!(v, [0, 1]);
+/// # std::io::Result::Ok(()) });
+/// ```
+pub fn try_unfold<T, E, F, Fut, Item>(init: T, f: F) -> TryUnfold<T, F, Fut>
+where
+ F: FnMut(T) -> Fut,
+ Fut: Future<Output = Result<Option<(Item, T)>, E>>,
+{
+ TryUnfold {
+ f,
+ state: Some(init),
+ fut: None,
+ }
+}
+
+pin_project! {
+ /// Stream for the [`try_unfold()`] function.
+ #[derive(Clone)]
+ #[must_use = "streams do nothing unless polled"]
+ pub struct TryUnfold<T, F, Fut> {
+ f: F,
+ state: Option<T>,
+ #[pin]
+ fut: Option<Fut>,
+ }
+}
+
+impl<T, F, Fut> fmt::Debug for TryUnfold<T, F, Fut>
+where
+ T: fmt::Debug,
+ Fut: fmt::Debug,
+{
+ fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+ f.debug_struct("TryUnfold")
+ .field("state", &self.state)
+ .field("fut", &self.fut)
+ .finish()
+ }
+}
+
+impl<T, E, F, Fut, Item> Stream for TryUnfold<T, F, Fut>
+where
+ F: FnMut(T) -> Fut,
+ Fut: Future<Output = Result<Option<(Item, T)>, E>>,
+{
+ type Item = Result<Item, E>;
+
+ fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+ let mut this = self.project();
+
+ if let Some(state) = this.state.take() {
+ this.fut.set(Some((this.f)(state)));
+ }
+
+ match this.fut.as_mut().as_pin_mut() {
+ None => {
+ // The future previously errored
+ Poll::Ready(None)
+ }
+ Some(future) => {
+ let step = ready!(future.poll(cx));
+ this.fut.set(None);
+
+ match step {
+ Ok(Some((item, next_state))) => {
+ *this.state = Some(next_state);
+ Poll::Ready(Some(Ok(item)))
+ }
+ Ok(None) => Poll::Ready(None),
+ Err(e) => Poll::Ready(Some(Err(e))),
+ }
+ }
+ }
+ }
+}
+
+/// Creates a stream that invokes the given future as its first item, and then
+/// produces no more items.
+///
+/// # Example
+///
+/// ```
+/// use futures_lite::{stream, prelude::*};
+///
+/// # spin_on::spin_on(async {
+/// let mut stream = Box::pin(stream::once_future(async { 1 }));
+/// assert_eq!(stream.next().await, Some(1));
+/// assert_eq!(stream.next().await, None);
+/// # });
+/// ```
+pub fn once_future<F: Future>(future: F) -> OnceFuture<F> {
+ OnceFuture {
+ future: Some(future),
+ }
+}
+
+pin_project! {
+ /// Stream for the [`once_future()`] function.
+ #[derive(Debug)]
+ #[must_use = "futures do nothing unless you `.await` or poll them"]
+ pub struct OnceFuture<F> {
+ #[pin]
+ future: Option<F>,
+ }
+}
+
+impl<F: Future> Stream for OnceFuture<F> {
+ type Item = F::Output;
+
+ fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+ let mut this = self.project();
+
+ match this.future.as_mut().as_pin_mut().map(|f| f.poll(cx)) {
+ Some(Poll::Ready(t)) => {
+ this.future.set(None);
+ Poll::Ready(Some(t))
+ }
+ Some(Poll::Pending) => Poll::Pending,
+ None => Poll::Ready(None),
+ }
+ }
+}
+
+/// Take elements from this stream until the provided future resolves.
+///
+/// This function will take elements from the stream until the provided
+/// stopping future `fut` resolves. Once the `fut` future becomes ready,
+/// this stream combinator will always return that the stream is done.
+///
+/// The stopping future may return any type. Once the stream is stopped
+/// the result of the stopping future may be accessed with `StopAfterFuture::take_result()`.
+/// The stream may also be resumed with `StopAfterFuture::take_future()`.
+/// See the documentation of [`StopAfterFuture`] for more information.
+///
+/// ```
+/// use futures_lite::stream::{self, StreamExt, stop_after_future};
+/// use futures_lite::future;
+/// use std::task::Poll;
+///
+/// let stream = stream::iter(1..=10);
+///
+/// # spin_on::spin_on(async {
+/// let mut i = 0;
+/// let stop_fut = future::poll_fn(|_cx| {
+/// i += 1;
+/// if i <= 5 {
+/// Poll::Pending
+/// } else {
+/// Poll::Ready(())
+/// }
+/// });
+///
+/// let stream = stop_after_future(stream, stop_fut);
+///
+/// assert_eq!(vec![1, 2, 3, 4, 5], stream.collect::<Vec<_>>().await);
+/// # });
+pub fn stop_after_future<S, F>(stream: S, future: F) -> StopAfterFuture<S, F>
+where
+ S: Sized + Stream,
+ F: Future,
+{
+ StopAfterFuture {
+ stream,
+ fut: Some(future),
+ fut_result: None,
+ free: false,
+ }
+}
+
+pin_project! {
+ /// Stream for the [`stop_after_future()`] function.
+ #[derive(Clone, Debug)]
+ #[must_use = "streams do nothing unless polled"]
+ pub struct StopAfterFuture<S: Stream, Fut: Future> {
+ #[pin]
+ stream: S,
+ // Contains the inner Future on start and None once the inner Future is resolved
+ // or taken out by the user.
+ #[pin]
+ fut: Option<Fut>,
+ // Contains fut's return value once fut is resolved
+ fut_result: Option<Fut::Output>,
+ // Whether the future was taken out by the user.
+ free: bool,
+ }
+}
+
+impl<St, Fut> StopAfterFuture<St, Fut>
+where
+ St: Stream,
+ Fut: Future,
+{
+ /// Extract the stopping future out of the combinator.
+ ///
+ /// The future is returned only if it isn't resolved yet, ie. if the stream isn't stopped yet.
+ /// Taking out the future means the combinator will be yielding
+ /// elements from the wrapped stream without ever stopping it.
+ pub fn take_future(&mut self) -> Option<Fut> {
+ if self.fut.is_some() {
+ self.free = true;
+ }
+
+ self.fut.take()
+ }
+
+ /// Once the stopping future is resolved, this method can be used
+ /// to extract the value returned by the stopping future.
+ ///
+ /// This may be used to retrieve arbitrary data from the stopping
+ /// future, for example a reason why the stream was stopped.
+ ///
+ /// This method will return `None` if the future isn't resolved yet,
+ /// or if the result was already taken out.
+ ///
+ /// # Examples
+ ///
+ /// ```
+ /// # spin_on::spin_on(async {
+ /// use futures_lite::stream::{self, StreamExt, stop_after_future};
+ /// use futures_lite::future;
+ /// use std::task::Poll;
+ ///
+ /// let stream = stream::iter(1..=10);
+ ///
+ /// let mut i = 0;
+ /// let stop_fut = future::poll_fn(|_cx| {
+ /// i += 1;
+ /// if i <= 5 {
+ /// Poll::Pending
+ /// } else {
+ /// Poll::Ready("reason")
+ /// }
+ /// });
+ ///
+ /// let mut stream = stop_after_future(stream, stop_fut);
+ /// let _ = (&mut stream).collect::<Vec<_>>().await;
+ ///
+ /// let result = stream.take_result().unwrap();
+ /// assert_eq!(result, "reason");
+ /// # });
+ /// ```
+ pub fn take_result(&mut self) -> Option<Fut::Output> {
+ self.fut_result.take()
+ }
+
+ /// Whether the stream was stopped yet by the stopping future
+ /// being resolved.
+ pub fn is_stopped(&self) -> bool {
+ !self.free && self.fut.is_none()
+ }
+}
+
+impl<St, Fut> Stream for StopAfterFuture<St, Fut>
+where
+ St: Stream,
+ Fut: Future,
+{
+ type Item = St::Item;
+
+ fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<St::Item>> {
+ let mut this = self.project();
+
+ if let Some(f) = this.fut.as_mut().as_pin_mut() {
+ if let Poll::Ready(result) = f.poll(cx) {
+ this.fut.set(None);
+ *this.fut_result = Some(result);
+ }
+ }
+
+ if !*this.free && this.fut.is_none() {
+ // Future resolved, inner stream stopped
+ Poll::Ready(None)
+ } else {
+ // Future either not resolved yet or taken out by the user
+ let item = ready!(this.stream.poll_next(cx));
+ if item.is_none() {
+ this.fut.set(None);
+ }
+ Poll::Ready(item)
+ }
+ }
+
+ fn size_hint(&self) -> (usize, Option<usize>) {
+ if self.is_stopped() {
+ return (0, Some(0));
+ }
+
+ // Original stream can be truncated at any moment, so the lower bound isn't reliable.
+ let (_, upper_bound) = self.stream.size_hint();
+ (0, upper_bound)
+ }
+}
+
+/// Extension trait for [`Stream`].
+pub trait StreamExt: Stream {
+ /// A convenience for calling [`Stream::poll_next()`] on `!`[`Unpin`] types.
+ fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll<Option<Self::Item>>
+ where
+ Self: Unpin,
+ {
+ Stream::poll_next(Pin::new(self), cx)
+ }
+
+ /// Retrieves the next item in the stream.
+ ///
+ /// Returns [`None`] when iteration is finished. Stream implementations may choose to or not to
+ /// resume iteration after that.
+ ///
+ /// # Examples
+ ///
+ /// ```
+ /// use futures_lite::stream::{self, StreamExt};
+ ///
+ /// # spin_on::spin_on(async {
+ /// let mut s = stream::iter(1..=3);
+ ///
+ /// assert_eq!(s.next().await, Some(1));
+ /// assert_eq!(s.next().await, Some(2));
+ /// assert_eq!(s.next().await, Some(3));
+ /// assert_eq!(s.next().await, None);
+ /// # });
+ /// ```
+ fn next(&mut self) -> NextFuture<'_, Self>
+ where
+ Self: Unpin,
+ {
+ NextFuture { stream: self }
+ }
+
+ /// Retrieves the next item in the stream.
+ ///
+ /// This is similar to the [`next()`][`StreamExt::next()`] method, but returns
+ /// `Result<Option<T>, E>` rather than `Option<Result<T, E>>`.
+ ///
+ /// Note that `s.try_next().await` is equivalent to `s.next().await.transpose()`.
+ ///
+ /// # Examples
+ ///
+ /// ```
+ /// use futures_lite::stream::{self, StreamExt};
+ ///
+ /// # spin_on::spin_on(async {
+ /// let mut s = stream::iter(vec![Ok(1), Ok(2), Err("error")]);
+ ///
+ /// assert_eq!(s.try_next().await, Ok(Some(1)));
+ /// assert_eq!(s.try_next().await, Ok(Some(2)));
+ /// assert_eq!(s.try_next().await, Err("error"));
+ /// assert_eq!(s.try_next().await, Ok(None));
+ /// # });
+ /// ```
+ fn try_next<T, E>(&mut self) -> TryNextFuture<'_, Self>
+ where
+ Self: Stream<Item = Result<T, E>> + Unpin,
+ {
+ TryNextFuture { stream: self }
+ }
+
+ /// Counts the number of items in the stream.
+ ///
+ /// # Examples
+ ///
+ /// ```
+ /// use futures_lite::stream::{self, StreamExt};
+ ///
+ /// # spin_on::spin_on(async {
+ /// let s1 = stream::iter(vec![0]);
+ /// let s2 = stream::iter(vec![1, 2, 3]);
+ ///
+ /// assert_eq!(s1.count().await, 1);
+ /// assert_eq!(s2.count().await, 3);
+ /// # });
+ /// ```
+ fn count(self) -> CountFuture<Self>
+ where
+ Self: Sized,
+ {
+ CountFuture {
+ stream: self,
+ count: 0,
+ }
+ }
+
+ /// Maps items of the stream to new values using a closure.
+ ///
+ /// # Examples
+ ///
+ /// ```
+ /// use futures_lite::stream::{self, StreamExt};
+ ///
+ /// # spin_on::spin_on(async {
+ /// let s = stream::iter(vec![1, 2, 3]);
+ /// let mut s = s.map(|x| 2 * x);
+ ///
+ /// assert_eq!(s.next().await, Some(2));
+ /// assert_eq!(s.next().await, Some(4));
+ /// assert_eq!(s.next().await, Some(6));
+ /// assert_eq!(s.next().await, None);
+ /// # });
+ /// ```
+ fn map<T, F>(self, f: F) -> Map<Self, F>
+ where
+ Self: Sized,
+ F: FnMut(Self::Item) -> T,
+ {
+ Map { stream: self, f }
+ }
+
+ /// Maps items to streams and then concatenates them.
+ ///
+ /// # Examples
+ ///
+ /// ```
+ /// use futures_lite::stream::{self, StreamExt};
+ ///
+ /// # spin_on::spin_on(async {
+ /// let words = stream::iter(vec!["one", "two"]);
+ ///
+ /// let s: String = words
+ /// .flat_map(|s| stream::iter(s.chars()))
+ /// .collect()
+ /// .await;
+ ///
+ /// assert_eq!(s, "onetwo");
+ /// # });
+ /// ```
+ fn flat_map<U, F>(self, f: F) -> FlatMap<Self, U, F>
+ where
+ Self: Sized,
+ U: Stream,
+ F: FnMut(Self::Item) -> U,
+ {
+ FlatMap {
+ stream: self.map(f),
+ inner_stream: None,
+ }
+ }
+
+ /// Concatenates inner streams.
+ ///
+ /// # Examples
+ ///
+ /// ```
+ /// use futures_lite::stream::{self, StreamExt};
+ ///
+ /// # spin_on::spin_on(async {
+ /// let s1 = stream::iter(vec![1, 2, 3]);
+ /// let s2 = stream::iter(vec![4, 5]);
+ ///
+ /// let s = stream::iter(vec![s1, s2]);
+ /// let v: Vec<_> = s.flatten().collect().await;
+ /// assert_eq!(v, [1, 2, 3, 4, 5]);
+ /// # });
+ /// ```
+ fn flatten(self) -> Flatten<Self>
+ where
+ Self: Sized,
+ Self::Item: Stream,
+ {
+ Flatten {
+ stream: self,
+ inner_stream: None,
+ }
+ }
+
+ /// Maps items of the stream to new values using an async closure.
+ ///
+ /// # Examples
+ ///
+ /// ```
+ /// use futures_lite::pin;
+ /// use futures_lite::stream::{self, StreamExt};
+ ///
+ /// # spin_on::spin_on(async {
+ /// let s = stream::iter(vec![1, 2, 3]);
+ /// let mut s = s.then(|x| async move { 2 * x });
+ ///
+ /// pin!(s);
+ /// assert_eq!(s.next().await, Some(2));
+ /// assert_eq!(s.next().await, Some(4));
+ /// assert_eq!(s.next().await, Some(6));
+ /// assert_eq!(s.next().await, None);
+ /// # });
+ /// ```
+ fn then<F, Fut>(self, f: F) -> Then<Self, F, Fut>
+ where
+ Self: Sized,
+ F: FnMut(Self::Item) -> Fut,
+ Fut: Future,
+ {
+ Then {
+ stream: self,
+ future: None,
+ f,
+ }
+ }
+
+ /// Keeps items of the stream for which `predicate` returns `true`.
+ ///
+ /// # Examples
+ ///
+ /// ```
+ /// use futures_lite::stream::{self, StreamExt};
+ ///
+ /// # spin_on::spin_on(async {
+ /// let s = stream::iter(vec![1, 2, 3, 4]);
+ /// let mut s = s.filter(|i| i % 2 == 0);
+ ///
+ /// assert_eq!(s.next().await, Some(2));
+ /// assert_eq!(s.next().await, Some(4));
+ /// assert_eq!(s.next().await, None);
+ /// # });
+ /// ```
+ fn filter<P>(self, predicate: P) -> Filter<Self, P>
+ where
+ Self: Sized,
+ P: FnMut(&Self::Item) -> bool,
+ {
+ Filter {
+ stream: self,
+ predicate,
+ }
+ }
+
+ /// Filters and maps items of the stream using a closure.
+ ///
+ /// # Examples
+ ///
+ /// ```
+ /// use futures_lite::stream::{self, StreamExt};
+ ///
+ /// # spin_on::spin_on(async {
+ /// let s = stream::iter(vec!["1", "lol", "3", "NaN", "5"]);
+ /// let mut s = s.filter_map(|a| a.parse::<u32>().ok());
+ ///
+ /// assert_eq!(s.next().await, Some(1));
+ /// assert_eq!(s.next().await, Some(3));
+ /// assert_eq!(s.next().await, Some(5));
+ /// assert_eq!(s.next().await, None);
+ /// # });
+ /// ```
+ fn filter_map<T, F>(self, f: F) -> FilterMap<Self, F>
+ where
+ Self: Sized,
+ F: FnMut(Self::Item) -> Option<T>,
+ {
+ FilterMap { stream: self, f }
+ }
+
+ /// Takes only the first `n` items of the stream.
+ ///
+ /// # Examples
+ ///
+ /// ```
+ /// use futures_lite::stream::{self, StreamExt};
+ ///
+ /// # spin_on::spin_on(async {
+ /// let mut s = stream::repeat(7).take(2);
+ ///
+ /// assert_eq!(s.next().await, Some(7));
+ /// assert_eq!(s.next().await, Some(7));
+ /// assert_eq!(s.next().await, None);
+ /// # });
+ /// ```
+ fn take(self, n: usize) -> Take<Self>
+ where
+ Self: Sized,
+ {
+ Take { stream: self, n }
+ }
+
+ /// Takes items while `predicate` returns `true`.
+ ///
+ /// # Examples
+ ///
+ /// ```
+ /// use futures_lite::stream::{self, StreamExt};
+ ///
+ /// # spin_on::spin_on(async {
+ /// let s = stream::iter(vec![1, 2, 3, 4]);
+ /// let mut s = s.take_while(|x| *x < 3);
+ ///
+ /// assert_eq!(s.next().await, Some(1));
+ /// assert_eq!(s.next().await, Some(2));
+ /// assert_eq!(s.next().await, None);
+ /// # });
+ /// ```
+ fn take_while<P>(self, predicate: P) -> TakeWhile<Self, P>
+ where
+ Self: Sized,
+ P: FnMut(&Self::Item) -> bool,
+ {
+ TakeWhile {
+ stream: self,
+ predicate,
+ }
+ }
+
+ /// Maps items while `predicate` returns [`Some`].
+ ///
+ /// This stream is not fused. After the predicate returns [`None`] the stream still
+ /// contains remaining items that can be obtained by subsequent `next` calls.
+ /// You can [`fuse`](StreamExt::fuse) the stream if this behavior is undesirable.
+ ///
+ /// # Examples
+ ///
+ /// ```
+ /// use futures_lite::stream::{self, StreamExt};
+ ///
+ /// # spin_on::spin_on(async {
+ /// let s = stream::iter(vec![1, 2, 0, 3]);
+ /// let mut s = s.map_while(|x: u32| x.checked_sub(1));
+ ///
+ /// assert_eq!(s.next().await, Some(0));
+ /// assert_eq!(s.next().await, Some(1));
+ /// assert_eq!(s.next().await, None);
+ ///
+ /// // Continue to iterate the stream.
+ /// assert_eq!(s.next().await, Some(2));
+ /// assert_eq!(s.next().await, None);
+ /// # });
+ /// ```
+ fn map_while<B, P>(self, predicate: P) -> MapWhile<Self, P>
+ where
+ Self: Sized,
+ P: FnMut(Self::Item) -> Option<B>,
+ {
+ MapWhile {
+ stream: self,
+ predicate,
+ }
+ }
+
+ /// Skips the first `n` items of the stream.
+ ///
+ /// # Examples
+ ///
+ /// ```
+ /// use futures_lite::stream::{self, StreamExt};
+ ///
+ /// # spin_on::spin_on(async {
+ /// let s = stream::iter(vec![1, 2, 3]);
+ /// let mut s = s.skip(2);
+ ///
+ /// assert_eq!(s.next().await, Some(3));
+ /// assert_eq!(s.next().await, None);
+ /// # });
+ /// ```
+ fn skip(self, n: usize) -> Skip<Self>
+ where
+ Self: Sized,
+ {
+ Skip { stream: self, n }
+ }
+
+ /// Skips items while `predicate` returns `true`.
+ ///
+ /// # Examples
+ ///
+ /// ```
+ /// use futures_lite::stream::{self, StreamExt};
+ ///
+ /// # spin_on::spin_on(async {
+ /// let s = stream::iter(vec![-1i32, 0, 1]);
+ /// let mut s = s.skip_while(|x| x.is_negative());
+ ///
+ /// assert_eq!(s.next().await, Some(0));
+ /// assert_eq!(s.next().await, Some(1));
+ /// assert_eq!(s.next().await, None);
+ /// # });
+ /// ```
+ fn skip_while<P>(self, predicate: P) -> SkipWhile<Self, P>
+ where
+ Self: Sized,
+ P: FnMut(&Self::Item) -> bool,
+ {
+ SkipWhile {
+ stream: self,
+ predicate: Some(predicate),
+ }
+ }
+
+ /// Yields every `step`th item.
+ ///
+ /// # Panics
+ ///
+ /// This method will panic if the `step` is 0.
+ ///
+ /// # Examples
+ ///
+ /// ```
+ /// use futures_lite::stream::{self, StreamExt};
+ ///
+ /// # spin_on::spin_on(async {
+ /// let s = stream::iter(vec![0, 1, 2, 3, 4]);
+ /// let mut s = s.step_by(2);
+ ///
+ /// assert_eq!(s.next().await, Some(0));
+ /// assert_eq!(s.next().await, Some(2));
+ /// assert_eq!(s.next().await, Some(4));
+ /// assert_eq!(s.next().await, None);
+ /// # });
+ /// ```
+ fn step_by(self, step: usize) -> StepBy<Self>
+ where
+ Self: Sized,
+ {
+ assert!(step > 0, "`step` must be greater than zero");
+ StepBy {
+ stream: self,
+ step,
+ i: 0,
+ }
+ }
+
+ /// Appends another stream to the end of this one.
+ ///
+ /// # Examples
+ ///
+ /// ```
+ /// use futures_lite::stream::{self, StreamExt};
+ ///
+ /// # spin_on::spin_on(async {
+ /// let s1 = stream::iter(vec![1, 2]);
+ /// let s2 = stream::iter(vec![7, 8]);
+ /// let mut s = s1.chain(s2);
+ ///
+ /// assert_eq!(s.next().await, Some(1));
+ /// assert_eq!(s.next().await, Some(2));
+ /// assert_eq!(s.next().await, Some(7));
+ /// assert_eq!(s.next().await, Some(8));
+ /// assert_eq!(s.next().await, None);
+ /// # });
+ /// ```
+ fn chain<U>(self, other: U) -> Chain<Self, U>
+ where
+ Self: Sized,
+ U: Stream<Item = Self::Item> + Sized,
+ {
+ Chain {
+ first: self.fuse(),
+ second: other.fuse(),
+ }
+ }
+
+ /// Clones all items.
+ ///
+ /// # Examples
+ ///
+ /// ```
+ /// use futures_lite::stream::{self, StreamExt};
+ ///
+ /// # spin_on::spin_on(async {
+ /// let s = stream::iter(vec![&1, &2]);
+ /// let mut s = s.cloned();
+ ///
+ /// assert_eq!(s.next().await, Some(1));
+ /// assert_eq!(s.next().await, Some(2));
+ /// assert_eq!(s.next().await, None);
+ /// # });
+ /// ```
+ fn cloned<'a, T>(self) -> Cloned<Self>
+ where
+ Self: Stream<Item = &'a T> + Sized,
+ T: Clone + 'a,
+ {
+ Cloned { stream: self }
+ }
+
+ /// Copies all items.
+ ///
+ /// # Examples
+ ///
+ /// ```
+ /// use futures_lite::stream::{self, StreamExt};
+ ///
+ /// # spin_on::spin_on(async {
+ /// let s = stream::iter(vec![&1, &2]);
+ /// let mut s = s.copied();
+ ///
+ /// assert_eq!(s.next().await, Some(1));
+ /// assert_eq!(s.next().await, Some(2));
+ /// assert_eq!(s.next().await, None);
+ /// # });
+ /// ```
+ fn copied<'a, T>(self) -> Copied<Self>
+ where
+ Self: Stream<Item = &'a T> + Sized,
+ T: Copy + 'a,
+ {
+ Copied { stream: self }
+ }
+
+ /// Collects all items in the stream into a collection.
+ ///
+ /// # Examples
+ ///
+ /// ```
+ /// use futures_lite::stream::{self, StreamExt};
+ ///
+ /// # spin_on::spin_on(async {
+ /// let mut s = stream::iter(1..=3);
+ ///
+ /// let items: Vec<_> = s.collect().await;
+ /// assert_eq!(items, [1, 2, 3]);
+ /// # });
+ /// ```
+ fn collect<C>(self) -> CollectFuture<Self, C>
+ where
+ Self: Sized,
+ C: Default + Extend<Self::Item>,
+ {
+ CollectFuture {
+ stream: self,
+ collection: Default::default(),
+ }
+ }
+
+ /// Collects all items in the fallible stream into a collection.
+ ///
+ /// ```
+ /// use futures_lite::stream::{self, StreamExt};
+ ///
+ /// # spin_on::spin_on(async {
+ /// let s = stream::iter(vec![Ok(1), Err(2), Ok(3)]);
+ /// let res: Result<Vec<i32>, i32> = s.try_collect().await;
+ /// assert_eq!(res, Err(2));
+ ///
+ /// let s = stream::iter(vec![Ok(1), Ok(2), Ok(3)]);
+ /// let res: Result<Vec<i32>, i32> = s.try_collect().await;
+ /// assert_eq!(res, Ok(vec![1, 2, 3]));
+ /// # })
+ /// ```
+ fn try_collect<T, E, C>(self) -> TryCollectFuture<Self, C>
+ where
+ Self: Stream<Item = Result<T, E>> + Sized,
+ C: Default + Extend<T>,
+ {
+ TryCollectFuture {
+ stream: self,
+ items: Default::default(),
+ }
+ }
+
+ /// Partitions items into those for which `predicate` is `true` and those for which it is
+ /// `false`, and then collects them into two collections.
+ ///
+ /// # Examples
+ ///
+ /// ```
+ /// use futures_lite::stream::{self, StreamExt};
+ ///
+ /// # spin_on::spin_on(async {
+ /// let s = stream::iter(vec![1, 2, 3]);
+ /// let (even, odd): (Vec<_>, Vec<_>) = s.partition(|&n| n % 2 == 0).await;
+ ///
+ /// assert_eq!(even, &[2]);
+ /// assert_eq!(odd, &[1, 3]);
+ /// # })
+ /// ```
+ fn partition<B, P>(self, predicate: P) -> PartitionFuture<Self, P, B>
+ where
+ Self: Sized,
+ B: Default + Extend<Self::Item>,
+ P: FnMut(&Self::Item) -> bool,
+ {
+ PartitionFuture {
+ stream: self,
+ predicate,
+ Why this scored 18/100
Community notes
Notes can correct, qualify, or add evidence to the AI analysis. Every note shown here has been validated by a human moderator.
The AI analysis stands alone for now. Submit a note if you can add evidence or important context.