Massive changes to support async streams.