Making stream.connection.end() work for both collectors and subscriptions.