@@ -21,12 +21,13 @@ use crate::{
2121use bytes:: { Buf , BytesMut } ;
2222use fallible_iterator:: FallibleIterator ;
2323use futures_channel:: mpsc;
24- use futures_util:: { future, pin_mut, ready, StreamExt , TryStreamExt } ;
24+ use futures_util:: { future, pin_mut, ready, Stream , StreamExt , TryStreamExt } ;
2525use parking_lot:: Mutex ;
2626use postgres_protocol:: message:: { backend:: Message , frontend} ;
2727use postgres_types:: BorrowToSql ;
2828use std:: collections:: HashMap ;
2929use std:: fmt;
30+ use std:: pin:: Pin ;
3031use std:: sync:: Arc ;
3132use std:: task:: { Context , Poll } ;
3233#[ cfg( feature ="runtime" ) ]
@@ -59,6 +60,17 @@ impl Responses {
5960}
6061}
6162
63+ impl Stream for Responses {
64+ type Item =Result < Message , Error > ;
65+
66+ fn poll_next ( mut self : Pin < & mut Self > , cx : & mut Context < ' _ > ) ->Poll < Option < Self :: Item > > {
67+ match ready ! ( ( * self ) . poll_next( cx) ) {
68+ Err ( err) if err. is_closed ( ) =>Poll :: Ready ( None ) ,
69+ msg =>Poll :: Ready ( Some ( msg) ) ,
70+ }
71+ }
72+ }
73+
6274/// A cache of type info and prepared statements for fetching type info
6375/// (corresponding to the queries in the [prepare](prepare) module).
6476#[ derive( Default ) ]