package com.connorpenhale.examples.scala

import org.apache.camel.CamelContext
import org.apache.camel.Exchange
import org.apache.camel.scala.dsl.builder.ScalaRouteBuilder
import org.apache.camel.processor.aggregate.GroupedExchangeAggregationStrategy

/**
 * A Camel Router using the Scala DSL
 */
class MyRouteBuilder(override val context: CamelContext) extends ScalaRouteBuilder(context) {

  "file://./bigFiles/?fileName=100TBFile.txt" ==> {
    split(_.in[String].toArray).parallelProcessing {
      process { e =>
        if (e.getIn.getBody(classOf[String]).matches("\\ ")) {
          e.getIn.setBody("space")
        }
        if (e.getIn.getBody(classOf[String]).matches("\\n")) {
          e.getIn.setBody("newline")
        }
        if (e.getIn.getBody(classOf[String]).matches("\\r")) {
          e.getIn.setBody("return")
        }
      }
      setHeader("character", e => e.getIn.getBody(classOf[String]))
      aggregate(e => e.getIn.getHeader("character", classOf[String]), new GroupedExchangeAggregationStrategy()).completionTimeout(60000) {
        process { e =>
          val character = e.getIn.getBody(classOf[List[Exchange]])(0).getIn.getBody(classOf[String])
          e.getIn.setHeader("character", character)
        }
        log("Aggregate ${in.header.character} ${in.body.size}")
      }
    }
  }

}