fix OnEachStream