I'm trying to set a hyper::Body
to the results of tokio_postgres
stream like this:
fn stream_ads(&self, req: Request) -> ResponseFuture {
let connector = OpenSsl::new().unwrap();
let notifications = Connection::connect(
self.database_url.clone(),
TlsMode::Require(Box::new(connector)),
&self.handle.clone(),
).then(|c| c.unwrap().batch_execute("listen ad_update"))
.map(|c| {
let mut resp = Response::new()
.with_header(ContentType("text/event-stream".parse().unwrap()))
.with_header(CacheControl(
vec![CacheDirective::NoStore, CacheDirective::Private],
));
let stream = c.notifications().map(|n| n.payload).map_err(|e| {
hyper::Error::Io(StdIoError::new(std::io::ErrorKind::Other, e.description()))
});
resp.set_body(Box::new(stream));
resp
})
.map_err(|(e, _)| {
hyper::Error::Io(StdIoError::new(std::io::ErrorKind::Other, e.description()))
});
Box::new(notifications)
}
However I can't seem to make it work, the hyper docs seem to imply that any stream can be a body, but my call to set_body errors out at:
error[E0277]: the trait bound `hyper::Body: std::convert::From<std::boxed::Box<futures::stream::MapErr<futures::stream::Map<tokio_postgres::Notifications, [closure@src/server.rs:294:52: 294:65]>, [closure@src/server.rs:294:75: 296:18]>>>` is not satisfied
--> src/server.rs:297:22
|
297 | resp.set_body(Box::new(stream));
| ^^^^^^^^ the trait `std::convert::From<std::boxed::Box<futures::stream::MapErr<futures::stream::Map<tokio_postgres::Notifications, [closure@src/server.rs:294:52: 294:65]>, [closure@src/server.rs:294:75: 296:18]>>>` is not implemented for `hyper::Body`
|
= help: the following implementations were found:
<hyper::Body as std::convert::From<std::borrow::Cow<'static, [u8]>>>
<hyper::Body as std::convert::From<&'static [u8]>>
<hyper::Body as std::convert::From<std::string::String>>
<hyper::Body as std::convert::From<&'static str>>
and 7 others
= note: required because of the requirements on the impl of `std::convert::Into<hyper::Body>` for `std::boxed::Box<futures::stream::MapErr<futures::stream::Map<tokio_postgres::Notifications, [closure@src/server.rs:294:52: 294:65]>, [closure@src/server.rs:294:75: 296:18]>>`
Is there something I'm not understanding? Or a simpler way to go about this?