[streaming] refactor internal states

Sorry the previous commitment commits to the wrong branch. Please adopt this pull request.