Skip to content

Creating a Source Module

Jump to bottom Edit New page
jencompgeek edited this page Jun 10, 2013 · 14 revisions

Introduction

As outlined in the modules document, XD currently supports 3 types of modules: source, sink, and processor. This document walks through creation of a custom source module.

The first module in a stream is always a source. Source modules are built with Spring Integration and are typically very fine-grained. A module of type source is responsible for placing a message on a channel named output. This message can then be consumed by the other processor and sink modules in the stream. A source module is typically fed data by an inbound channel adapter, configured with a poller.

Spring Integration provides a number of adapters out of the box to support various transports, such as JMS, File, HTTP, Web Services, Mail, and more. You can typically create a source module that uses these inbound channel adapters by writing just a single Spring application context file.

These steps will demonstrate how to create and deploy a source module using the Spring Integration Twitter Search Inbound Message Channel Adapter.

To get a consumerKey and consumerSecret you need to register a twitter application. If you don’t already have one set up, you can create an app at the Twitter Developers site to get these credentials.

Create the module Application Context file

Create the Inbound Channel Adapter in a file called tweetsearch.xml:

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
	xmlns:int="http://www.springframework.org/schema/integration"
	xmlns:twitter="http://www.springframework.org/schema/integration/twitter"
	xsi:schemaLocation="http://www.springframework.org/schema/beans
		http://www.springframework.org/schema/beans/spring-beans.xsd
		http://www.springframework.org/schema/integration
		http://www.springframework.org/schema/integration/spring-integration.xsd
		http://www.springframework.org/schema/integration/twitter
		http://www.springframework.org/schema/integration/twitter/spring-integration-twitter.xsd">

        <context:property-placeholder location="tweetsearch.properties"/>

	<twitter:search-inbound-channel-adapter id="output" query="spring" twitter-template="twitterTemplate">
		<int:poller fixed-rate="5000"/>
	</twitter:search-inbound-channel-adapter>

        <bean id="twitterTemplate" class="org.springframework.social.oauth2.OAuth2Template">
		<constructor-arg index="0" value="${consumerKey}" />
		<constructor-arg index="1" value="${consumerSecret}" />
		<constructor-arg index="2" value="http://notused" />
		<constructor-arg index="3" value="http://notused" />
		<constructor-arg index="4" value="https://api.twitter.com/oauth2/token" />
	</bean>

</beans>

The adapter is configured to search Twitter for the word "spring" every 5 seconds. Once a tweet is found, it will create a message with a Tweet domain object payload and write it to a message channel called output. Since the id attribute is set, a channel with the name output is implicitly created. Alternatively, you could set channel="output" and create the output channel with <int:channel id="output"/>. The name output should be used by convention so that your source module can easily be combined with any processor and sink module in a stream.

Note
 The existing twittersearch source module contains additional logic to convert the Tweet domain object to JSON before sending the message to the output channel. This makes the data consumable by a wider range of processor and sink modules.

Make the module configurable

Users may want to specify a different Twitter query or polling interval when creating a stream. Spring XD will automatically make a PropertyPlaceholderConfigurer available to your application context. You can simply reference property names and users can then pass in values when creating a stream using the DSL.

<twitter:search-inbound-channel-adapter id="output" query="${query:spring}">
     <int:poller fixed-rate="${polling-interval:5000}"/>
</twitter:search-inbound-channel-adapter>

Now users can optionally pass query and polling-interval property values on stream creation. If not present, the specified defaults will be used.

Add auth properties

Create a tweetsearch.properties file containing the Twitter authentication properties:

consumerKey=<your key>
consumerSecret=<your secret>

Test the module locally

This section covers setup of a local project containing some code for testing outside of an XD container. This step can be skipped if you prefer to test the module by deploying to Spring XD.

Create a project

The module can be tested by writing a Spring integration test to load the context file and validate that tweets are received. In order to write the test, you will need to create a project in an IDE such as STS, Eclipse, or IDEA. Eclipse will be used for this example.

Create a tweetsearch directory and add tweetsearch.xml and tweetsearch.properties to src/main/resources. Add the following build.gradle (or an equivalent pom.xml) to the root directory:

description = 'Tweet Search Source Module'
group = 'org.springframework.xd.samples'

repositories {
  maven { url "http://repo.springsource.org/libs-snapshot" }
  maven { url "http://repo.springsource.org/plugins-release" }
}

apply plugin: 'java'
apply plugin: 'eclipse'
apply plugin: 'idea'

ext {
    junitVersion = '4.11'
    springVersion = '3.2.2.RELEASE'
    springIntegrationVersion = '3.0.0.M2'
    springSocialVersion = '1.0.5.RELEASE'
}

dependencies {
    compile("org.springframework:spring-core:$springVersion")
    compile "org.springframework:spring-context-support:$springVersion"
    compile "org.springframework.integration:spring-integration-core:$springIntegrationVersion"
    compile "org.springframework.integration:spring-integration-twitter:$springIntegrationVersion"

    // Testing
    testCompile "junit:junit:$junitVersion"
    testCompile "org.springframework:spring-test:$springVersion"
    testCompile "org.springframework.social:spring-social-twitter:$springSocialVersion"
}

defaultTasks 'build'

Run gradle eclipse to generate the Eclipse project. Import the project into Eclipse.

Create the Spring integration test

The main objective of the test is to ensure that tweets are received once the module’s Application Context is loaded. This can be tested by adding an Outbound Channel Adapter that will direct tweets to a POJO that can store them for validation.

Add the following src/test/resources/org/springframework/xd/samples/test-context.xml:

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
	xmlns:int="http://www.springframework.org/schema/integration"
	xmlns:context="http://www.springframework.org/schema/context"
	xsi:schemaLocation="http://www.springframework.org/schema/beans
		http://www.springframework.org/schema/beans/spring-beans.xsd
		http://www.springframework.org/schema/context
		http://www.springframework.org/schema/context/spring-context.xsd
		http://www.springframework.org/schema/integration
		http://www.springframework.org/schema/integration/spring-integration.xsd">

	<int:outbound-channel-adapter channel="output" ref="target" method="add" />

	<bean id="target" class="org.springframework.xd.samples.TweetCache" />

</beans>

This context creates an Outbound Channel Adapter that will subscribe to all messages on the output channel and pass the message payload to the add method of a TweetCache object. The context also creates the PropertyPlaceholderConfigurer that is ordinarily provided by the XD container.

Create the src/test/java/org/springframework/xd/samples/TweetCache class:

package org.springframework.xd.samples;
import ...

public class TweetCache {

	final BlockingDeque<Tweet> tweets = new LinkedBlockingDeque<Tweet>(99);

	public void add(Tweet tweet) {
	     tweets.add(tweet);
	}
}

The TweetCache places all received Tweets on a BlockingDeque that our test can use to validate successful routing of messages.

Lastly, create and run the src/test/java/org/springframework/xd/samples/TweetsearchSourceModuleTest:

package org.springframework.xd.samples;
import ...

@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations={"classpath:tweetsearch.xml", "test-context.xml"})
public class TweetsearchSourceModuleTest {

	@Autowired
	TweetCache tweetCache;

	@Test
	public void testTweetSearch() throws Exception {
	     assertNotNull(tweetCache.tweets.poll(5, TimeUnit.SECONDS));
	}
}

The test will load an Application Context using our tweetsearch and test context files. It will fail if a tweet is not placed into the TweetCache within 5 seconds.

You now have a way to build and test your new module independently. Time to deploy to Spring XD!

Deploy the module

Spring XD looks for modules in the ${xd.home}/modules directory. The modules directory organizes module types in sub-directories. So you will see something like:

modules/processor
modules/sink
modules/source

Simply drop tweetsearch.xml into the modules/source directory and drop tweetsearch.properties into ${xd.home}/config and fire up the server. See Getting Started to learn how to start the Spring XD server.

Test the deployed module

Once the XD server is running, create a stream to test it out. This stream will write tweets containing the word "java" to the XD log:

$ curl -d "tweetsearch --query=java | log" http://localhost:8080/streams/javasearch

You should start seeing messages like the following in the container console window:

WARN logger.javasearch: org.springframework.social.twitter.api.Tweet@7db81d4f

As noted before, logging the Tweet domain object directly isn’t much to look at. To make it prettier, create a processor module to further transform the tweet or modify this module to convert the tweet to JSON or String before sending the message to the output channel.

Clone this wiki locally