stream API with highland