Converts a MongoDB observable to a ZIO stream

ZioMongoSupport.scala

import org.mongodb.scala.{Observable, Observer}
import zio.stream._
import zio.{IO, ZIO}

object ZioMongoSupport {

  implicit class ObsConverter[T](observable: Observable[T]) {

    def toStream: Stream[Throwable, T] = toStream(identity)

    def toStream[E](handleError: Throwable => E): Stream[E, T] =
      Stream.effectAsync { f =>
        observable.subscribe {
          new Observer[T] {
            override def onNext(result: T): Unit = f(ZIO.succeed(result))

            override def onError(e: Throwable): Unit = f(ZIO.fail(Some(handleError(e))))

            override def onComplete(): Unit = f(ZIO.fail(None))
          }
        }
      }
  }
}