refactored stream to mongo persistence