rx (feature): Add Rx.trasnformFailure for error handling (#3214)

Author: xerial

airframe-rx/src/main/scala/wvlet/airframe/rx/Rx.scala

@@ -312,6 +312,12 @@ trait Rx[+A] extends RxOps[A] {
     TransformRxOp(this, f)
   }
 
+  /**
+    * Transform a Success(v) or Failure(Throwable) input with a given function.
+    * @param f
+    * @tparam B
+    * @return
+    */
   def transform[B](f: Try[A] => B): Rx[B] = {
     TransformOp(this, f)
   }
@@ -324,6 +330,18 @@ trait Rx[+A] extends RxOps[A] {
     TransformTryOp(this, f)
   }
 
+  /**
+    * Transform a specific type of an exception into another exception. This is useful for handling exceptions.
+    * @param f
+    * @return
+    */
+  def transformFailure(f: PartialFunction[Throwable, Throwable]): Rx[A] = {
+    transformTry[A] {
+      case Failure(ex) if f.isDefinedAt(ex) => Failure(f.apply(ex))
+      case other                            => other
+    }
+  }
+
   def concat[A1 >: A](other: Rx[A1]): Rx[A1] = Rx.concat(this, other)
   def lastOption: RxOption[A]                = LastOp(this).toOption
 

airframe-rx/src/test/scala/wvlet/airframe/rx/RxTransformTest.scala

@@ -95,4 +95,17 @@ class RxTransformTest extends AirSpec {
           fail(s"Unexpected: ${other}")
       }
   }
+
+  test("Rx.transformFailure") {
+    Rx.fromTry(Failure(new UnsupportedOperationException("N/A")))
+      .transformFailure { case ex: UnsupportedOperationException =>
+        new IllegalStateException("unimplemented")
+      }
+      .recover {
+        case e: IllegalStateException =>
+          "ok"
+        case other =>
+          fail(s"Unexpected: ${other}")
+      }
+  }
 }