[streaming] refactor internal states