Migrate to Camel 4.x

Author: pbernet

build.sbt

@@ -101,10 +101,11 @@ libraryDependencies ++= Seq(
   "org.apache.commons" % "commons-lang3" % "3.18.0",
   "com.sksamuel.avro4s" %% "avro4s-core" % "4.1.2", // 5.x for Scala 3
 
-  "org.apache.camel" % "camel-core" % "3.20.2",
-  "org.apache.camel" % "camel-reactive-streams" % "3.20.2",
-  "io.projectreactor" % "reactor-core" % "3.7.11",
-  "io.reactivex.rxjava3" % "rxjava" % "3.1.11",
+  "org.apache.camel" % "camel-core" % "4.16.0",
+  "org.apache.camel" % "camel-seda" % "4.16.0",
+  "org.apache.camel" % "camel-reactive-streams" % "4.16.0",
+  "io.projectreactor" % "reactor-core" % "3.8.0",
+  "io.reactivex.rxjava3" % "rxjava" % "3.1.12",
 
   "com.github.blemale" %% "scaffeine" % "5.3.0",
   "ch.qos.logback" % "logback-classic" % "1.5.18",

src/main/scala/interop/ReactiveStreamsInterop.scala

@@ -23,9 +23,13 @@ import scala.concurrent.duration.DurationInt
   *  - RxJava
   *  - pekko-streams
   *
+  * Remarks:
+  *  - SEDA (Staged Event-Driven Architecture)  provides asynchronous in-memory messaging within a single CamelContext
+  *
   * Doc:
   * https://doc.akka.io/docs/akka/current/stream/reactive-streams-interop.html
-  * https://camel.apache.org/components/3.18.x/reactive-streams-component.html
+  * https://camel.apache.org/components/4.14.x/reactive-streams-component.html
+  * https://camel.apache.org/components/4.14.x/seda-component.html
   * https://projectreactor.io/docs/core/release/reference/
   * https://github.com/ReactiveX/RxJava
   */
@@ -40,8 +44,8 @@ object ReactiveStreamsInterop extends App {
   val rsCamel: CamelReactiveStreamsService = CamelReactiveStreams.get(camel)
   camel.start()
 
-  // Consumer endpoint with Camel
-  val publisher: Publisher[String] = rsCamel.from("vm:words", classOf[String])
+  // Producer/Publisher from Camel SEDA queue using Reactive Streams
+  val publisher: Publisher[String] = rsCamel.from("seda:words", classOf[String])
 
   // Slow consumer with Reactor 3
   Flux.from(publisher)
@@ -64,15 +68,14 @@ object ReactiveStreamsInterop extends App {
     .wireTap(each => logger.info(s"Consumed with pekko-streams: $each"))
     .runWith(Sink.ignore)
 
-  // Sender endpoint with Camel
-  val template: FluentProducerTemplate = camel.createFluentProducerTemplate
+  val producerTemplate: FluentProducerTemplate = camel.createFluentProducerTemplate
 
   Source(1 to 10)
     .throttle(1, 1.seconds, 1, ThrottleMode.shaping)
     .mapAsync(1) { i =>
-      template
+      producerTemplate
         .withBody(s"Camel$i")
-        .to("vm:words")
+        .to("seda:words")
         .send
       Future(i)
     }.runWith(Sink.ignore)