Function bridge

gateway/src/proxy/io_bridge.rs:93–164

(mut a: A, mut b: B, config: &ProxyConfig)

}

pub(crate) async fn bridge<A, B>(mut a: A, mut b: B, config: &ProxyConfig) -> Result<()>
where
    A: AsyncRead + AsyncWrite + Unpin,
    B: AsyncRead + AsyncWrite + Unpin,
{
    let buf_size = config.buffer_size;
    if !config.timeouts.data_timeout_enabled {
        debug!("copying bidirectionally");
        tokio::io::copy_bidirectional_with_sizes(&mut a, &mut b, buf_size, buf_size)
            .await
            .context("failed to copy")?;
        return Ok(());
    }

    let (mut ra, mut wa) = tokio::io::split(a);
    let (mut rb, mut wb) = tokio::io::split(b);

    let mut a2b = OneDirection {
        dir: "a2b",
        cfg: config,
        buf: BytesMut::with_capacity(buf_size),
        reader: &mut ra,
        writer: &mut wb,
        next_step: NextStep::Read,
    };
    let mut b2a = OneDirection {
        dir: "b2a",
        cfg: config,
        buf: BytesMut::with_capacity(buf_size),
        reader: &mut rb,
        writer: &mut wa,
        next_step: NextStep::Read,
    };

    let mut rest;
    // Transfer data between a and b bidirectionally.
    loop {
        tokio::select! {
            done = a2b.step() => {
                if done? {
                    // a to b is EOF, switch to b to a only
                    rest = Rest::B2a(b2a);
                    drop(a2b);
                    break;
                }
            }
            done = b2a.step() => {
                if done? {
                    // b to a is EOF, switch to a to b only
                    rest = Rest::A2b(a2b);
                    drop(b2a);
                    break;
                }
            }
        }
    }

    // One of the direction is closed, copy the other direction.