is there an example of how to generate streaming response in a separate thread?

[vitalije]

I am trying to implement response handler which should generate a very large JSON result in a separate thread, one property at the time. I wish that server starts sending response by sending headers first and the opening brace followed by the first property - value pair, then to send each property as it is produced by the separate thread. I wish to avoid having the entire JSON object in the memory. Instead I wish to produce it peace by peace and send it as soon as the buffer fills up.

I've tried many things but without success. Here is where I am right now:

async fn handler_abc(r:web::HttpRequest) -> web::HttpResponse {
    let (tx, rx) = ntex::channel::mpsc::channel();
    _ = std::thread::spawn(move ||{
        let mut buf = Vec::<u8>::with_capacity(1024);
        let mut n = 0u32;
        buf.extend_from_slice("{\"prop00000000\":\"dummy content\"".as_bytes());
        while sender.send(Bytes::from(buf)).is_ok() {
            buf[0] = b',';
            n += 1;
            buf[6] = {let x = (((n >> 28) & 0xF) + 48) as u8; if x < 58 {x} else {x + 7}};
            buf[7] = {let x = (((n >> 24) & 0xF) + 48) as u8; if x < 58 {x} else {x + 7}};
            buf[8] = {let x = (((n >> 20) & 0xF) + 48) as u8; if x < 58 {x} else {x + 7}};
            buf[9] = {let x = (((n >> 16) & 0xF) + 48) as u8; if x < 58 {x} else {x + 7}};
            buf[10] = {let x = (((n >> 12) & 0xF) + 48) as u8; if x < 58 {x} else {x + 7}};
            buf[11] = {let x = (((n >> 8) & 0xF) + 48) as u8; if x < 58 {x} else {x + 7}};
            buf[12] = {let x = (((n >> 4) & 0xF) + 48) as u8; if x < 58 {x} else {x + 7}};
            buf[13] = {let x = ((n & 0xF) + 48) as u8; if x < 58 {x} else {x + 7}};
            if n == 100 {
                _ = sender.send(Bytes::from("}".as_bytes()));
                sender.close();
                break;
            }
        }
    });
    Some(web::HttpResponse::Ok().content_type("application/json").streaming(stream.into()))
}

Compiler complains about std::rc::Rc<UnsafeCell<ntex::channel::mpsc::Shared<ntex::util::Bytes>>>` cannot be sent between threads safely

Can anyone help me achieve my goal or direct me in the right direction.

[fafhrd91]

ntex::channel is not thread safe. you have to use thread-safe async channel, something like async-channel

[vitalije]

Thanks, I'll try it now.