concurrency - Rust で MultiplexStream を実装する

Question

さびにマルチプレクサ/デマルチプレクサを実装したい。アップストリームの識別子をパケットの先頭に追加するだけで、DuplexStream単一の「ダウンストリーム」を介して複数の「アップストリーム」のデータを送信する必要があります。もちろん、これは逆の方法でも機能するはずです。つまり、ダウンストリームから受信したパケットから を読み取り、それを正しいアップストリーム ストリームに送信します。DuplexStreamport_numDuplexStreamport_num

そのような MultiplexStream の実装を開始しました (以下のコードはコンパイルされません)。ただし、問題に直面しています。対応するアップストリームにopen_portsマップされる変数は、Rust では許可されていない複数のタスクにアクセスできる必要があります。port_numDuplexStream

問題を解決するために、ここで適用できる設計パターンは何ですか?

impl MultiplexStream<T,U> {
    fn new(downstream: DuplexStream<(u32,T), U>) -> MultiplexStream<T,U> {
        let mut open_ports = HashMap::<u32, DuplexStream<(u32,T), U>>new();

        spawn do {
            let res = try {
                loop {
                    let (port_num, data) = downstream.recv();

                    match open_ports.find(port_num) {
                        Some(intermediate) => {
                            let res = try {
                                intermediate.send(data)
                            }

                            if res.is_err() {
                                open_ports.remove(port_num);
                            }
                        }
                        None => {}
                    }
                }
            }

            // downstream was closed => cleanup
            for intermediate in open_ports.values() {
                intermediate.close();
            }
            open_ports.clear();
        }

    }

    fn open<V: Send>(port_num: u32) -> Result(DuplexStream<V,T>, ()) {
        if open_ports.contains_key(port_num) {
            return Err(());
        }

        let (upstream, intermediate) = DuplexStream<V,T>::new();
        open_ports.insert(port_num, intermediate);

        spawn do {
            let res = try {
                loop {
                    let data = intermediate.recv();
                    downstream.send(~(port_num, data));
                }
            }

            // upstream was closed => cleanup
            intermediate.close();
            open_ports.remove(port_num);
        }

        return Ok(upstream);
    }
}