flatmap implemented
to_stream