@@ -19,7 +19,7 @@ use crate::{
1919use bytes:: { Buf , BytesMut } ;
2020use fallible_iterator:: FallibleIterator ;
2121use futures_channel:: mpsc;
22- use futures_util:: { future, pin_mut, ready, StreamExt , TryStreamExt } ;
22+ use futures_util:: { future, pin_mut, ready, Stream , StreamExt , TryStreamExt } ;
2323use parking_lot:: Mutex ;
2424use postgres_protocol:: message:: backend:: Message ;
2525use postgres_types:: BorrowToSql ;
@@ -29,6 +29,7 @@ use std::fmt;
2929use std:: net:: IpAddr ;
3030#[ cfg( feature ="runtime" ) ]
3131use std:: path:: PathBuf ;
32+ use std:: pin:: Pin ;
3233use std:: sync:: Arc ;
3334use std:: task:: { Context , Poll } ;
3435#[ cfg( feature ="runtime" ) ]
@@ -61,6 +62,17 @@ impl Responses {
6162}
6263}
6364
65+ impl Stream for Responses {
66+ type Item =Result < Message , Error > ;
67+
68+ fn poll_next ( mut self : Pin < & mut Self > , cx : & mut Context < ' _ > ) ->Poll < Option < Self :: Item > > {
69+ match ready ! ( ( * self ) . poll_next( cx) ) {
70+ Err ( err) if err. is_closed ( ) =>Poll :: Ready ( None ) ,
71+ msg =>Poll :: Ready ( Some ( msg) ) ,
72+ }
73+ }
74+ }
75+
6476/// A cache of type info and prepared statements for fetching type info
6577/// (corresponding to the queries in the [prepare](prepare) module).
6678#[ derive( Default ) ]