Support backpressure (albeit still old-style)

This pull request is needed for streams that need to provide backpressure via the old-style pause/resume interface.

Without this PR, pulling in millions of salesforce records will cause the use of streams to blow up in memory and often crash out of memory usage, which was a disaster for us. This keeps the memory nice and steady.