Add EndpointRead and EndpointWrite

Cargo.toml

@@ -35,10 +35,11 @@ io-kit-sys = "0.4.0"
 [target.'cfg(any(target_os="linux", target_os="android", target_os="windows", target_os="macos"))'.dependencies]
 blocking = { version = "1.6.1", optional = true }
 tokio = { version = "1", optional = true, features = ["rt"] }
+futures-io = { version = "0.3", optional = true }
 
 [features]
 # Use the `blocking` crate for making blocking IO async
-smol = ["dep:blocking"]
+smol = ["dep:blocking", "dep:futures-io"]
 
 # Use `tokio`'s IO threadpool for making blocking IO async
 tokio = ["dep:tokio"]

examples/bulk_io.rs

@@ -0,0 +1,87 @@
+use std::{
+    io::{BufRead, Read, Write},
+    time::Duration,
+};
+
+use nusb::{
+    transfer::{Bulk, In, Out},
+    MaybeFuture,
+};
+
+fn main() {
+    env_logger::init();
+    let di = nusb::list_devices()
+        .wait()
+        .unwrap()
+        .find(|d| d.vendor_id() == 0x59e3 && d.product_id() == 0x00aa)
+        .expect("device should be connected");
+
+    println!("Device info: {di:?}");
+
+    let device = di.open().wait().unwrap();
+
+    let main_interface = device.claim_interface(0).wait().unwrap();
+
+    let mut writer = main_interface
+        .endpoint::<Bulk, Out>(0x03)
+        .unwrap()
+        .writer(128)
+        .with_num_transfers(8);
+
+    let mut reader = main_interface
+        .endpoint::<Bulk, In>(0x83)
+        .unwrap()
+        .reader(128);
+
+    writer.write_all(&[1; 16]).unwrap();
+    writer.write_all(&[2; 256]).unwrap();
+    writer.flush().unwrap();
+    writer.write_all(&[3; 64]).unwrap();
+    writer.flush_end().unwrap();
+
+    let mut buf = [0; 16];
+    reader.read_exact(&mut buf).unwrap();
+
+    let mut buf = [0; 64];
+    reader.read_exact(&mut buf).unwrap();
+
+    dbg!(reader.fill_buf().unwrap().len());
+
+    let echo_interface = device.claim_interface(1).wait().unwrap();
+    echo_interface.set_alt_setting(1).wait().unwrap();
+
+    let mut writer = echo_interface
+        .endpoint::<Bulk, Out>(0x01)
+        .unwrap()
+        .writer(64)
+        .with_num_transfers(1);
+    let mut reader = echo_interface
+        .endpoint::<Bulk, In>(0x81)
+        .unwrap()
+        .reader(64)
+        .with_num_transfers(8)
+        .with_read_timeout(Duration::from_millis(100));
+
+    assert_eq!(
+        reader.fill_buf().unwrap_err().kind(),
+        std::io::ErrorKind::TimedOut
+    );
+
+    let mut pkt_reader = reader.until_short_packet();
+
+    writer.write_all(&[1; 16]).unwrap();
+    writer.flush_end().unwrap();
+
+    writer.write_all(&[2; 128]).unwrap();
+    writer.flush_end().unwrap();
+
+    let mut v = Vec::new();
+    pkt_reader.read_to_end(&mut v).unwrap();
+    assert_eq!(&v[..], &[1; 16]);
+    pkt_reader.consume_end().unwrap();
+
+    let mut v = Vec::new();
+    pkt_reader.read_to_end(&mut v).unwrap();
+    assert_eq!(&v[..], &[2; 128]);
+    pkt_reader.consume_end().unwrap();
+}

src/device.rs

@@ -3,10 +3,11 @@ use crate::{
         decode_string_descriptor, validate_string_descriptor, ConfigurationDescriptor,
         DeviceDescriptor, InterfaceDescriptor, DESCRIPTOR_TYPE_STRING,
     },
+    io::{EndpointRead, EndpointWrite},
     platform,
     transfer::{
         Buffer, BulkOrInterrupt, Completion, ControlIn, ControlOut, Direction, EndpointDirection,
-        EndpointType, TransferError,
+        EndpointType, In, Out, TransferError,
     },
     ActiveConfigurationError, DeviceInfo, Error, ErrorKind, GetDescriptorError, MaybeFuture, Speed,
 };
@@ -544,9 +545,9 @@ impl Debug for Interface {
 /// a non-blocking operation that adds the operation to the queue. Completed
 /// transfers can be popped from the queue synchronously or asynchronously.
 ///
-/// This separation of submission and completion makes the API cancel-safe,
-/// and makes it easy to submit multiple transfers at once, regardless of
-/// whether you are using asynchronous or blocking waits.
+/// This separation of submission and completion makes the API cancel-safe, and
+/// makes it easy to submit multiple transfers at once, regardless of whether
+/// you are using asynchronous or blocking waits.
 ///
 /// To maximize throughput and minimize latency when streaming data, the host
 /// controller needs to attempt a transfer in every possible frame. That
@@ -555,6 +556,11 @@ impl Debug for Interface {
 /// complete.
 ///
 /// When the `Endpoint` is dropped, any pending transfers are cancelled.
+///
+/// This type provides a low-level API letting you manage transfers directly.
+/// For a higher-level buffered API implementing [`std::io::Read`] and
+/// [`std::io::Write`], use the adapters from [`nusb::io`][`crate::io`]. See
+/// [`Self::reader`] and [`Self::writer`].
 pub struct Endpoint<EpType, Dir> {
     backend: platform::Endpoint,
     ep_type: PhantomData<EpType>,
@@ -592,6 +598,28 @@ impl<EpType: EndpointType, Dir: EndpointDirection> Endpoint<EpType, Dir> {
     }
 }
 
+impl<EpType: BulkOrInterrupt> Endpoint<EpType, Out> {
+    /// Create an [`EndpointWrite`] wrapping the given endpoint to provide a
+    /// high-level buffered API implementing [`std::io::Write`] and async
+    /// equivalents.
+    ///
+    /// See [`EndpointWrite::new`][`crate::io::EndpointWrite::new`] for details.
+    pub fn writer(self, buffer_size: usize) -> EndpointWrite<EpType> {
+        EndpointWrite::new(self, buffer_size)
+    }
+}
+
+impl<EpType: BulkOrInterrupt> Endpoint<EpType, In> {
+    /// Create an [`EndpointRead`] wrapping the given endpoint to provide a
+    /// high-level buffered API implementing [`std::io::Read`] and  async
+    /// equivalents.
+    ///
+    /// See [`EndpointRead::new`][`crate::io::EndpointRead::new`] for details.
+    pub fn reader(self, buffer_size: usize) -> EndpointRead<EpType> {
+        EndpointRead::new(self, buffer_size)
+    }
+}
+
 /// Methods for Bulk and Interrupt endpoints.
 impl<EpType: BulkOrInterrupt, Dir: EndpointDirection> Endpoint<EpType, Dir> {
     /// Allocate a buffer for use on this endpoint, zero-copy if possible.

src/io/mod.rs

@@ -0,0 +1,41 @@
+//! Adapters for using [`std::io`] traits and their async equivalents with Bulk
+//! and Interrupt endpoints.
+//!
+//! These types wrap an [`Endpoint`](crate::Endpoint) and manage transfers to
+//! provide a higher-level buffered API.
+//!
+//! ## Examples
+//!
+//! ### Request-response
+//!
+//! ```no_run
+//! use std::{io::{Read, Write}, time::Duration};
+//! use nusb::{self, MaybeFuture, transfer::{Bulk, In, Out}};
+//! let device_info = nusb::list_devices().wait().unwrap()
+//!     .find(|dev| dev.vendor_id() == 0xAAAA && dev.product_id() == 0xBBBB)
+//!     .expect("device not connected");
+//!
+//! let device = device_info.open().wait().expect("failed to open device");
+//! let interface = device.claim_interface(0).wait().expect("failed to claim interface");
+//!
+//! let mut tx = interface.endpoint::<Bulk, Out>(0x01).unwrap()
+//!     .writer(256)
+//!     .with_num_transfers(4);
+//! let mut rx = interface.endpoint::<Bulk, In>(0x81).unwrap()
+//!     .reader(256)
+//!     .with_num_transfers(4)
+//!     .with_read_timeout(Duration::from_secs(1));
+//!
+//! tx.write_all(&[0x01, 0x02, 0x03]).unwrap();
+//! tx.flush_end().unwrap();
+//!
+//! let mut rx_pkt = rx.until_short_packet();
+//! let mut v = Vec::new();
+//! rx_pkt.read_to_end(&mut v).unwrap();
+//! rx_pkt.consume_end().unwrap();
+//! ```
+mod read;
+pub use read::*;
+
+mod write;
+pub use write::*;