added process stream function and emitting end