advance-practice/bridging-with-sync

[giscus[bot]]

advance-practice/bridging-with-sync

https://course.rs/advance-practice/bridging-with-sync.html

[LKY-stephen]

我才意识到运行时是runtime...

[chanble]

发现个奇怪的问题， 用block请求chunk的http， 中间就断了。 如果用单纯的异步是可以的， 谁知道怎么回事吗？

use std::sync::mpsc::{self, Sender};
use std::thread;

use futures_util::{StreamExt, Stream, TryStreamExt};
use http::{request::Builder, Method};

use hyper::body::HttpBody;
use hyper::{Body};
use hyper::client::Client;
use serde_derive::{Serialize, Deserialize};
use http::header::CONTENT_TYPE;
use futures_util::future::FutureExt;
use tokio::sync::mpsc as tokio_mpsc;

/**
cargo.toml文件： 
[package]
name = "stream-test"
version = "0.1.0"
edition = "2021"

# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html

[dependencies]
hyper = { version = "0.14", features = [ "full" ]}
tokio = {  version = "1", features = ["full"] }
http = "0.2"
serde = "1.0"
serde_json = "1.0"
base64 = "0.13"
serde_derive = "1.0"
futures-util = "0.3"
*/

fn main() {
    // println!("{:?}", response.body());
    // let (tx, mut rx) = tokio_mpsc::channel::<String>(1000);
    let rt = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .unwrap();
    let handle = thread::spawn(move || {
        rt.block_on(async {
            let client = Client::new();
            let req_builder = Builder::new().method(Method::POST).header("X-Registry-Auth", base64_url_encode(&get_payload()));
            let req = req_builder
                .uri("http://127.0.0.1:2375/v1.40/images/create?fromImage=ibmcom%2Fhelloworld&fromSrc=&repo=&tag=latest&platform=")
                .header(CONTENT_TYPE, "application/json")
                .body(Body::empty()).unwrap();
            let mut response = client.request(req).await.unwrap();
            // let mut response = client.request(req).await.unwrap();
            let body  = response.body_mut();
            let mut data = body.data().into_stream();
        
            while let Some(data) = data.next().await {
                println!("aaaaaaaaaaaaaaaaaaaaaaa: {:?}", data);
                if let Some(Ok(d)) = data {
                    let string = String::from_utf8(d.into_iter().collect::<Vec<u8>>()).unwrap_or_default();
                    println!("{}", string);
                    // tx.send(string);
                } else {
                    println!("data_stream next value {:?}", data);
                }
            }
        
        });
    });

    // for x in tokio::runtime::Builder::new_current_thread().enable_io().build().unwrap().block_on(rx.recv()) {
    //     println!("x: {}", x);
    // }
    handle.join().unwrap();
}

[bxb100]

tokio::runtime::Builder::new_current_thread() 和 thread::spawn 里面不是一个线程

[478619644h]

请求的是一个chunked http

            let mut data = body.data().into_stream(); //应该是这里在body里取了一次数据导致后续数据中断了
           //改成这样就可以
            let body  = response.body_mut();
            let mut data = body.into_stream();
        
            while let Some(data) = data.next().await {
                if let Ok(d) = data {
                    let string = String::from_utf8(d.into_iter().collect::<Vec<u8>>()).unwrap_or_default();
                    println!("{}", string);
                    // tx.send(string);
                } else {
                    println!("data_stream next value {:?}", data);
                }
            }

或者

use std::thread;

use futures_util::TryStreamExt;
use http::header::TRANSFER_ENCODING;
use http::{header::CONTENT_TYPE, request::Builder, Method};
use hyper::client::Client;
use hyper::Body;
use tokio_stream::StreamExt;
fn main() {
    let rt = tokio::runtime::Builder::new_multi_thread()
        .worker_threads(1)
        .enable_all()
        .build()
        .unwrap();
    let handle = thread::spawn(move || {
        rt.block_on(async {
            let client = Client::new();
            let req_builder = Builder::new().method(Method::GET);
            let req = req_builder
                .uri("http://127.0.0.1:8080/chunked-test")
                .header(CONTENT_TYPE, "application/octet-stream")
                .header(TRANSFER_ENCODING, "chunked")
                .body(Body::empty())
                .unwrap();

            println!("request sent {:?}",req);
            let mut response = client.request(req).await.unwrap();
            let vec = response.body_mut().into_stream().fold(Vec::new(), |mut vec,chunk| {
                vec.extend_from_slice(&chunk.unwrap());
                vec
            }).await; 
            let string =
                         String::from_utf8(vec).unwrap_or_default();
                     println!("{}", string);

        });
    });

    handle.join().unwrap();
}