Author: gertv
Date: Thu Jul 24 00:31:38 2008
New Revision: 679291
URL: http://svn.apache.org/viewvc?rev=679291&view=rev
Log:
CAMEL-463: Adding support for a message aggregator
Added:
activemq/camel/trunk/components/camel-scala/src/main/scala/org/apache/camel/scala/dsl/SAggregatorType.scala
activemq/camel/trunk/components/camel-scala/src/test/scala/org/apache/camel/scala/dsl/AggregatorTest.scala
Modified:
activemq/camel/trunk/components/camel-scala/src/main/scala/org/apache/camel/scala/dsl/DSL.scala
activemq/camel/trunk/components/camel-scala/src/main/scala/org/apache/camel/scala/dsl/SAbstractType.scala
activemq/camel/trunk/components/camel-scala/src/main/scala/org/apache/camel/scala/dsl/builder/RouteBuilder.scala
Modified:
activemq/camel/trunk/components/camel-scala/src/main/scala/org/apache/camel/scala/dsl/DSL.scala
URL:
http://svn.apache.org/viewvc/activemq/camel/trunk/components/camel-scala/src/main/scala/org/apache/camel/scala/dsl/DSL.scala?rev=679291&r1=679290&r2=679291&view=diff
==============================================================================
---
activemq/camel/trunk/components/camel-scala/src/main/scala/org/apache/camel/scala/dsl/DSL.scala
(original)
+++
activemq/camel/trunk/components/camel-scala/src/main/scala/org/apache/camel/scala/dsl/DSL.scala
Thu Jul 24 00:31:38 2008
@@ -35,5 +35,6 @@
def loadbalance : SLoadBalanceType
def delay(delay: Period) : SDelayerType
def resequence(expression: Exchange => Any) : SResequencerType
+ def aggregate(expression: Exchange => Any) : SAggregatorType
}
Modified:
activemq/camel/trunk/components/camel-scala/src/main/scala/org/apache/camel/scala/dsl/SAbstractType.scala
URL:
http://svn.apache.org/viewvc/activemq/camel/trunk/components/camel-scala/src/main/scala/org/apache/camel/scala/dsl/SAbstractType.scala?rev=679291&r1=679290&r2=679291&view=diff
==============================================================================
---
activemq/camel/trunk/components/camel-scala/src/main/scala/org/apache/camel/scala/dsl/SAbstractType.scala
(original)
+++
activemq/camel/trunk/components/camel-scala/src/main/scala/org/apache/camel/scala/dsl/SAbstractType.scala
Thu Jul 24 00:31:38 2008
@@ -1,4 +1,4 @@
-/**
+/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
@@ -78,5 +78,7 @@
def delay(period: Period) = new
SDelayerType(target.delayer(period.milliseconds))
def resequence(expression: Exchange => Any) = new
SResequencerType(target.resequencer(expression))
+
+ def aggregate(expression: Exchange => Any) = new
SAggregatorType(target.aggregator(expression))
}
Added:
activemq/camel/trunk/components/camel-scala/src/main/scala/org/apache/camel/scala/dsl/SAggregatorType.scala
URL:
http://svn.apache.org/viewvc/activemq/camel/trunk/components/camel-scala/src/main/scala/org/apache/camel/scala/dsl/SAggregatorType.scala?rev=679291&view=auto
==============================================================================
---
activemq/camel/trunk/components/camel-scala/src/main/scala/org/apache/camel/scala/dsl/SAggregatorType.scala
(added)
+++
activemq/camel/trunk/components/camel-scala/src/main/scala/org/apache/camel/scala/dsl/SAggregatorType.scala
Thu Jul 24 00:31:38 2008
@@ -0,0 +1,56 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.scala.dsl;
+
+import org.apache.camel.model.AggregatorType
+import org.apache.camel.processor.aggregate.AggregationStrategy
+import org.apache.camel.model.config.BatchResequencerConfig
+import org.apache.camel.scala.dsl.builder.RouteBuilder
+
+/**
+ * Scala wrapper for Camel AggregatorType
+ */
+class SAggregatorType(val target: AggregatorType)(implicit val builder:
RouteBuilder) extends SAbstractType with Wrapper[AggregatorType] {
+
+ val unwrap = target
+
+ def strategy(function: (Exchange, Exchange) => Exchange) = {
+ println("testing")
+ target.setAggregationStrategy(
+ new AggregationStrategy() {
+ def aggregate(oldExchange: Exchange, newExchange: Exchange) ={
+ println(oldExchange + " + " + newExchange)
+ try {
+ val result = function(oldExchange, newExchange)
+ println(" -> " + result)
+ result
+ } catch {
+ case e:Exception => println(e); e.printStackTrace()
+ }
+ null
+ }
+ }
+ )
+ this
+ }
+
+ def batch(count: Int) = {
+ target.batchSize(count)
+ this
+ }
+
+}
Modified:
activemq/camel/trunk/components/camel-scala/src/main/scala/org/apache/camel/scala/dsl/builder/RouteBuilder.scala
URL:
http://svn.apache.org/viewvc/activemq/camel/trunk/components/camel-scala/src/main/scala/org/apache/camel/scala/dsl/builder/RouteBuilder.scala?rev=679291&r1=679290&r2=679291&view=diff
==============================================================================
---
activemq/camel/trunk/components/camel-scala/src/main/scala/org/apache/camel/scala/dsl/builder/RouteBuilder.scala
(original)
+++
activemq/camel/trunk/components/camel-scala/src/main/scala/org/apache/camel/scala/dsl/builder/RouteBuilder.scala
Thu Jul 24 00:31:38 2008
@@ -64,5 +64,6 @@
def loadbalance = stack.top.loadbalance
def delay(delay: Period) = stack.top.delay(delay)
def resequence(expression: Exchange => Any) =
stack.top.resequence(expression)
+ def aggregate(expression: Exchange => Any) = stack.top.aggregate(expression)
}
Added:
activemq/camel/trunk/components/camel-scala/src/test/scala/org/apache/camel/scala/dsl/AggregatorTest.scala
URL:
http://svn.apache.org/viewvc/activemq/camel/trunk/components/camel-scala/src/test/scala/org/apache/camel/scala/dsl/AggregatorTest.scala?rev=679291&view=auto
==============================================================================
---
activemq/camel/trunk/components/camel-scala/src/test/scala/org/apache/camel/scala/dsl/AggregatorTest.scala
(added)
+++
activemq/camel/trunk/components/camel-scala/src/test/scala/org/apache/camel/scala/dsl/AggregatorTest.scala
Thu Jul 24 00:31:38 2008
@@ -0,0 +1,61 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.scala.dsl;
+
+import org.w3c.dom.Document
+import scala.dsl.builder.RouteBuilder
+
+/**
+ * Test case for message aggregator
+ */
+class AggregatorTest extends ScalaTestSupport {
+
+ val count = 100
+
+ def testSimpleAggregator = {
+ "mock:a" expect { _.received("message " + count) }
+ test {
+ for (i <- 1 to count) {
+ "direct:a" ! ("message " + i)
+ }
+ }
+ }
+
+ def testBlockAggregator = {
+ "mock:b" expect { _.received("message " + count) }
+ test {
+ for (i <- 1 to count) {
+ "direct:b" ! ("message " + i)
+ }
+ }
+ }
+
+ val builder =
+ new RouteBuilder {
+ //START SNIPPET: simple
+ "direct:a" aggregate (_.in(classOf[String]).substring(0, 7)) to "mock:a"
+ //END SNIPPET: simple
+
+ //START SNIPPET: block
+ "direct:b" ==> {
+ aggregate(_.in(classOf[String]).substring(0,7)) {
+ to ("mock:b")
+ }
+ }
+ //END SNIPPET: block
+ }
+}
