stream: properly implement shutdown