Tengo un filtro para filtrar elementos incorrectos en una secuencia. En algunos casos extremos, esto podría dar lugar a que se filtren todos los elementos. Cuando esto sucede, la transmisión falla con un error: java.util.NoSuchElementException: reduce over empty stream al ejecutar reduce.
¿Cómo manejar este caso para devolver una respuesta significativa?
He intentado supervisión como -
val decider: Supervision.Decider = { case _ => Supervision.Stop case e : NoSuchElementException => Supervision.stop } RunnableGraph. toMat(Sink.reduce[Int](_ + _) .withAttributes(ActorAttributes.supervisionStrategy(decider)))(Keep.both) .run() También intenté recover , pero nada parece funcionar.
Necesito manejar este caso para devolver una respuesta significativa.
Cualquier ayuda será apreciada.
Solo porque está usando Sink.reduce[Int] , podría agregar una Source que garantice tener un solo elemento 0 y, por lo tanto, Sink.reduce[Int] funcionará y producirá 0 como resultado.
aquí hay un ejemplo
val zero = Source.single(0) val possiblyEmpty = Source(List[Int](1, 3, 5)).filter(_ % 2 == 0) val eventualInt = zero.merge(possiblyEmpty).toMat(Sink.reduce[Int](_ + _))(Keep.right).run()Puede valer la pena considerar usar Sink.fold en lugar de Sink.reduce :
val possiblyEmpty = Source(Seq(1, 3, 5)).filter(_ % 2 == 0) val eventualInt = possiblyEmpty.toMat(Sink.fold[Int](0)(_ + _))(Keep.right).run()Si no hay un elemento de identidad/cero razonable, puede tener algo un poco más generalizado en este sentido:
def reducePossiblyEmpty[T](source: Source[T])(f: (T, T) => T): RunnableGraph[Future[Option[T]]] = { val lifted = { (x: Option[T], y: Option[T]) => x.flatMap(a => y.map(f)) } source.map(x => Some(x)) .concat(Source.single(None)) .statefulMapConcat[Option[T]] { () => var emptyStream = true { x => x match { case Some(x) => // element from the given stream emptyStream = false List(x) case None => // given stream completed if (emptyStream) { List(x) } else { Nil // don't emit anything } } } } .toMat(Sink.reduce[Option[T]](lifted))(Keep.right) } El gráfico devuelto se completará con None si no hubiera elementos, o con Some del resultado de la reducción.
EDITAR: también puede usar orElse en Source / Flow , en lugar de .concat.statefulMapConcat en lo anterior:
def reducePossiblyEmpty[T](source: Source[T])(f: (T, T) => T): RunnableGraph[Future[Option[T]]] = { val lifted = { (x: Option[T], y: Option[T]) => x.flatMap(a => y.map(f)) } source.map(x => Some(x)) .orElse(Source.single(None)) .toMat(Sink.reduce[Option[T]](lifted))(Keep.right) }