@@ -17,12 +17,13 @@ use crate::{
1717use bytes:: { Buf , BytesMut } ;
1818use fallible_iterator:: FallibleIterator ;
1919use futures:: channel:: mpsc;
20- use futures:: { future, pin_mut, ready, StreamExt , TryStreamExt } ;
20+ use futures:: { future, pin_mut, ready, Stream , StreamExt , TryStreamExt } ;
2121use parking_lot:: Mutex ;
2222use postgres_protocol:: message:: backend:: Message ;
2323use postgres_types:: BorrowToSql ;
2424use std:: collections:: HashMap ;
2525use std:: fmt;
26+ use std:: pin:: Pin ;
2627use std:: sync:: Arc ;
2728use std:: task:: { Context , Poll } ;
2829use std:: time:: Duration ;
@@ -54,6 +55,17 @@ impl Responses {
5455}
5556}
5657
58+ impl Stream for Responses {
59+ type Item =Result < Message , Error > ;
60+
61+ fn poll_next ( mut self : Pin < & mut Self > , cx : & mut Context < ' _ > ) ->Poll < Option < Self :: Item > > {
62+ match ready ! ( ( * self ) . poll_next( cx) ) {
63+ Err ( err) if err. is_closed ( ) =>Poll :: Ready ( None ) ,
64+ msg =>Poll :: Ready ( Some ( msg) ) ,
65+ }
66+ }
67+ }
68+
5769/// A cache of type info and prepared statements for fetching type info
5870/// (corresponding to the queries in the [prepare](prepare) module).
5971#[ derive( Default ) ]