Merge pull request #8 from ebenfield/INTSCALA-41-v4.0

Intscala 41 v4.0

settings.gradle

@@ -1,10 +1,11 @@
-rootProject.name = 'spring-integration-dsl-scala'
-
-include 'spring-integration-dsl-scala-core'
-include 'spring-integration-dsl-scala-http'
-include 'spring-integration-dsl-scala-jms'
-include 'spring-integration-dsl-scala-gemfire'
-include 'spring-integration-dsl-scala-file'
-include 'spring-integration-dsl-scala-ftp'
-include 'spring-integration-dsl-scala-sftp'
+rootProject.name = 'spring-integration-dsl-scala'
+
+include 'spring-integration-dsl-scala-core'
+include 'spring-integration-dsl-scala-http'
+include 'spring-integration-dsl-scala-jms'
+include 'spring-integration-dsl-scala-gemfire'
+include 'spring-integration-dsl-scala-file'
+include 'spring-integration-dsl-scala-jdbc'
+include 'spring-integration-dsl-scala-ftp'
+include 'spring-integration-dsl-scala-sftp'
 include 'spring-integration-dsl-scala-integration-demos'

spring-integration-dsl-scala-core/src/main/scala/org/springframework/integration/dsl/BaseIntegrationComposition.scala

@@ -16,20 +16,16 @@
 package org.springframework.integration.dsl
 import java.util.UUID
 import org.apache.commons.logging.LogFactory
-import org.springframework.beans.factory.support.BeanDefinitionBuilder
 import org.springframework.integration.dsl.utils.DslUtils
-import org.springframework.beans.factory.support.BeanDefinitionRegistry
-import org.springframework.context.ApplicationContext
+import org.springframework.context.{ApplicationContextAware, ApplicationContext}
 import org.w3c.dom.Element
 import org.w3c.dom.Document
-import org.springframework.context.ConfigurableApplicationContext
-import org.springframework.beans.factory.config.ConfigurableBeanFactory
 
 /**
  * @author Oleg Zhurakousky
  */
 private[dsl] case class BaseIntegrationComposition(private[dsl] val parentComposition: BaseIntegrationComposition,
-                                                   private[dsl] val target: IntegrationComponent) {
+                                                   private[dsl] val target: IntegrationComponent) extends AnyRef with ApplicationContextAware {
 
   val logger = LogFactory.getLog(this.getClass());
 
@@ -138,6 +134,17 @@ private[dsl] case class BaseIntegrationComposition(private[dsl] val parentCompos
       }
     }
   }
+
+  /**
+   * Lifecycle method
+   */
+  def setApplicationContext(applicationContext: ApplicationContext)  {
+
+    if (this.logger.isDebugEnabled())
+      this.logger.debug("Getting IntegrationContext for: " + this)
+
+    this.getContext(applicationContext).start()
+  }
 }
 
 /**
@@ -207,7 +214,7 @@ private[dsl] class ListeningIntegrationComposition(parentComposition: BaseIntegr
    */
   def start(parentContext:ApplicationContext = null) = this.getContext(parentContext).start
 
-  def start = this.getContext(null).start
+  def start = this.getContext(null)
 
   def stop = this.getContext(null).stop
 

spring-integration-dsl-scala-core/src/main/scala/org/springframework/integration/dsl/IntegrationContext.scala

@@ -23,14 +23,14 @@ import scala.collection.JavaConversions
 import org.apache.commons.logging.LogFactory
 import org.springframework.context.ApplicationContext
 import org.springframework.integration.channel.QueueChannel
-import org.springframework.integration.dsl.utils.DslUtils
+import utils.{IntegrationDomTreeBuilder, DslUtils}
 import org.springframework.integration.support.MessageBuilder
 import org.springframework.integration.Message
 import org.springframework.integration.MessageChannel
 import org.springframework.integration.MessagingException
 import org.springframework.util.CollectionUtils
 import org.springframework.util.StringUtils
-import org.springframework.integration.dsl.utils.IntegrationDomTreeBuilder
+import org.springframework.context.support.GenericApplicationContext
 
 /**
  * @author Oleg Zhurakousky
@@ -66,7 +66,7 @@ private[dsl] class IntegrationContext(parentContext: ApplicationContext, composi
 
 //  val applicationContext = ApplicationContextBuilder.build(parentContext, normalizedComposition)
 
-  val applicationContext = IntegrationDomTreeBuilder.buildApplicationContext(normalizedComposition)
+  val applicationContext = IntegrationDomTreeBuilder.buildApplicationContext(parentContext, normalizedComposition)
 
   def start() = this.applicationContext.start()
 

spring-integration-dsl-scala-core/src/main/scala/org/springframework/integration/dsl/utils/IntegrationDomTreeBuilder.scala

@@ -17,14 +17,12 @@ package org.springframework.integration.dsl.utils
 import java.io.ByteArrayInputStream
 import java.io.ByteArrayOutputStream
 import java.io.StringWriter
-import java.util.HashMap
 import java.util.UUID
 import org.apache.commons.logging.LogFactory
 import org.springframework.beans.factory.xml.XmlBeanDefinitionReader
 import org.springframework.context.support.GenericApplicationContext
 import org.springframework.context.ApplicationContext
 import org.springframework.core.io.InputStreamResource
-import org.springframework.integration.dsl.AbstractChannel
 import org.springframework.integration.dsl.BaseIntegrationComposition
 import org.springframework.integration.dsl.Channel
 import org.springframework.integration.dsl.FunctionInvoker
@@ -44,10 +42,9 @@ import javax.xml.transform.stream.StreamResult
 import javax.xml.transform.OutputKeys
 import javax.xml.transform.TransformerFactory
 import org.springframework.context.ConfigurableApplicationContext
-import org.springframework.integration.dsl.Router
-import org.springframework.integration.dsl.OutboundAdapterEndpoint
 import org.springframework.beans.factory.BeanFactoryAware
 import org.springframework.integration.dsl.AbstractChannel
+
 /**
  * @author Oleg Zhurakousky
  */
@@ -57,7 +54,10 @@ object IntegrationDomTreeBuilder {
   }
 
   def buildApplicationContext[T <: BaseIntegrationComposition](integrationComposition: T) =
-    new IntegrationDomTreeBuilder().buildApplicationContext(integrationComposition)
+    new IntegrationDomTreeBuilder().buildApplicationContext(null, integrationComposition)
+
+  def buildApplicationContext[T <: BaseIntegrationComposition](applicationContext: ApplicationContext = null, integrationComposition: T) =
+    new IntegrationDomTreeBuilder().buildApplicationContext(applicationContext, integrationComposition)
 }
 /**
  *
@@ -90,9 +90,10 @@ class IntegrationDomTreeBuilder {
   /**
    *
    */
-  def buildApplicationContext[T <: BaseIntegrationComposition](integrationComposition: T): ConfigurableApplicationContext = {
+  def buildApplicationContext[T <: BaseIntegrationComposition](parentContext: ApplicationContext, integrationComposition: T): ConfigurableApplicationContext = {
     this.toDocument(integrationComposition)
-    val applicationContext = new GenericApplicationContext()
+
+    val applicationContext = new GenericApplicationContext(parentContext)
 
     this.supportingBeans.foreach { element: Tuple2[String, Any] =>
       applicationContext.getBeanFactory().registerSingleton(element._1, element._2)

spring-integration-dsl-scala-core/src/test/resources/parent-config.xml

@@ -0,0 +1,19 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+       xsi:schemaLocation="
+http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.0.xsd">
+
+    <!-- <bean id="simpleService" class="org.springframework.integration.dsl.SimpleService"/>-->
+
+    <!--<bean id="messageFlow"
+          class="org.springframework.integration.dsl.messageFlowFactory"
+          factory-method="createFlow">
+    </bean>-->
+
+    <bean id="messageFlow"
+          class="org.springframework.integration.dsl.scalaDslFlowFactory"
+          factory-method="createFlow">
+    </bean>
+
+</beans>

spring-integration-dsl-scala-core/src/test/scala/org/springframework/integration/dsl/ParentApplicationContextTests.scala

@@ -17,18 +17,33 @@ package org.springframework.integration.dsl
 import org.junit.Test
 import org.springframework.context.support.ClassPathXmlApplicationContext
 import org.junit.Assert
+import org.springframework.beans.factory.BeanFactoryAware
+
 /**
- * @author Oleg Zhurakousky
+ * @author Oleg Zhurakousky / Ewan Benfield
  */
 class ParentApplicationContextTests {
 
   @Test
   def validateInitializationOfParentAc = {
-//    val context = new ClassPathXmlApplicationContext("parent-config.xml")
-//    
-//    val messageFlow = handle.using(context.getBean("simpleService"))
-//   
-//    Assert.assertEquals(context.getBean("simpleService"), messageFlow.getContext(context).applicationContext.getBean("simpleService"))
+    val context = new ClassPathXmlApplicationContext("parent-config.xml")
+
+    val messageFlow:BaseIntegrationComposition = context.getBean("messageFlow")  match {
+      case elt:BaseIntegrationComposition => elt
+      case _ => throw new ClassCastException
+    }
+    Assert.assertEquals(context.getBean("messageFlow"), messageFlow.getContext(context).applicationContext.getBean("messageFlow"))
   }
+  object scalaDslFlowFactory {
+
+    def createFlow(): BaseIntegrationComposition = {
 
+      val messageFlow =
+        filter {payload: String => payload == "World"} -->
+          transform { payload: String => "Hello " + payload} -->
+          handle { payload: String => println(payload) }
+
+      messageFlow
+    }
+  }
 }

spring-integration-dsl-scala-core/src/test/scala/org/springframework/integration/dsl/ScalaDslFlowFactory.scala

@@ -0,0 +1,39 @@
+/*
+ * Copyright 2002-2012 the original author or authors.
+ *
+ * Licensed 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.springframework.integration.dsl
+
+
+/**
+ * @author Ewan Benfield
+ */
+object scalaDslFlowFactory {
+
+  def createFlow(): BaseIntegrationComposition = {
+
+    val messageFlow =
+      filter {
+        payload: String => payload == "World"
+      } -->
+        transform {
+          payload: String => "Hello " + payload
+        } -->
+        handle {
+          payload: String => println(payload)
+        }
+
+    messageFlow
+  }
+}

spring-integration-dsl-scala-jdbc/src/main/scala/org/springframework/integration/dsl/JdbcDsl.scala

@@ -0,0 +1,90 @@
+/*
+ * Copyright 2002-2012 the original author or authors.
+ *
+ * Licensed 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.springframework.integration.dsl
+import java.util.concurrent.Executor
+import javax.sql.DataSource
+
+/**
+ * @author Ewan Benfield
+ */
+
+
+private[dsl] object JdbcDsl {
+  val jdbcSchema = " http://www.springframework.org/schema/integration/jdbc " +
+    "http://www.springframework.org/schema/integration/jdbc/spring-integration-jdbc.xsd"
+
+}
+
+//TODO Decide on appropriate API here
+object jdbc {
+  def poll(dataSource: DataSource) = new {
+
+    def atFixedRate(query: String, rate: Int): ListeningIntegrationComposition = {
+      val poller = new Poller(fixedRate = rate)
+      new ListeningIntegrationComposition(null, new JdbcInboundAdapterConfig(target = query, poller = poller, dataSource = dataSource)) {
+
+        def withMaxMessagesPerPoll(maxMessagesPerPoll: Int): ListeningIntegrationComposition = {
+          val poller = new Poller(fixedRate = rate, maxMessagesPerPoll = maxMessagesPerPoll)
+          new ListeningIntegrationComposition(null, new JdbcInboundAdapterConfig(target = query, poller = poller, dataSource = dataSource)) {
+
+            def withTaskExecutor(taskExecutor: Executor): ListeningIntegrationComposition = {
+              val poller = new Poller(fixedRate = rate, maxMessagesPerPoll = maxMessagesPerPoll, taskExecutor = taskExecutor)
+              new ListeningIntegrationComposition(null, new JdbcInboundAdapterConfig(target = query, poller = poller, dataSource = dataSource))
+            }
+          }
+        }
+
+        def withTaskExecutor(taskExecutor: Executor): ListeningIntegrationComposition = {
+          val poller = new Poller(fixedRate = rate, taskExecutor = taskExecutor)
+          new ListeningIntegrationComposition(null, new JdbcInboundAdapterConfig(target = query, poller = poller, dataSource = dataSource))
+        }
+
+      }
+
+    }
+
+    def withFixedDelay(query: String, delay: Int) = {
+      val poller = new Poller(fixedDelay = delay)
+      new ListeningIntegrationComposition(null, new JdbcInboundAdapterConfig(target= query, poller = poller, dataSource = dataSource)) {
+
+        def withMaxMessagesPerPoll(maxMessagesPerPoll: Int): ListeningIntegrationComposition = {
+          val poller = new Poller(fixedDelay = delay, maxMessagesPerPoll = maxMessagesPerPoll)
+          new ListeningIntegrationComposition(null, new JdbcInboundAdapterConfig(target= query, poller = poller, dataSource = dataSource)) {
+
+            def withTaskExecutor(taskExecutor: Executor): ListeningIntegrationComposition = {
+              val poller = new Poller(fixedDelay = delay, maxMessagesPerPoll = maxMessagesPerPoll, taskExecutor = taskExecutor)
+              new ListeningIntegrationComposition(null, new JdbcInboundAdapterConfig(target= query, poller = poller, dataSource = dataSource))
+            }
+          }
+        }
+
+        def withTaskExecutor(taskExecutor: Executor): ListeningIntegrationComposition = {
+          val poller = new Poller(fixedDelay = delay, taskExecutor = taskExecutor)
+          new ListeningIntegrationComposition(null, new JdbcInboundAdapterConfig(target= query, poller = poller, dataSource = dataSource))
+        }
+
+      }
+
+    }
+  }
+
+  def store(query:String, dataSource: DataSource) =
+    new SendingEndpointComposition(null, new JdbcOutboundAdapterConfig(target = query, oneway = true, dataSource = dataSource)) {
+
+      /*def asFileName(fileNameGenerationFunction: _ => String) =
+      new SendingEndpointComposition(null, new JdbcOutboundGatewayConfig(target = query, dataSource = dataSource))*/
+    }
+}