Set exception on CamelExchange if message conversion fails

- re #37
krasserm · Jun 18, 2017 · 6c5a869 · 6c5a869
1 parent 5ae4d75
commit 6c5a869
Showing 1 changed file with 5 additions and 2 deletions.
diff --git a/streamz-camel-fs2/src/main/scala/streamz/camel/fs2/dsl/package.scala b/streamz-camel-fs2/src/main/scala/streamz/camel/fs2/dsl/package.scala
@@ -148,8 +148,11 @@ package object dsl {
             context.consumerTemplate.doneUoW(ce)
           case Success(ce) =>
             Try(StreamMessage.from[A](ce.getIn)) match {
-              case Success(m) => callback(Right(m))
-              case Failure(e) => callback(Left(e))
+              case Success(m) =>
+                callback(Right(m))
+              case Failure(e) =>
+                callback(Left(e))
+                ce.setException(e)
             }
             context.consumerTemplate.doneUoW(ce)
           case Failure(ex) =>