jktcp

https://docs.rs/jktcp/
Admin
jktcp is a userspace TCP stack that runs over any ReadWrite transport, targeting reliable, ordered...
Tokens:51,803
Snippets:297
Trust Score:9.1
Update:1 month ago
Show doc for...
Tokens:
Raw
### Speed test setup and echo server

Source: https://docs.rs/jktcp/latest/src/jktcp/lib.rs.html

Configures a TUN device with MTU 16000, creates an Adapter, and spawns a TCP echo server that reads and writes back data.

```Rust
const CHUNK_SIZE: usize = 4096;

// Target device typically uses MTU 16000.
const MTU: u16 = 16000;
const MSS: usize = MTU as usize - 40 - 20;

let our_ip = Ipv6Addr::from_str("fd12:3456:789b::1").unwrap();
let their_ip = Ipv6Addr::from_str("fd12:3456:789b::2").unwrap();
let dev = DeviceBuilder::new()
    .ipv6(their_ip, "ffff:ffff:ffff:ffff::")
    .mtu(MTU)
    .build_async()
    .expect("Failed to create tunnel. Are you root?");

println!("Created tunnel [{:?}] {}", dev.name(), their_ip);

let mut adapter = Adapter::new(
    Box::new(AsyncDeviceWrapper {
        device: dev,
        buffer: BytesMut::new(),
    }),
    IpAddr::V6(our_ip),
    IpAddr::V6(their_ip),
);
adapter.set_mss(MSS);

// Echo server: read whatever arrives, write it straight back.
tokio::task::spawn(async move {
    let listener = tokio::net::TcpListener::bind(format!("[::0]:{SPEED_PORT}"))
        .await
        .unwrap();
    while let Ok((mut stream, _)) = listener.accept().await {
        tokio::task::spawn(async move {
            let mut buf = vec![0u8; 16 * 1024];
            loop {
                let n = match stream.read(&mut buf).await {
                    Ok(0) | Err(_) => return,
                    Ok(n) => n,
                };
                if stream.write_all(&buf[..n]).await.is_err() {
                    return;
                }
            }
        });
    }
});

// Give the kernel a moment to bring the interface up and start listening.
tokio::time::sleep(std::time::Duration::from_millis(500)).await;
```

--------------------------------

### AdapterHandle::pcap

Source: https://docs.rs/jktcp/latest/jktcp/handle/struct.AdapterHandle.html

Starts packet capture to the specified path.

```APIDOC
## AdapterHandle::pcap

### Description
Starts packet capture to the specified path.

### Method
pub async fn pcap(&mut self, path: impl Into<PathBuf>) -> Result<(), Error>

### Parameters
- **path** (impl Into<PathBuf>) - Required - The file path where packet capture data will be written.

### Returns
- **Result<(), Error>** - Ok(()) on success, or an Error on failure.
```

--------------------------------

### AdapterHandle multi-threaded usage

Source: https://docs.rs/jktcp/latest/index.html

Use this pattern when you need multiple connections, cross-task usage, or 'static bounds. AdapterHandle spawns the adapter's I/O loop onto the Tokio runtime, and StreamHandle is Send + Sync + 'static. The example shows creating a handle, connecting, and spawning a task that writes and reads.

```rust
use jktcp::adapter::Adapter;
use tokio::io::{AsyncReadExt, AsyncWriteExt};

let adapter = Adapter::new(Box::new(transport), todo!(), todo!());
let mut handle = adapter.to_async_handle();

let mut stream = handle.connect(1234).await?;

// stream is Send + 'static — spawn it freely
tokio::spawn(async move {
    stream.write_all(b"hello").await.unwrap();
    let mut buf = [0u8; 32];
    stream.read(&mut buf).await.unwrap();
});
```

--------------------------------

### pcap

Source: https://docs.rs/jktcp/latest/src/jktcp/handle.rs.html

Starts packet capture to the specified file path. This method is only available when the 'pcap' feature is enabled.

```APIDOC
## pcap

### Description
Starts packet capture to the specified file path. This method is only available when the 'pcap' feature is enabled.

### Method
POST

### Endpoint
pcap(path: impl Into<PathBuf>) -> Result<(), std::io::Error>

### Parameters
#### Request Body
- **path** (PathBuf) - Required - The file path where packet capture data will be written.

### Response
#### Success Response
- **()** - Indicates successful start of packet capture.

#### Error Response
- **std::io::Error** - Returns an error if the adapter is closed or the operation fails.
```

--------------------------------

### Adapter Configuration Methods

Source: https://docs.rs/jktcp/latest/src/jktcp/adapter.rs.html

Methods to configure the MSS and send window for the adapter.

```rust
    /// Set the Maximum Segment Size (MSS) for outbound TCP segments.
    ///
    /// This should be `MTU - 40 (IPv6 header) - 20 (TCP header)` for the
    /// tunnel you are using. Defaults to 1220 (based on IPv6 min MTU 1280).
    pub fn set_mss(&mut self, mss: usize) -> &mut Self {
        self.mss = mss;
        self
    }

    pub fn set_send_window(&mut self, window: usize) -> &mut Self {
        self.send_window = window;
        self
    }
```

--------------------------------

### Verify SYN window scale negotiation

Source: https://docs.rs/jktcp/latest/src/jktcp/adapter.rs.html

This snippet demonstrates how to drive the adapter's connect process and verify that the initial SYN packet contains the expected window scale option.

```rust
        Box::new(TestTransport(adapter_end)),
            IpAddr::V6(HOST_IP),
            IpAddr::V6(PEER_IP),
        );

        // Drive connect() far enough that the SYN hits the wire.
        let connect_fut = adapter.connect(PEER_PORT);
        let (_unused, syn) = tokio::join!(connect_fut, async { read_pkt(&mut test_rx).await });

        assert!(syn.flags.syn, "first packet must be SYN");
        let shift =
            crate::packets::parse_window_scale(&syn.options).expect("SYN should include WSopt");
        assert_eq!(shift, OUR_WSCALE, "advertised wscale should match constant");
    }
```

--------------------------------

### Get connection status

Source: https://docs.rs/jktcp/latest/src/jktcp/adapter.rs.html

Retrieves the current status of the connection associated with the host port.

```rust
pub(crate) fn get_status(&self, host_port: u16) -> Result<ConnectionStatus, std::io::Error> {
    match self.states.get(&host_port) {
        Some(s) => Ok(s.status.clone()),
        None => Err(std::io::Error::new(
            ErrorKind::NotConnected,
            "not connected",
        )),
    }
}
```

--------------------------------

### take

Source: https://docs.rs/jktcp/latest/jktcp/stream/struct.AdapterStream.html

Creates an adaptor which reads at most `limit` bytes from it.

```APIDOC
## take

### Description
Creates an adaptor which reads at most `limit` bytes from it.

### Method
take

### Signature
fn take(self, limit: u64) -> Take<Self> where Self: Sized

### Parameters
- **self** (Self) - Required - The reader to adapt.
- **limit** (u64) - Required - The maximum number of bytes to read.

### Returns
- **Take<Self>** - An adaptor that limits the number of bytes read.
```

--------------------------------

### AdapterHandle::connect

Source: https://docs.rs/jktcp/latest/jktcp/handle/struct.AdapterHandle.html

Establishes a connection to the specified port.

```APIDOC
## AdapterHandle::connect

### Description
Establishes a connection to the specified port.

### Method
pub async fn connect(&mut self, port: u16) -> Result<StreamHandle, Error>

### Parameters
- **port** (u16) - Required - The port to connect to.

### Returns
- **Result<StreamHandle, Error>** - A StreamHandle on success, or an Error on failure.
```

--------------------------------

### AdapterStream single-threaded usage

Source: https://docs.rs/jktcp/latest/index.html

Use this pattern when opening exactly one connection from a single async task. AdapterStream holds a mutable reference to the Adapter, so it cannot be moved across tasks or stored on the heap. The example shows connecting, writing, reading, and closing the stream.

```rust
use jktcp::adapter::Adapter;
use jktcp::stream::AdapterStream;
use tokio::io::{AsyncReadExt, AsyncWriteExt};

let mut adapter = Adapter::new(Box::new(transport), todo!(), todo!());

let mut stream = AdapterStream::connect(&mut adapter, 1234).await?;
stream.write_all(b"hello").await?;

let mut buf = [0u8; 32];
let n = stream.read(&mut buf).await?;
println!("received: {:?}", &buf[..n]);

stream.close().await?;
```

--------------------------------

### Initialize PCAP file

Source: https://docs.rs/jktcp/latest/src/jktcp/adapter.rs.html

Creates a new PCAP file at the specified path and writes the global header.

```rust
pub async fn pcap(&mut self, path: impl AsRef<Path>) -> Result<(), std::io::Error> {
    let mut file = tokio::fs::File::create(path).await?;
    file.write_all(&0xa1b2c3d4_u32.to_le_bytes()).await?;
    file.write_all(&2_u16.to_le_bytes()).await?;
    file.write_all(&4_u16.to_le_bytes()).await?;
    file.write_all(&0_i32.to_le_bytes()).await?;
    file.write_all(&0_u32.to_le_bytes()).await?;
    file.write_all(&(u16::MAX as u32).to_le_bytes()).await?;
    file.write_all(&101_u32.to_le_bytes()).await?;
    self.pcap = Some(std::sync::Arc::new(tokio::sync::Mutex::new(file)));
    Ok(())
}
```

--------------------------------

### Adapter Constructor

Source: https://docs.rs/jktcp/latest/src/jktcp/adapter.rs.html

Initializes a new Adapter instance with the provided peer and IP addresses.

```rust
impl Adapter {
    pub fn new(peer: Box<dyn ReadWrite>, host_ip: IpAddr, peer_ip: IpAddr) -> Self {
        Self {
            peer,
            host_ip,
            peer_ip,
            states: HashMap::new(),
            dropped: Vec::new(),
            read_buf: [0u8; 65536],
            bytes_in_buf: 0,
            pcap: None,
            mss: DEFAULT_MSS,
            send_window: DEFAULT_SEND_WINDOW,
            udp_sockets: HashMap::new(),
        }
    }
```

--------------------------------

### Main event loop handling HandleMessage variants and TCP packet processing

Source: https://docs.rs/jktcp/latest/src/jktcp/handle.rs.html

This snippet is the core event loop of the jktcp handle module. It processes incoming HandleMessage messages, handling connection setup, closing, sending data, UDP binding/unbinding, UDP sending, and optional pcap support. It also processes TCP packets, propagating errors to all streams and cleaning up dead connections.

```rust
127                                        }
128                                        Err(e) => Err(e),
129                                    };
130                                    res.send(connect_response).ok();
131                                }
132                                HandleMessage::Close { host_port } => {
133                                    handles.remove(&host_port);
134                                    adapter.close(host_port).await.ok();
135                                }
136                                HandleMessage::Send {
137                                    host_port,
138                                    data,
139                                    res,
140                                } => {
141                                    if let Err(e) = adapter.queue_send(&data, host_port) {
142                                        res.send(Err(e)).ok();
143                                    } else {
144                                        let response = adapter.write_buffer_flush().await;
145                                        res.send(response).ok();
146                                    }
147                                }
148                                HandleMessage::BindUdp { port, res } => {
149                                    let bound = adapter.bind_udp(port);
150                                    let (utx, urx) = mpsc::unbounded_channel();
151                                    udp_handles.insert(bound, utx);
152                                    res.send((bound, urx)).ok();
153                                }
154                                HandleMessage::UnbindUdp { port } => {
155                                    udp_handles.remove(&port);
156                                    adapter.unbind_udp(port);
157                                }
158                                HandleMessage::SendUdp {
159                                    source_port,
160                                    destination_port,
161                                    data,
162                                    res,
163                                } => {
164                                    let r = adapter
165                                        .send_udp(source_port, destination_port, &data)
166                                        .await;
167                                    res.send(r).ok();
168                                }
169                                #[cfg(feature = "pcap")]
170                                HandleMessage::Pcap {
171                                    path,
172                                    res
173                                } => {
174                                    res.send(adapter.pcap(path).await).ok();
175                                },
176                                HandleMessage::Die => {
177                                    break;
178                                }
179                            },
180                            None => {
181                                break;
182                            },
183                        }
184                    }
185
186                    r = adapter.process_tcp_packet() => {
187                        if let Err(e) = r {
188                            // propagate error to all streams; close them
189                            for (hp, tx) in handles.drain() {
190                                let _ = tx.send(Err(e.kind().into())); // or clone/convert
191                                let _ = adapter.close(hp).await;
192                            }
193                            break;
194                        }
195
196                        // Push any newly available bytes to per-conn channels
197                        let mut dead = Vec::new();
198                        for (&hp, tx) in &handles {
199                            match adapter.uncache_all(hp) {
200                                Ok(buf) if !buf.is_empty() => {
201                                    if tx.send(Ok(buf)).is_err() {
202                                        dead.push(hp);
203                                    }
204                                }
205                                Err(e) => {
206                                    let _ = tx.send(Err(e));
207                                    dead.push(hp);
208                                }
209                                _ => {}
210                            }
211                        }
212                        for hp in dead {
213                            handles.remove(&hp);
214                            let _ = adapter.close(hp).await;
215                        }
216
217                        let mut to_close = Vec::new();
218                        for (&hp, tx) in &handles {
219                            if let Ok(ConnectionStatus::Error(kind)) = adapter.get_status(hp) {
220                                if kind == std::io::ErrorKind::UnexpectedEof {
```

--------------------------------

### Adapter::new

Source: https://docs.rs/jktcp/latest/jktcp/adapter/struct.Adapter.html

Creates a new Adapter instance with the given peer stream and IP addresses.

```APIDOC
## Adapter::new

### Description
Creates a new Adapter instance with the given peer stream and IP addresses.

### Signature
pub fn new(peer: Box<dyn ReadWrite>, host_ip: IpAddr, peer_ip: IpAddr) -> Self

### Parameters
- **peer** (Box<dyn ReadWrite>) - Required - The peer stream for the tunnel.
- **host_ip** (IpAddr) - Required - The local (host) IP address.
- **peer_ip** (IpAddr) - Required - The remote (peer/device) IP address.

### Returns
- **Self** - A new Adapter instance.
```

--------------------------------

### AdapterHandle::new

Source: https://docs.rs/jktcp/latest/jktcp/handle/struct.AdapterHandle.html

Creates a new AdapterHandle from an Adapter instance.

```APIDOC
## AdapterHandle::new

### Description
Creates a new AdapterHandle from an Adapter instance.

### Method
pub fn new(adapter: Adapter) -> Self

### Parameters
- **adapter** (Adapter) - Required - The adapter to wrap in the handle.

### Returns
- **Self** - A new AdapterHandle instance.
```

--------------------------------

### Adapter Connection Logic

Source: https://docs.rs/jktcp/latest/src/jktcp/adapter.rs.html

Initiates a TCP connection by sending a SYN packet and waiting for a SYN-ACK, including retransmission logic.

```rust
    pub(crate) async fn connect(&mut self, port: u16) -> Result<u16, std::io::Error> {
        let host_port = loop {
            let p: u16 = rand::random();
            if !self.states.contains_key(&p) {
                break p;
            }
        };
        let state = ConnectionState::new(host_port, port);
        self.states.insert(host_port, state);

        // Send initial SYN.
        self.send_syn(host_port).await?;

        // Wait for SYN-ACK, retransmitting the SYN on timeout.
        let start = Instant::now();
        let mut last_syn = Instant::now();
        loop {
            self.process_tcp_packet().await?;

            if let Some(s) = self.states.get(&host_port) {
                match s.status {
                    ConnectionStatus::Connected => break,
                    ConnectionStatus::Error(e) => {
                        self.states.remove(&host_port);
                        return Err(std::io::Error::new(e, "failed to connect"));
                    }
                    ConnectionStatus::WaitingForSyn => {
                        if start.elapsed() > std::time::Duration::from_secs(10) {
                            self.states.remove(&host_port);
                            return Err(std::io::Error::new(ErrorKind::TimedOut, "SYN timed out"));
                        }
                        // Retransmit SYN every INITIAL_RTO_MS * 2^retry ms.
                        if last_syn.elapsed() >= std::time::Duration::from_millis(INITIAL_RTO_MS) {
                            debug!("retransmitting SYN for port {host_port}");
                            self.send_syn(host_port).await?;
                            last_syn = Instant::now();
                        }
                    }
                }
            }
        }

        Ok(host_port)
    }
```

--------------------------------

### TCP Connection Establishment

Source: https://docs.rs/jktcp/latest/src/jktcp/handle.rs.html

Initiates a connection to a target port with an 8-second timeout.

```rust
pub async fn connect(&mut self, port: u16) -> Result<StreamHandle, std::io::Error> {
        let (res_tx, res_rx) = oneshot::channel();
        if self
            .sender
            .send(HandleMessage::ConnectToPort {
                target: port,
                res: res_tx,
            })
            .is_err()
        {
            return Err(std::io::Error::new(
                std::io::ErrorKind::NetworkUnreachable,
                "adapter closed",
            ));
        }

        match timeout(std::time::Duration::from_secs(8), res_rx).await {
            Ok(Ok(r)) => {
                let (host_port, recv_channel) = r?;
                Ok(StreamHandle {
                    host_port,
                    recv_channel: Mutex::new(recv_channel),
                    send_channel: self.sender.clone(),
                    read_buffer: Vec::new(),
                    pending_writes: FuturesUnordered::new(),
                })
            }
            Ok(Err(_)) => Err(std::io::Error::new(
                std::io::ErrorKind::BrokenPipe,
                "adapter closed",
            )),
            Err(_) => Err(std::io::Error::new(
                std::io::ErrorKind::TimedOut,
                "channel recv timeout",
            )),
        }
    }
```

--------------------------------

### Speed test client and measurement

Source: https://docs.rs/jktcp/latest/src/jktcp/lib.rs.html

Connects to the echo server, splits the stream, and measures throughput by writing and reading data in chunks, then prints the result.

```Rust
let mut handle = adapter.to_async_handle();
let stream = match handle.connect(SPEED_PORT).await {
    Ok(s) => s,
    Err(e) => panic!("connect failed: {e:?}"),
};

let (mut reader, mut writer) = tokio::io::split(stream);

let payload = vec![0xABu8; CHUNK_SIZE];
let start = std::time::Instant::now();

let write_task = tokio::task::spawn(async move {
    let chunks = TOTAL_BYTES / CHUNK_SIZE;
    for _ in 0..chunks {
        writer.write_all(&payload).await.expect("write_all");
    }
    writer.flush().await.expect("flush");
});

let read_task = tokio::task::spawn(async move {
    let mut buf = vec![0u8; CHUNK_SIZE];
    let mut got = 0usize;
    while got < TOTAL_BYTES {
        let n = reader.read(&mut buf).await.expect("read");
        if n == 0 {
            panic!("eof at {got}/{TOTAL_BYTES}");
        }
        got += n;
    }
});

write_task.await.expect("writer panicked");
read_task.await.expect("reader panicked");
let total_elapsed = start.elapsed();

let mib = TOTAL_BYTES as f64 / (1024.0 * 1024.0);
println!(
    "handle_speed: {mib:.2} MiB echo in {CHUNK_SIZE}-byte chunks   \
     {:>7.2} ms   {:>7.3} MiB/s",
    total_elapsed.as_secs_f64() * 1000.0,
    mib / total_elapsed.as_secs_f64(),
);

handle.close().await.ok();
```

--------------------------------

### window_scale_option

Source: https://docs.rs/jktcp/latest/jktcp/packets/index.html

Builds a TCP Window Scale option.

```APIDOC
## window_scale_option

### Description
Builds a TCP Window Scale option.

### Function Signature
window_scale_option(shift_count: u8) -> Vec<u8>

### Parameters
- **shift_count** (u8) - Required - The shift count for the Window Scale option.

### Returns
- **Vec<u8>** - The constructed TCP Window Scale option bytes.
```

--------------------------------

### connect

Source: https://docs.rs/jktcp/latest/src/jktcp/handle.rs.html

Establishes a TCP connection to the specified port on the remote device. Returns a StreamHandle for reading and writing data.

```APIDOC
## connect

### Description
Establishes a TCP connection to the specified port on the remote device. Returns a StreamHandle for reading and writing data.

### Method
POST

### Endpoint
connect(port: u16) -> Result<StreamHandle, std::io::Error>

### Parameters
#### Request Body
- **port** (u16) - Required - The remote port to connect to.

### Response
#### Success Response
- **StreamHandle** - A handle to the established TCP stream, providing read/write capabilities.

#### Error Response
- **std::io::Error** - Returns an error if the adapter is closed, the connection times out, or the operation fails.
```

--------------------------------

### Adapter Accessor Methods

Source: https://docs.rs/jktcp/latest/src/jktcp/adapter.rs.html

Methods to retrieve the host and peer IP addresses.

```rust
    /// The local (host) IP address of this tunnel endpoint.
    pub fn host_ip(&self) -> IpAddr {
        self.host_ip
    }

    /// The remote (peer/device) IP address of this tunnel endpoint.
    pub fn peer_ip(&self) -> IpAddr {
        self.peer_ip
    }
```

--------------------------------

### TCP Window Scale Option Utilities

Source: https://docs.rs/jktcp/latest/src/jktcp/packets.rs.html

Functions to build and parse TCP Window Scale options from a byte slice.

```rust
pub fn window_scale_option(shift: u8) -> [u8; 3] {
    // kind=3, length=3, shift_count
    [3, 3, shift.min(14)]
}
```

```rust
pub fn parse_window_scale(options: &[u8]) -> Option<u8> {
    let mut i = 0;
    while i < options.len() {
        match options[i] {
            0 => return None, // End-of-options
            1 => {
                i += 1;
                continue;
            } // NOP
            3 => {
                // Window scale: kind=3, length=3, shift.
                if i + 2 < options.len() && options[i + 1] == 3 {
                    return Some(options[i + 2].min(14));
                }
                return None;
            }
            _ => {
                // Skip unknown TLV option by its length byte.
                if i + 1 >= options.len() {
                    return None;
                }
                let len = options[i + 1] as usize;
                if len < 2 {
                    return None;
                }
                i += len;
            }
        }
    }
    None
}
```

--------------------------------

### Local TCP Integration Test

Source: https://docs.rs/jktcp/latest/src/jktcp/lib.rs.html

A tokio test that sets up a virtual device, initializes an adapter, and performs TCP communication.

```rust
const SERVER_PORT: u16 = 5555;

    #[tokio::test]
    async fn local_tcp() {
        let our_ip = Ipv6Addr::from_str("fd12:3456:789a::1").unwrap();
        let their_ip = Ipv6Addr::from_str("fd12:3456:789a::2").unwrap();
        let dev = DeviceBuilder::new()
            .ipv6(their_ip, "ffff:ffff:ffff:ffff::")
            .mtu(1420)
            .build_async()
            .expect("Failed to create tunnel. Are you root?");

        println!("Created tunnel [{:?}] {}", dev.name(), their_ip);

        let mut adapter = Adapter::new(
            Box::new(AsyncDeviceWrapper {
                device: dev,
                buffer: BytesMut::new(),
            }),
            IpAddr::V6(our_ip),
            IpAddr::V6(their_ip),
        );
        adapter.pcap("./local_tcp.pcap").await.expect("no pcap");

        tokio::task::spawn(async move {
            let listener = tokio::net::TcpListener::bind(format!("[::0]:{SERVER_PORT}"))
                .await
                .unwrap();
            while let Ok((mut stream, addr)) = listener.accept().await {
                println!("Accepted connection from {addr:?}");

                tokio::task::spawn(async move {
                    loop {
                        let mut buf = [0; 1024];
                        let read_len = stream.read(&mut buf).await.unwrap();
                        stream.write_all(&buf[..read_len]).await.unwrap();
                    }
                });
            }
        });

        println!("Attach Wireshark, press enter to continue\n");
        let mut buf = Vec::new();
        let _ = tokio::io::stdin().read(&mut buf).await.unwrap();

        let mut stream = match AdapterStream::connect(&mut adapter, SERVER_PORT).await {
            Ok(s) => s,
            Err(e) => {
                println!("no connect: {e:?}");
                return;
            }
        };

        if let Err(e) = stream.write_all(&[1, 2, 3, 4, 5]).await {
            println!("no send: {e:?}");
        } else {
            let mut buf = [0u8; 4];
            match stream.read_exact(&mut buf).await {
                Ok(_) => println!("recv'd {buf:?}"),
                Err(e) => println!("no recv: {e:?}"),
            }
        }

        if let Err(e) = stream.write_all(&[69, 69, 42, 0, 1]).await {
            println!("no send: {e:?}");
        } else {
            let mut buf = [0u8; 6];
            match stream.read_exact(&mut buf).await {
                Ok(_) => println!("recv'd {buf:?}"),
                Err(e) => println!("no recv: {e:?}"),
            }
        }

        if let Err(e) = stream.close().await {
            println!("no close: {e:?}");
        }

        tokio::time::sleep(std::time::Duration::from_secs(2)).await;
        println!("\n\npress enter");
        let mut buf = Vec::new();
        let _ = tokio::io::stdin().read(&mut buf).await.unwrap();
    }
```

--------------------------------

### pub async fn from_reader<R: AsyncRead + Unpin + AsyncReadExt>(reader: &mut R, log: &Option<PcapLog>) -> Result<Self, Error>

Source: https://docs.rs/jktcp/latest/jktcp/packets/struct.Ipv4Packet.html

Asynchronously reads an IPv4 packet from a Tokio AsyncRead source. Returns a Result with the packet or an Error.

```APIDOC
## pub async fn from_reader<R: AsyncRead + Unpin + AsyncReadExt>(reader: &mut R, log: &Option<PcapLog>) -> Result<Self, Error>

### Description
Asynchronously reads an IPv4 packet from a Tokio `AsyncRead` source. The `log` parameter is used for optional Pcap logging.

### Parameters
- **reader**: `&mut R` - A mutable reference to an `AsyncRead` + `Unpin` + `AsyncReadExt` type.
- **log**: `&Option<PcapLog>` - An optional Pcap log reference.

### Returns
- `Result<Self, Error>` - `Ok(Ipv4Packet)` on success, `Err(Error)` on failure.
```

--------------------------------

### Packet construction helpers

Source: https://docs.rs/jktcp/latest/src/jktcp/adapter.rs.html

Functions to build raw TCP segments and IPv6-wrapped packets for simulating peer-to-adapter traffic.

```rust
fn tcp_seg(dst_port: u16, seq: u32, ack_num: u32, flags: TcpFlags, payload: &[u8]) -> Vec<u8> {
    tcp_seg_with_opts(dst_port, seq, ack_num, flags, &[], payload)
}

fn tcp_seg_with_opts(
    dst_port: u16,
    seq: u32,
    ack_num: u32,
    flags: TcpFlags,
    options: &[u8],
    payload: &[u8],
) -> Vec<u8> {
    TcpPacket::create(
        IpAddr::V6(PEER_IP),
        IpAddr::V6(HOST_IP),
        PEER_PORT,
        dst_port,
        seq,
        ack_num,
        flags,
        u16::MAX - 1,
        options,
        payload,
    )
}

fn peer_ipv6_pkt(
    dst_port: u16,
    seq: u32,
    ack_num: u32,
    flags: TcpFlags,
    payload: &[u8],
) -> Vec<u8> {
    let tcp = tcp_seg(dst_port, seq, ack_num, flags, payload);
    crate::packets::Ipv6Packet::create(PEER_IP, HOST_IP, ProtocolNumber::Tcp, 255, &tcp)
}
```

--------------------------------

### Adapter::to_async_handle

Source: https://docs.rs/jktcp/latest/jktcp/adapter/struct.Adapter.html

Wraps this adapter in a thread-safe handle.

```APIDOC
## Adapter::to_async_handle

### Description
Wraps this adapter in a thread-safe handle.

### Signature
pub fn to_async_handle(self) -> AdapterHandle

### Returns
- **AdapterHandle** - A thread-safe handle to the adapter.
```

--------------------------------

### pub fn parse_window_scale(options: &[u8]) -> Option<u8>

Source: https://docs.rs/jktcp/latest/jktcp/packets/fn.parse_window_scale.html

Scans a TCP options blob for the Window Scale option. Returns the shift count if present and well-formed; otherwise None.

```APIDOC
## Function: parse_window_scale

### Description
Scans a TCP options blob for the Window Scale option. Returns the shift count if present and well-formed; otherwise None.

### Signature
```rust
pub fn parse_window_scale(options: &[u8]) -> Option<u8>
```

### Parameters
- **options** (`&[u8]`) - The TCP options blob to scan.

### Returns
- `Option<u8>` - The shift count if the Window Scale option is present and well-formed; otherwise `None`.

```

--------------------------------

### Create IPv6 packet bytes (create)

Source: https://docs.rs/jktcp/latest/src/jktcp/packets.rs.html

Builds an IPv6 packet as a byte vector from source and destination addresses, next header, hop limit, and payload. Sets version to 6, traffic class and flow label to 0, and payload length to the payload's length.

```rust
pub fn create(
    source: Ipv6Addr,
    destination: Ipv6Addr,
    next_header: ProtocolNumber,
    hop_limit: u8,
    payload: &[u8],
) -> Vec<u8> {
    let mut packet = Vec::with_capacity(40 + payload.len());

    // Version (6) and Traffic Class (0)
    let version_traffic_class = 6 << 4;
    packet.push(version_traffic_class);
    packet.push(0); // The rest of the Traffic Class and the start of the Flow Label

    // Flow Label (0)
    let flow_label = 0u16;
    packet.extend_from_slice(&flow_label.to_be_bytes()[..]);

    // Payload Length (length of the payload only)
    packet.extend_from_slice(&(payload.len() as u16).to_be_bytes());

    // Next Header and Hop Limit
    packet.push(next_header as u8);
    packet.push(hop_limit);

    // Source and Destination Addresses
```

--------------------------------

### AdapterStream::connect

Source: https://docs.rs/jktcp/latest/jktcp/stream/struct.AdapterStream.html

Asynchronously connects to the specified port using the provided adapter.

```APIDOC
## AdapterStream::connect

### Description
Asynchronously connects to the specified port using the provided adapter.

### Method
async fn connect

### Signature
pub async fn connect(adapter: &'a mut Adapter, port: u16) -> Result<Self, Error>

### Parameters
- **adapter** (&'a mut Adapter) - Required - The adapter to use for the connection.
- **port** (u16) - Required - The port to connect to.

### Returns
- **Result<Self, Error>** - The connected AdapterStream or an error.
```

--------------------------------

### Adapter::new

Source: https://docs.rs/jktcp/latest/src/jktcp/adapter.rs.html

Constructs a new Adapter with the given framed transport and local/remote IP addresses. This is the entry point for creating a TCP tunnel adapter.

```APIDOC
## Adapter::new

### Description
Constructs a new Adapter with the given framed transport and local/remote IP addresses. This is the entry point for creating a TCP tunnel adapter.

### Method
Constructor

### Parameters
- **peer** (Box<dyn ReadWrite>) - Required - The framed transport for sending and receiving raw packets.
- **host_ip** (IpAddr) - Required - The local (host) IP address of this tunnel endpoint.
- **peer_ip** (IpAddr) - Required - The remote (peer/device) IP address of this tunnel endpoint.

### Returns
- **Adapter** - The newly constructed Adapter instance.
```

--------------------------------

### window_scale_option(shift: u8) -> [u8; 3]

Source: https://docs.rs/jktcp/latest/jktcp/packets/fn.window_scale_option.html

Builds a TCP Window Scale option as a 3-byte array. The shift parameter specifies the window scale shift count.

```APIDOC
## Function: window_scale_option

### Description
Builds a TCP Window Scale option.

### Signature
pub fn window_scale_option(shift: u8) -> [u8; 3]

### Parameters
- **shift** (u8) - Required - The window scale shift count.

### Return Value
- **array** ([u8; 3]) - The TCP Window Scale option as a 3-byte array.

### Example
```rust
let option = window_scale_option(7);
```
```

--------------------------------

### Test sliding window flow control

Source: https://docs.rs/jktcp/latest/src/jktcp/adapter.rs.html

Demonstrates how the adapter holds back data when the send window is full and releases it upon receiving an ACK.

```rust
let (adapter_end, test_end) = tokio::io::duplex(1 << 16);
        let (mut test_rx, mut test_tx) = tokio::io::split(test_end);
        let mut adapter = Adapter::new(
            Box::new(TestTransport(adapter_end)),
            IpAddr::V6(HOST_IP),
            IpAddr::V6(PEER_IP),
        );
        adapter.set_send_window(5);

        let hp = handshake(&mut adapter, &mut test_rx, &mut test_tx).await;

        adapter.queue_send(b"first", hp).unwrap();
        adapter.write_buffer_flush().await.unwrap();
        let psh1 = read_pkt(&mut test_rx).await;
        assert_eq!(psh1.payload, b"first");

        // Queue second chunk; window is full, must be held.
        adapter.queue_send(b"secnd", hp).unwrap();
        adapter.write_buffer_flush().await.unwrap();

        // Nothing should appear on the wire yet.
        let nothing = tokio::time::timeout(Duration::from_millis(10), read_pkt(&mut test_rx)).await;
        assert!(
            nothing.is_err(),
            "second chunk must not be sent while window is full"
        );

        // ACK the first chunk (ack = psh1.seq + len).
        let peer_ack_num = psh1.sequence_number.wrapping_add(psh1.payload.len() as u32);
        adapter
            .process_tcp_packet_from_payload(&tcp_seg(
                hp,
                PEER_ISN + 1,
                peer_ack_num,
                TcpFlags {
                    ack: true,
                    ..Default::default()
                },
                &[],
            ))
            .await
            .unwrap();

        // ACK frees window; second chunk goes out on the next flush.
        adapter.write_buffer_flush().await.unwrap();
        let psh2 = read_pkt(&mut test_rx).await;
        assert_eq!(psh2.payload, b"secnd");
        assert_eq!(
            psh2.sequence_number, peer_ack_num,
            "second segment starts where first left off"
        );
    }
```

--------------------------------

### write_all

Source: https://docs.rs/jktcp/latest/jktcp/stream/struct.AdapterStream.html

Attempts to write an entire buffer into this writer.

```APIDOC
## write_all

### Description
Attempts to write an entire buffer into this writer.

### Method
write_all

### Signature
fn write_all<'a>(&'a mut self, src: &'a [u8]) -> WriteAll<'a, Self> where Self: Unpin

### Parameters
- **self** (&'a mut Self) - Required - The writer to write to.
- **src** (&'a [u8]) - Required - The buffer to write.

### Returns
- **WriteAll<'a, Self>** - A future that resolves to the number of bytes written.
```

--------------------------------

### Verify peer window scale is honored

Source: https://docs.rs/jktcp/latest/src/jktcp/adapter.rs.html

This test ensures that the adapter correctly interprets window advertisements from the peer by applying the negotiated window scale factor.

```rust
    #[tokio::test]
    async fn peer_window_scale_is_honored() {
        tokio::time::pause();

        let (adapter_end, test_end) = tokio::io::duplex(1 << 16);
        let (mut test_rx, mut test_tx) = tokio::io::split(test_end);
        let mut adapter = Adapter::new(
            Box::new(TestTransport(adapter_end)),
            IpAddr::V6(HOST_IP),
            IpAddr::V6(PEER_IP),
        );
        adapter.set_mss(10);
        adapter.set_send_window(1 << 20); // effectively unlimited from our side.

        let peer_wscale = 4u8;
        let peer_raw_window = 2u16; // scaled: 2 << 4 = 32 bytes.

        // Drive connect() while we manually act as the peer.
        let (hp_result, peer_hp) = tokio::join!(adapter.connect(PEER_PORT), async {
            let syn = read_pkt(&mut test_rx).await;
            assert!(syn.flags.syn);
            let hp = syn.source_port;

            // SYN-ACK with WSopt and a tiny scaled window.
            let wscale_opt = crate::packets::window_scale_option(peer_wscale);
            let tcp = TcpPacket::create(
                IpAddr::V6(PEER_IP),
                IpAddr::V6(HOST_IP),
                PEER_PORT,
                hp,
                PEER_ISN,
                syn.sequence_number.wrapping_add(1),
                TcpFlags {
                    syn: true,
                    ack: true,
                    ..Default::default()
                },
                peer_raw_window,
                &wscale_opt,
                &[],
            );
            let ip = crate::packets::Ipv6Packet::create(
                PEER_IP,
                HOST_IP,
                ProtocolNumber::Tcp,
                255,
                &tcp,
            );
            test_tx.write_all(&ip).await.unwrap();

            // Consume the adapter's ACK.
            let _ack = read_pkt(&mut test_rx).await;
            hp
        });
        let hp = hp_result.expect("connect failed");
        assert_eq!(hp, peer_hp);

        // Try to send 100 bytes. With peer_window=32 and mss=10, we expect
        // 3 full segments (30 bytes) plus one short segment (2 bytes) = 32
        // bytes on the wire; the remaining 68 bytes must be held.
        adapter.queue_send(&[b'a'; 100], hp).unwrap();
        adapter.write_buffer_flush().await.unwrap();

        let mut sent = 0;
        while sent < 32 {
            let p = read_pkt(&mut test_rx).await;
            sent += p.payload.len();
            assert!(sent <= 32, "must never exceed peer_window={}", 32);
        }
        assert_eq!(sent, 32);

        let blocked = tokio::time::timeout(Duration::from_millis(10), read_pkt(&mut test_rx)).await;
        assert!(blocked.is_err(), "must not exceed peer's scaled window");
    }
```

--------------------------------

### AdapterStream::connect

Source: https://docs.rs/jktcp/latest/src/jktcp/stream.rs.html

Connects to the specified port on the adapter and returns a new AdapterStream instance.

```APIDOC
## AdapterStream::connect

### Description
Connects to the specified port on the adapter and returns a new AdapterStream instance.

### Method
async fn connect(adapter: &'a mut Adapter, port: u16) -> Result<Self, std::io::Error>

### Parameters
- **adapter** (&'a mut Adapter) - Required - The adapter to connect through.
- **port** (u16) - Required - The target port to connect to.

### Returns
- **Result<Self, std::io::Error>** - The new AdapterStream on success, or an I/O error on failure.

### Example
rust
let stream = AdapterStream::connect(&mut adapter, 8080).await?;

```

--------------------------------

### handshake helper

Source: https://docs.rs/jktcp/latest/src/jktcp/adapter.rs.html

Drives the adapter's connect method and the test-peer response concurrently.

```rust
async fn handshake(
    adapter: &mut Adapter,
    test_rx: &mut (impl AsyncRead + Unpin),
    test_tx: &mut (impl AsyncWrite + Unpin),
) -> u16 {
    let (hp_result, peer_hp) = tokio::join!(adapter.connect(PEER_PORT), async {
```

--------------------------------

### Parse IPv6 packet from async reader (from_reader)

Source: https://docs.rs/jktcp/latest/src/jktcp/packets.rs.html

Reads an IPv6 packet from an async reader, validating the version, extracting header fields, and reading the payload. Returns an error if the version is not 6. Optionally logs the packet.

```rust
pub async fn from_reader<R: AsyncRead + Unpin>(
    reader: &mut R,
    log: &Option<PcapLog>,
) -> Result<Self, std::io::Error> {
    let mut log_packet = Vec::new();

    let mut header = [0u8; 40]; // IPv6 header size is fixed at 40 bytes
    reader.read_exact(&mut header).await?;

    if log.is_some() {
        log_packet.extend_from_slice(&header);
    }

    let version = header[0] >> 4;
    if version != 6 {
        debug!("Got an invalid IPv6 header from reader");
        return Err(std::io::Error::new(
            std::io::ErrorKind::InvalidData,
            "Invalid IPv6 header",
        ));
    }

    let traffic_class = ((header[0] & 0x0F) << 4) | ((header[1] & 0xF0) >> 4);
    let flow_label =
        ((header[1] as u32 & 0x0F) << 16) | ((header[2] as u32) << 8) | (header[3] as u32);
    let payload_length = u16::from_be_bytes([header[4], header[5]]);
    let next_header = header[6];
    let hop_limit = header[7];
    let source = Ipv6Addr::new(
        u16::from_be_bytes([header[8], header[9]]),
        u16::from_be_bytes([header[10], header[11]]),
        u16::from_be_bytes([header[12], header[13]]),
        u16::from_be_bytes([header[14], header[15]]),
        u16::from_be_bytes([header[16], header[17]]),
        u16::from_be_bytes([header[18], header[19]]),
        u16::from_be_bytes([header[20], header[21]]),
        u16::from_be_bytes([header[22], header[23]]),
    );
    let destination = Ipv6Addr::new(
        u16::from_be_bytes([header[24], header[25]]),
        u16::from_be_bytes([header[26], header[27]]),
        u16::from_be_bytes([header[28], header[29]]),
        u16::from_be_bytes([header[30], header[31]]),
        u16::from_be_bytes([header[32], header[33]]),
        u16::from_be_bytes([header[34], header[35]]),
        u16::from_be_bytes([header[36], header[37]]),
        u16::from_be_bytes([header[38], header[39]]),
    );

    // Read the payload
    let mut payload = vec![0u8; payload_length as usize];
    reader.read_exact(&mut payload).await?;
    if let Some(log) = log {
        log_packet.extend_from_slice(&payload);
        crate::log_packet(log, &log_packet);
    }

    Ok(Self {
        version,
        traffic_class,
        flow_label,
        payload_length,
        next_header,
        hop_limit,
        source,
        destination,
        payload,
    })
}
```

--------------------------------

### AsyncDeviceWrapper implementation

Source: https://docs.rs/jktcp/latest/src/jktcp/lib.rs.html

A test utility that wraps an AsyncDevice to provide AsyncRead functionality with internal buffering.

```rust
pub struct AsyncDeviceWrapper {
    device: AsyncDevice,
    // Buffer to store unread data
    buffer: BytesMut,
}

impl std::fmt::Debug for AsyncDeviceWrapper {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("AsyncDeviceWrapper")
            .field("buffer", &self.buffer)
            .finish()
    }
}

impl AsyncRead for AsyncDeviceWrapper {
    fn poll_read(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        buf: &mut ReadBuf<'_>,
    ) -> Poll<std::io::Result<()>> {
        // First, check if we have data in our buffer
        let this = self.get_mut();

        if !this.buffer.is_empty() {
            // We have buffered data, copy as much as possible to the output buffer
            let bytes_to_copy = std::cmp::min(this.buffer.len(), buf.remaining());
            let data_to_copy = this.buffer.split_to(bytes_to_copy);
            buf.put_slice(&data_to_copy);

            return Poll::Ready(Ok(()));
        }

        // If our buffer is empty, try to read more data
        let mut temp_buf = vec![0u8; 65536];

        match this.device.poll_recv(cx, &mut temp_buf) {
            Poll::Ready(Ok(n)) => {
                if n > 0 {
                    // Got some data, first fill the output buffer
```
Tokens:
Raw