Add EventStream.mapError

[ajaychandran]

This can be implemented using:

final class MapErrorEventStream[A](val parent: Observable[A], f: PartialFunction[Throwable, Throwable])
    extends EventStream[A]
    with SingleParentObservable[A, A]
    with InternalNextErrorObserver[A] {

  protected[airstream] val topoRank: Int = parent.topoRank + 1

  protected[airstream] def onNext(nextValue: A, transaction: Transaction): Unit =
    fireValue(nextValue, transaction)

  protected[airstream] def onError(nextError: Throwable, transaction: Transaction): Unit =
    try fireError(f.applyOrElse(nextError, identity[Throwable]), transaction)
    catch { case NonFatal(t) => fireError(AirstreamError.ObserverErrorHandlingError(t, nextError), transaction) }
}

This is useful when using Ajax streams with services that can return specific errors.

[raquo]

mapError would need to accept a total function. I guess we can call this collectError.

[ajaychandran]

+1 on total function (since it would support both mapError and collectError).

[raquo]

Do you mean you won't need collectError that accepts a PF if mapError accepts a total function?

…

--Nikita

On Sat., May 1, 2021, 9:37 p.m. Ajay Chandran, ***@***.***> wrote: +1 on total function (since it would support both mapError and collectError). — You are receiving this because you commented. Reply to this email directly, view it on GitHub <#78 (comment)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AAECBMC2W56S7XSD6LRFGZDTLTJHDANCNFSM42PEMKYA> .

[ajaychandran]

I meant both combinators can be provided.

def mapError(f: Throwable => Throwable)
def collectError(pf: PartialFunction[Throwable, Throwable]) = mapError(pf.applyOrElse(_, identity[Throwable]))