diff --git a/spring-xd-samples/sink/src/main/resources/bootstrap.yml b/spring-xd-samples/sink/src/main/resources/bootstrap.yml index 3c1bf0f28..49b2737af 100644 --- a/spring-xd-samples/sink/src/main/resources/bootstrap.yml +++ b/spring-xd-samples/sink/src/main/resources/bootstrap.yml @@ -1,10 +1,11 @@ --- spring: - bus: - group: testtock - name: logger - index: 1 - # uncomment below to have this module consume from a specific partition - # in the range of 0 to N-1, where N is the upstream module's partitionCount - #consumerProperties: - # partitionIndex: 1 + cloud: + channels: + group: testtock + name: logger + index: 1 + # uncomment below to have this module consume from a specific partition + # in the range of 0 to N-1, where N is the upstream module's partitionCount + #consumerProperties: + # partitionIndex: 1 diff --git a/spring-xd-samples/source-xml/src/main/resources/bootstrap.yml b/spring-xd-samples/source-xml/src/main/resources/bootstrap.yml index 0acc7746d..6c327a9c3 100644 --- a/spring-xd-samples/source-xml/src/main/resources/bootstrap.yml +++ b/spring-xd-samples/source-xml/src/main/resources/bootstrap.yml @@ -1,5 +1,6 @@ --- spring: - bus: - group: testtock - name: ticker + cloud: + channels: + group: testtock + name: ticker diff --git a/spring-xd-samples/source/src/main/resources/bootstrap.yml b/spring-xd-samples/source/src/main/resources/bootstrap.yml index c67aff2a8..4d4899f20 100644 --- a/spring-xd-samples/source/src/main/resources/bootstrap.yml +++ b/spring-xd-samples/source/src/main/resources/bootstrap.yml @@ -1,11 +1,12 @@ --- spring: - bus: - group: testtock - name: ticker - # uncomment below to use the last digit of the seconds as a partition key - # hashcode(key) % N is then applied with N being the partitionCount value - # thus, even seconds should go to the 0 queue, odd seconds to the 1 queue - #producerProperties: - # partitionKeyExpression: payload.charAt(payload.length()-1) - # partitionCount: 2 + cloud: + channels: + group: testtock + name: ticker + # uncomment below to use the last digit of the seconds as a partition key + # hashcode(key) % N is then applied with N being the partitionCount value + # thus, even seconds should go to the 0 queue, odd seconds to the 1 queue + #producerProperties: + # partitionKeyExpression: payload.charAt(payload.length()-1) + # partitionCount: 2 diff --git a/spring-xd-samples/tap/src/main/resources/bootstrap.yml b/spring-xd-samples/tap/src/main/resources/bootstrap.yml index 698dade5b..dd020503d 100644 --- a/spring-xd-samples/tap/src/main/resources/bootstrap.yml +++ b/spring-xd-samples/tap/src/main/resources/bootstrap.yml @@ -1,11 +1,12 @@ --- spring: - bus: - group: tocktap - name: logger - index: 0 - tap: - group: testtock - name: ticker + cloud: + channels: + group: tocktap + name: logger index: 0 + tap: + group: testtock + name: ticker + index: 0 \ No newline at end of file diff --git a/vanilla-samples/pom.xml b/vanilla-samples/pom.xml new file mode 100644 index 000000000..d9b1d3fd7 --- /dev/null +++ b/vanilla-samples/pom.xml @@ -0,0 +1,35 @@ + + + 4.0.0 + org.springframework.cloud + spring-cloud-streams-samples + 1.0.0.BUILD-SNAPSHOT + pom + http://projects.spring.io/spring-xd/ + + Pivotal Software, Inc. + http://www.spring.io + + + org.springframework.cloud + spring-cloud-streams-parent + 1.0.0.BUILD-SNAPSHOT + + + source + sink + + + + + + + maven-deploy-plugin + + true + + + + + + diff --git a/vanilla-samples/sink/pom.xml b/vanilla-samples/sink/pom.xml new file mode 100644 index 000000000..aa5b7e4b8 --- /dev/null +++ b/vanilla-samples/sink/pom.xml @@ -0,0 +1,69 @@ + + + 4.0.0 + + org.springframework.cloud + spring-cloud-streams-sample-sink + 1.0.0.BUILD-SNAPSHOT + jar + + spring-cloud-streams-sample-sink + Demo project for Spring XD module + + + org.springframework.cloud + spring-xd-samples + 1.0.0.BUILD-SNAPSHOT + + + + UTF-8 + demo.SinkApplication + 1.8 + + + + + org.springframework.cloud + spring-cloud-streams + + + org.springframework.xd + spring-xd-messagebus-redis + + + org.springframework.cloud + spring-cloud-lattice-connector + 1.0.2.BUILD-SNAPSHOT + + + org.springframework.boot + spring-boot-starter-redis + + + org.springframework.boot + spring-boot-configuration-processor + true + + + + org.springframework.boot + spring-boot-starter-test + test + + + + + + + org.springframework.boot + spring-boot-maven-plugin + + exec + + + + + + diff --git a/vanilla-samples/sink/src/main/java/config/ModuleDefinition.java b/vanilla-samples/sink/src/main/java/config/ModuleDefinition.java new file mode 100644 index 000000000..f7a9abd67 --- /dev/null +++ b/vanilla-samples/sink/src/main/java/config/ModuleDefinition.java @@ -0,0 +1,50 @@ +/* + * Copyright 2015 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 config; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.cloud.streams.EnableChannelBinding; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.integration.annotation.MessageEndpoint; +import org.springframework.integration.annotation.ServiceActivator; +import org.springframework.integration.channel.DirectChannel; +import org.springframework.messaging.MessageChannel; + +/** + * @author Dave Syer + * + */ +@Configuration +@EnableChannelBinding +@MessageEndpoint +public class ModuleDefinition { + + private static Logger logger = LoggerFactory.getLogger(ModuleDefinition.class); + + @Bean + public MessageChannel input() { + return new DirectChannel(); + } + + @ServiceActivator(inputChannel="input") + public void loggerSink(Object payload) { + logger.info("Received: " + payload); + } + +} diff --git a/vanilla-samples/sink/src/main/java/demo/SinkApplication.java b/vanilla-samples/sink/src/main/java/demo/SinkApplication.java new file mode 100644 index 000000000..b9f605791 --- /dev/null +++ b/vanilla-samples/sink/src/main/java/demo/SinkApplication.java @@ -0,0 +1,17 @@ +package demo; + +import org.springframework.boot.SpringApplication; +import org.springframework.boot.autoconfigure.SpringBootApplication; +import org.springframework.context.annotation.ComponentScan; + +import config.ModuleDefinition; + +@SpringBootApplication +@ComponentScan(basePackageClasses=ModuleDefinition.class) +public class SinkApplication { + + public static void main(String[] args) throws InterruptedException { + SpringApplication.run(SinkApplication.class, args); + } + +} diff --git a/vanilla-samples/sink/src/main/resources/application.yml b/vanilla-samples/sink/src/main/resources/application.yml new file mode 100644 index 000000000..d4d0437d9 --- /dev/null +++ b/vanilla-samples/sink/src/main/resources/application.yml @@ -0,0 +1,11 @@ +server: + port: 8081 +spring: + cloud: + channels: + outputChannelName: testtock + # uncomment below to have this module consume from a specific partition + # in the range of 0 to N-1, where N is the upstream module's partitionCount + #consumerProperties: + # partitionIndex: 1 + \ No newline at end of file diff --git a/vanilla-samples/sink/src/test/java/demo/ModuleApplicationTests.java b/vanilla-samples/sink/src/test/java/demo/ModuleApplicationTests.java new file mode 100644 index 000000000..f2d65e89f --- /dev/null +++ b/vanilla-samples/sink/src/test/java/demo/ModuleApplicationTests.java @@ -0,0 +1,20 @@ +package demo; + +import org.junit.Test; +import org.junit.runner.RunWith; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.web.WebAppConfiguration; +import org.springframework.boot.test.SpringApplicationConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +@RunWith(SpringJUnit4ClassRunner.class) +@SpringApplicationConfiguration(classes = SinkApplication.class) +@WebAppConfiguration +@DirtiesContext +public class ModuleApplicationTests { + + @Test + public void contextLoads() { + } + +} diff --git a/vanilla-samples/source/pom.xml b/vanilla-samples/source/pom.xml new file mode 100644 index 000000000..28f0af948 --- /dev/null +++ b/vanilla-samples/source/pom.xml @@ -0,0 +1,68 @@ + + + 4.0.0 + + org.springframework.cloud + spring-cloud-streams-sample-source + 1.0.0.BUILD-SNAPSHOT + jar + + spring-cloud-streams-sample-source + Demo project for Spring XD module + + + org.springframework.cloud + spring-xd-samples + 1.0.0.BUILD-SNAPSHOT + + + + UTF-8 + demo.SourceApplication + 1.8 + + + + + org.springframework.cloud + spring-cloud-streams + + + org.springframework.xd + spring-xd-messagebus-redis + + + org.springframework.cloud + spring-cloud-lattice-connector + 1.0.2.BUILD-SNAPSHOT + + + org.springframework.boot + spring-boot-configuration-processor + true + + + org.springframework.boot + spring-boot-starter-redis + + + org.springframework.boot + spring-boot-starter-test + test + + + + + + + org.springframework.boot + spring-boot-maven-plugin + + exec + + + + + + diff --git a/vanilla-samples/source/src/main/java/config/ModuleDefinition.java b/vanilla-samples/source/src/main/java/config/ModuleDefinition.java new file mode 100644 index 000000000..cae7d4a94 --- /dev/null +++ b/vanilla-samples/source/src/main/java/config/ModuleDefinition.java @@ -0,0 +1,57 @@ +/* + * Copyright 2015 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 config; + +import java.text.SimpleDateFormat; +import java.util.Date; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.context.properties.EnableConfigurationProperties; +import org.springframework.cloud.streams.EnableChannelBinding; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.integration.annotation.InboundChannelAdapter; +import org.springframework.integration.annotation.Poller; +import org.springframework.integration.channel.DirectChannel; +import org.springframework.integration.core.MessageSource; +import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.support.GenericMessage; + +/** + * @author Dave Syer + * + */ +@Configuration +@EnableChannelBinding +@EnableConfigurationProperties(TimeSourceOptionsMetadata.class) +public class ModuleDefinition { + + @Autowired + private TimeSourceOptionsMetadata options; + + @Bean + public MessageChannel output() { + return new DirectChannel(); + } + + @Bean + @InboundChannelAdapter(value = "output", autoStartup = "false", poller = @Poller(fixedDelay = "${fixedDelay}", maxMessagesPerPoll = "1")) + public MessageSource timerMessageSource() { + return () -> new GenericMessage<>(new SimpleDateFormat(this.options.getFormat()).format(new Date())); + } + +} diff --git a/vanilla-samples/source/src/main/java/config/TimeSourceOptionsMetadata.java b/vanilla-samples/source/src/main/java/config/TimeSourceOptionsMetadata.java new file mode 100644 index 000000000..2e1dfb232 --- /dev/null +++ b/vanilla-samples/source/src/main/java/config/TimeSourceOptionsMetadata.java @@ -0,0 +1,106 @@ +/* + * Copyright 2013-2015 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 config; + +import javax.validation.constraints.Min; +import javax.validation.constraints.Pattern; + +import org.springframework.xd.module.options.mixins.MaxMessagesDefaultOneMixin; +import org.springframework.xd.module.options.mixins.PeriodicTriggerMixin; +import org.springframework.xd.module.options.spi.Mixin; +import org.springframework.xd.module.options.validation.DateFormat; + +/** + * Describes options to the {@code time} source module. + * + * @author Eric Bottard + * @author Gary Russell + */ +@Mixin({ PeriodicTriggerMixin.class, MaxMessagesDefaultOneMixin.class }) +public class TimeSourceOptionsMetadata { + + /** + * how to render the current time, using SimpleDateFormat + */ + private String format = "yyyy-MM-dd HH:mm:ss"; + + /** + * time delay between messages, expressed in TimeUnits (seconds by default) + */ + private int fixedDelay = 1; + + /** + * an initial delay when using a fixed delay trigger, expressed in TimeUnits (seconds by default) + */ + private int initialDelay = 0; + + /** + * the time unit for the fixed and initial delays + */ + private String timeUnit = "SECONDS"; + + /** + * the maximum messages per poll; -1 for unlimited + */ + long maxMessages = 1; + + public long getMaxMessages() { + return this.maxMessages; + } + + public void setMaxMessages(long maxMessages) { + this.maxMessages = maxMessages; + } + + @Min(0) + public int getInitialDelay() { + return this.initialDelay; + } + + public void setInitialDelay(int initialDelay) { + this.initialDelay = initialDelay; + } + + @Pattern(regexp = "(?i)(NANOSECONDS|MICROSECONDS|MILLISECONDS|SECONDS|MINUTES|HOURS|DAYS)", + message = "timeUnit must be one of NANOSECONDS, MICROSECONDS, MILLISECONDS, SECONDS, MINUTES, HOURS, DAYS (case-insensitive)") + public String getTimeUnit() { + return this.timeUnit; + } + + public void setTimeUnit(String timeUnit) { + this.timeUnit = timeUnit.toUpperCase(); + } + + @DateFormat + public String getFormat() { + return this.format; + } + + public void setFormat(String format) { + this.format = format; + } + + public int getFixedDelay() { + return this.fixedDelay; + } + + public void setFixedDelay(int fixedDelay) { + this.fixedDelay = fixedDelay; + } + + +} diff --git a/vanilla-samples/source/src/main/java/demo/SourceApplication.java b/vanilla-samples/source/src/main/java/demo/SourceApplication.java new file mode 100644 index 000000000..38d7f2162 --- /dev/null +++ b/vanilla-samples/source/src/main/java/demo/SourceApplication.java @@ -0,0 +1,17 @@ +package demo; + +import org.springframework.boot.SpringApplication; +import org.springframework.boot.autoconfigure.SpringBootApplication; +import org.springframework.context.annotation.ComponentScan; + +import config.ModuleDefinition; + +@SpringBootApplication +@ComponentScan(basePackageClasses=ModuleDefinition.class) +public class SourceApplication { + + public static void main(String[] args) throws InterruptedException { + SpringApplication.run(SourceApplication.class, args); + } + +} diff --git a/vanilla-samples/source/src/main/resources/application.yml b/vanilla-samples/source/src/main/resources/application.yml new file mode 100644 index 000000000..6a079ed13 --- /dev/null +++ b/vanilla-samples/source/src/main/resources/application.yml @@ -0,0 +1,11 @@ +fixedDelay: 5000 +spring: + cloud: + channels: + inpoutChannelName: testtock + # uncomment below to use the last digit of the seconds as a partition key + # hashcode(key) % N is then applied with N being the partitionCount value + # thus, even seconds should go to the 0 queue, odd seconds to the 1 queue + #producerProperties: + # partitionKeyExpression: payload.charAt(payload.length()-1) + # partitionCount: 2 diff --git a/vanilla-samples/source/src/test/java/demo/ModuleApplicationTests.java b/vanilla-samples/source/src/test/java/demo/ModuleApplicationTests.java new file mode 100644 index 000000000..23a9cf3b0 --- /dev/null +++ b/vanilla-samples/source/src/test/java/demo/ModuleApplicationTests.java @@ -0,0 +1,20 @@ +package demo; + +import org.junit.Test; +import org.junit.runner.RunWith; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.web.WebAppConfiguration; +import org.springframework.boot.test.SpringApplicationConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +@RunWith(SpringJUnit4ClassRunner.class) +@SpringApplicationConfiguration(classes = SourceApplication.class) +@WebAppConfiguration +@DirtiesContext +public class ModuleApplicationTests { + + @Test + public void contextLoads() { + } + +}