How to use a `Stream` object inside the closure of `for_each` future?

Question

I have a structure which receives some websocket events. Here is a method to receive an event from the structure:

struct Connection;
impl Connection {
    pub fn recv_event(&mut self) -> Result<Event> {
        // ...
    }
}

I have implemented Stream for it:

#[cfg(feature = "stream")]
impl Stream for Connection {
    type Item = Event;
    type Error = ::Error;

    fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
        match self.recv_event() {
            Ok(event) => Ok(Async::Ready(Some(event))),
            Err(e) => Err(e),
        }
    }
}

It looks that it works okay, but later I decided to return a reference to Self also for being able to modify the object's state in this code:

connection.by_ref().for_each(|c, e| {
    println!("Event: {:?}", e);
    self.handle_events(c, e.clone());
    self.handle_messages(e);
    Ok(())
}).wait();

I need to return a reference to the connection object from the fn poll:

#[cfg(feature = "stream")]
impl Stream for Connection {
    type Item = (&mut Self, Event);
    type Error = ::Error;

    fn poll<'a>(&'a mut self) -> Poll<Option<(&'a mut Self, Event)>, Self::Error> {
        match self.recv_event() {
            Ok(event) => Ok(Async::Ready(Some(self, event))),
            Err(e) => Err(e),
        }
    }
}

Unfortunately, this does not work:

error[E0106]: missing lifetime specifier
   --> src/connection.rs:410:18
    |
410 |     type Item = (&mut Self, Event);
    |                  ^ expected lifetime parameter

Yes, I forgot to add a lifetime here, in the type. Oops. How can I do that? I tried that and some other ways but nothing worked:

#[cfg(feature = "stream")]
impl<'a> Stream for Connection {
    type Item = (&'a mut Self, Event);
    type Error = ::Error;

    fn poll<'a>(&'a mut self) -> Poll<Option<(&'a mut Self, Event)>, Self::Error> {
        match self.recv_event() {
            Ok(event) => Ok(Async::Ready(Some(self, event))),
            Err(e) => Err(e),
        }
    }
}

Which gives the error:

error[E0207]: the lifetime parameter `'a` is not constrained by the impl trait, self type, or predicates
   --> src/connection.rs:409:6
    |
409 | impl<'a> Stream for Connection {
    |      ^^ unconstrained lifetime parameter

Then I tried:

pub struct ConnectionStream<'a>(&'a mut Connection);
#[cfg(feature = "stream")]
impl<'a> Stream for ConnectionStream<'a> {
    type Item = (&'a mut Connection, Event);
    type Error = ::Error;

    fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
        let res = self.0.recv_event();
        match res {
            Ok(event) => Ok(Async::Ready(Some((self.0, event)))),
            Err(e) => Err(e),
        }
    }
}

But still no luck:

error[E0495]: cannot infer an appropriate lifetime for automatic coercion due to conflicting requirements
   --> src/connection.rs:417:42
    |
417 |             Ok(event) => Ok(Async::Ready(Some((self.0, event)))),
    |                                          ^^^^^^^^^^^^^^^^^^^^^
    |
note: first, the lifetime cannot outlive the anonymous lifetime #1 defined on the body at 414:64...
   --> src/connection.rs:414:65
    |
414 |       fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
    |  _________________________________________________________________^ starting here...
415 | |         let res = self.0.recv_event();
416 | |         match res {
417 | |             Ok(event) => Ok(Async::Ready(Some((self.0, event)))),
418 | |             Err(e) => Err(e),
419 | |         }
420 | |     }
    | |_____^ ...ending here
note: ...so that reference does not outlive borrowed content
   --> src/connection.rs:417:48
    |
417 |             Ok(event) => Ok(Async::Ready(Some((self.0, event)))),
    |                                                ^^^^^^
note: but, the lifetime must be valid for the lifetime 'a as defined on the body at 414:64...
   --> src/connection.rs:414:65
    |
414 |       fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
    |  _________________________________________________________________^ starting here...
415 | |         let res = self.0.recv_event();
416 | |         match res {
417 | |             Ok(event) => Ok(Async::Ready(Some((self.0, event)))),
418 | |             Err(e) => Err(e),
419 | |         }
420 | |     }
    | |_____^ ...ending here
note: ...so that types are compatible (expected futures::Stream, found futures::Stream)
   --> src/connection.rs:414:65
    |
414 |       fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
    |  _________________________________________________________________^ starting here...
415 | |         let res = self.0.recv_event();
416 | |         match res {
417 | |             Ok(event) => Ok(Async::Ready(Some((self.0, event)))),
418 | |             Err(e) => Err(e),
419 | |         }
420 | |     }
    | |_____^ ...ending here

I gave up and I think it is impossible. Am I correct?