That poll() implementation doesn’t look correct. If you poll the receiver and get a Async::Ready(Some), you should not return Async::NotReady. If you do, the future will not be polled again. You need to make sure the underlying future (or stream), which is the receiver here, registers itself with the reactor to be notified when it has more data to process. It’ll only do that if it returns NotReady - that’s the general contract of polling futures.

The easiest way to do that is to loop until you get a NotReady, and then return NotReady from State.

You should also consider Ready(None) to mean the underlying stream is done, and your future should resolve at this point; you’re returning NotReady, which isn’t right.

I don’t think this explains the premature termination you’re seeing but it’s an issue nonetheless. If the stream is ending too early, then the sender is being dropped somewhere (or the stream ends with an error). It’s a bit hard to pinpoint via the snippets you’ve posted since the whole picture has to be seen, but that’s something you should investigate.