Skip to content

Spring Integration channel adapter code samples #755

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
joaoandremartins wants to merge 5 commits into from
Closed

Spring Integration channel adapter code samples #755

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
6f66f37
Spring Integration channel adapter code samples
joaoandremartins Jul 12, 2017
f13fa3c
Remove unnecessary dependencies
joaoandremartins Jul 13, 2017
431123d
More code review fixes
joaoandremartins Jul 13, 2017
a06cd45
Code review fixes
joaoandremartins Jul 13, 2017
43ef2e8
Adds README and other smaller fixes
joaoandremartins Jul 14, 2017
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -89,6 +89,7 @@
<module>pubsub/cloud-client</module>
<module>spanner/cloud-client</module>
<module>speech/cloud-client</module>
<module>spring/integration/pubsub</module>
Copy link
Contributor

@jabubake jabubake Jul 13, 2017

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Lets just use spring/pubsub for now

<module>storage/cloud-client</module>
<module>storage/json-api</module>
<module>storage/storage-transfer</module>
Expand Down
130 changes: 130 additions & 0 deletions spring/integration/pubsub/pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,130 @@
<?xml version="1.0" encoding="UTF-8"?>
Copy link
Contributor

@jabubake jabubake Jul 13, 2017

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

All the files are missing the license header.

Copy link

@artembilan artembilan Jul 13, 2017

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why XML configs should be with the License as well?
Is that your policy?
We in Spring only care about classes (Java/Groovy/Kothlin etc.)

<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<parent>
<artifactId>doc-samples</artifactId>
<groupId>com.google.cloud</groupId>
<version>1.0.0</version>
<relativePath>../../..</relativePath>
</parent>
<modelVersion>4.0.0</modelVersion>
<packaging>jar</packaging>

<artifactId>spring-integration-pubsub-samples</artifactId>

<properties>
<spring-cloud-gcp.version>1.0.0.BUILD-SNAPSHOT</spring-cloud-gcp.version>
<spring-boot.version>1.5.2.RELEASE</spring-boot.version>
</properties>

<dependencies>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-gcp-core</artifactId>
Copy link
Member

@meltsufin meltsufin Jul 13, 2017

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

You shouldn't need this. It's a transitive dependency of spring-cloud-gcp-starter-pubsub.

<version>${spring-cloud-gcp.version}</version>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-gcp-pubsub</artifactId>
Copy link
Member

@meltsufin meltsufin Jul 13, 2017

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Also not needed.

<version>${spring-cloud-gcp.version}</version>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-gcp-starter-pubsub</artifactId>
<version>${spring-cloud-gcp.version}</version>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-integration-gcp</artifactId>
Copy link
Member

@meltsufin meltsufin Jul 13, 2017

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Again, I think it's a transitive dependency of spring-cloud-gcp-starter-pubsub.

Copy link
Author

@joaoandremartins joaoandremartins Jul 13, 2017

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This one isn't.

<version>${spring-cloud-gcp.version}</version>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
<version>${spring-boot.version}</version>
</dependency>
</dependencies>

<!-- Angular -->

<build>
<resources>
<resource>
<directory>src/main/resources</directory>
</resource>
<resource>
<directory>${project.build.directory}/generated-resources</directory>
</resource>
</resources>
<plugins>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
<version>1.5.2.RELEASE</version>
</plugin>
<plugin>
<artifactId>maven-resources-plugin</artifactId>
<version>3.0.1</version>
<executions>
<execution>
<!-- Serves *only* to filter the wro.xml so it can get an absolute
path for the project -->
<id>copy-resources</id>
<phase>validate</phase>
<goals>
<goal>copy-resources</goal>
</goals>
<configuration>
<outputDirectory>${basedir}/target/wro</outputDirectory>
<resources>
<resource>
<directory>src/main/wro</directory>
<filtering>true</filtering>
</resource>
</resources>
</configuration>
</execution>
</executions>
</plugin>
<plugin>
<groupId>ro.isdc.wro4j</groupId>
<artifactId>wro4j-maven-plugin</artifactId>
<version>1.7.6</version>
<executions>
<execution>
<phase>generate-resources</phase>
<goals>
<goal>run</goal>
</goals>
</execution>
</executions>
<configuration>
<wroManagerFactory>ro.isdc.wro.maven.plugin.manager.factory.ConfigurableWroManagerFactory</wroManagerFactory>
<cssDestinationFolder>${project.build.directory}/generated-resources/static/css</cssDestinationFolder>
<jsDestinationFolder>${project.build.directory}/generated-resources/static/js</jsDestinationFolder>
<wroFile>${project.build.directory}/wro/wro.xml</wroFile>
<extraConfigFile>src/main/wro/wro.properties</extraConfigFile>
<contextFolder>${basedir}/src/main/wro</contextFolder>
</configuration>
<dependencies>
<dependency>
<groupId>org.webjars</groupId>
<artifactId>jquery</artifactId>
<version>2.1.1</version>
</dependency>
<dependency>
<groupId>org.webjars</groupId>
<artifactId>angularjs</artifactId>
<version>1.3.8</version>
</dependency>
<dependency>
<groupId>org.webjars</groupId>
<artifactId>bootstrap</artifactId>
<version>3.2.0</version>
</dependency>
</dependencies>
</plugin>
</plugins>
</build>
</project>
153 changes: 153 additions & 0 deletions spring/integration/pubsub/src/main/java/com/example/spring/pubsub/PubsubApplication.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,153 @@
package com.example.spring.pubsub;
Copy link
Contributor

@jabubake jabubake Jul 13, 2017

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

License header missing


import com.google.pubsub.v1.Subscription;
import com.google.pubsub.v1.SubscriptionName;
import com.google.pubsub.v1.Topic;
import com.google.pubsub.v1.TopicName;
import java.io.IOException;

import com.google.cloud.pubsub.v1.AckReplyConsumer;
import com.google.protobuf.ByteString;

import java.util.List;
Copy link
Contributor

@jabubake jabubake Jul 13, 2017

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Please use Google Java Formatter for the source files.

import java.util.stream.Collectors;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.cloud.gcp.pubsub.PubsubAdmin;
import org.springframework.cloud.gcp.pubsub.core.PubsubTemplate;
import org.springframework.cloud.gcp.pubsub.support.GcpHeaders;
import org.springframework.cloud.gcp.pubsub.support.SubscriberFactory;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.ComponentScan;
import org.springframework.integration.annotation.IntegrationComponentScan;
import org.springframework.integration.annotation.MessagingGateway;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.PublishSubscribeChannel;
import org.springframework.integration.gcp.AckMode;
import org.springframework.integration.gcp.inbound.PubsubInboundChannelAdapter;
import org.springframework.integration.gcp.outbound.PubsubMessageHandler;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.servlet.view.RedirectView;

@SpringBootApplication
@IntegrationComponentScan
@RestController
@ComponentScan(basePackages = {"org.springframework.cloud.gcp"})
Copy link

@artembilan artembilan Jul 13, 2017

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We don't need @IntegrationComponentScan since Boot 1.5 any more.
Also it isn't clear why do you need that @ComponentScan.
If you don't see some beans from the Spring-GCP project, then some auto-configuration is missed.
Should be fixed there.

Remember META-INF/spring.factories with the entry org.springframework.boot.autoconfigure.EnableAutoConfiguration=\ ?

public class PubsubApplication {

private static final Log LOGGER = LogFactory.getLog(PubsubApplication.class);

@Autowired
private PubsubOutboundGateway messagingGateway;

@Autowired
private PubsubAdmin admin;

public static void main(String[] args) throws IOException {
SpringApplication.run(PubsubApplication.class, args);
}

@GetMapping("/listTopics")
Copy link
Contributor

@jabubake jabubake Jul 13, 2017

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Might be worth moving the @RestController and mappings to a different class and keeping the Application class simple (less cognitive load)

public List<String> listTopics() {
Copy link
Contributor

@jabubake jabubake Jul 13, 2017

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

public methods : please provide a Javadoc

return admin.listTopics().stream()
.map(Topic::getNameAsTopicName)
.map(TopicName::getTopic)
.collect(Collectors.toList());
}

@GetMapping("/listSubscriptions")
public List<String> listSubscriptions() {
return admin.listSubscriptions().stream()
.map(Subscription::getNameAsSubscriptionName)
.map(SubscriptionName::getSubscription)
.collect(Collectors.toList());
}

@PostMapping("/postMessage")
public RedirectView addMessage(@RequestParam("message") String message) {
messagingGateway.sendToPubsub(message);
return new RedirectView("/");
}

@PostMapping("/newTopic")
public RedirectView newTopic(@RequestParam("name") String topicName) {
admin.createTopic(topicName);
return new RedirectView("/");
}

@PostMapping("/newSubscription")
public RedirectView newSubscription(@RequestParam("name") String subscriptionName,
@RequestParam("topic") String topicName) {
admin.createSubscription(subscriptionName, topicName);
return new RedirectView("/");
}

@Bean
public MessageChannel pubsubInputChannel() {
return new PublishSubscribeChannel();
}

@Bean
public PubsubInboundChannelAdapter messageChannelAdapter(
@Qualifier("pubsubInputChannel") MessageChannel inputChannel,
SubscriberFactory subscriberFactory) {
PubsubInboundChannelAdapter adapter =
new PubsubInboundChannelAdapter(subscriberFactory, "messages");
adapter.setOutputChannel(inputChannel);
adapter.setAckMode(AckMode.MANUAL);

return adapter;
}

@Bean
@ServiceActivator(inputChannel = "pubsubInputChannel")
public MessageHandler receiveMessage() {
return message -> {
LOGGER.info("Message arrived! Payload: "
+ ((ByteString) message.getPayload()).toStringUtf8());
AckReplyConsumer consumer = (AckReplyConsumer) message.getHeaders().get(
GcpHeaders.ACKNOWLEDGEMENT);
consumer.ack();
};
}

@Bean
@ServiceActivator(inputChannel = "pubsubInputChannel")
public MessageHandler receiveMessageInParallel() {
Copy link

@artembilan artembilan Jul 13, 2017

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It's not in parallel if you don't use Executor option for the PublishSubscribeChannel

return message -> {
LOGGER.info("Message also arrived here! Payload: "
+ ((ByteString) message.getPayload()).toStringUtf8());
AckReplyConsumer consumer = (AckReplyConsumer) message.getHeaders().get(
GcpHeaders.ACKNOWLEDGEMENT);
consumer.ack();
};
}

@Bean
@ServiceActivator(inputChannel = "pubsubOutputChannel")
public MessageHandler messageSender(PubsubTemplate pubsubTemplate) throws IOException {
PubsubMessageHandler outboundAdapter = new PubsubMessageHandler(pubsubTemplate);
outboundAdapter.setTopic("test");
Copy link
Member

@meltsufin meltsufin Jul 13, 2017 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Who creates this "test" topic?

Copy link
Author

@joaoandremartins joaoandremartins Jul 13, 2017

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It has to have been previously created.

Copy link
Member

@meltsufin meltsufin Jul 13, 2017

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Maybe it's better to provide the ability to select a topic to which to post the message then?

Copy link
Author

@joaoandremartins joaoandremartins Jul 13, 2017

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

You can't do that dynamically.

return outboundAdapter;
}

@Bean
public MessageChannel pubsubOutputChannel() {
return new PublishSubscribeChannel();
Copy link

@artembilan artembilan Jul 13, 2017

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Would be much clear to determine why do we need PublishSubscribeChannel here as well.
Looks like DirectChannel is pretty enough.

Also you might consider to not declare @Bean for this channel at all - and Spring Integration will take care about channel auto-creation for you by provided name in the @ServiceActivator.

Copy link
Author

@joaoandremartins joaoandremartins Jul 13, 2017

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I wanted to test delivery to all subscribers, instead of round-robin, which is what I was seeing with DirectChannel.

From reading PublishSubscriberChannel Javadoc, it seems indeed that the delivery is indeed not parallel, but single-threaded. Parallelization isn't my main focus here anyway, but rather the ability to receive the sent message in multiple subscribers at a time. Maybe I should rename the methods to messageReceiver1() and messageReceiver2() or something in those lines to keep it simpler, without going into Executor details..

Copy link

@artembilan artembilan Jul 13, 2017

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Really? I don't see more subscriber to this pubsubOutputChannel.
That is my concern to keep sample as simple as possible.

I agree about renaming subscribers to the pubsubInputChannel to avoid similar questions about parallelism

Copy link
Author

@joaoandremartins joaoandremartins Jul 13, 2017

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

To your point about not needing @Bean, IIUC that's true for outbound adapters (and it appears we're already relying on @ServiceActivator to do that for us), but the inbound adapter seems to require the channel explicitly, e.g., adapter.setOutputChannel(inputChannel);.

Copy link

@artembilan artembilan Jul 13, 2017 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yeah... I don't tell anything about inbound part here.
That is fully fine, especially that you would like to get there publish-subscribe scenario.
So, yes, definitely we have to declare MessageChannel bean if we need something else rather than just plain DirectChannel.

My concern is on this line of code only about this pubsubOutputChannel bean definition.

}

@MessagingGateway(defaultRequestChannel = "pubsubOutputChannel")
public interface PubsubOutboundGateway {

void sendToPubsub(String text);
}
}
36 changes: 36 additions & 0 deletions spring/integration/pubsub/src/main/resources/static/index.html
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,36 @@
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>Spring Integration GCP sample</title>
<link href="css/angular-bootstrap.css" rel="stylesheet">
<script src="js/angular-bootstrap.js" type="text/javascript"></script>
<script src="js/pubsub.js"></script>
</head>
<body ng-app="pubsub">
<div name="formDiv">
<form action="/postMessage" method="post">
Post message: <input type="text" name="message" /> <input type="submit" />
</form>
<form action="/newTopic" method="post">
New topic: <input type="text" name="name" /> <input type="submit" />
</form>
<form action="/newSubscription" method="post">
New subscription: <input type="text" name="name" />
for topic <input type="text" name="topic" /> <input type="submit" />
</form>
</div>
<div ng-controller="listTopics">
<h1>Topics</h1>
<div ng-repeat="topic in topics">
{{ topic }}<br/>
</div>
</div>
<div ng-controller="listSubscriptions">
<h1>Subscriptions</h1>
<div ng-repeat="subscription in subscriptions">
{{ subscription }}<br/>
</div>
</div>
</body>
</html>
11 changes: 11 additions & 0 deletions spring/integration/pubsub/src/main/resources/static/js/pubsub.js
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,11 @@
angular.module('pubsub', [])
.controller('listTopics', function($scope, $http) {
$http.get('/listTopics').success(function(data) {
$scope.topics = data;
});
})
.controller('listSubscriptions', function($scope, $http) {
$http.get('/listSubscriptions').success(function(data) {
$scope.subscriptions = data;
})
});
Empty file added spring/integration/pubsub/src/main/wro/main.less
View file
Open in desktop
Empty file.
4 changes: 4 additions & 0 deletions spring/integration/pubsub/src/main/wro/wro.properties
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,4 @@
#List of preProcessors
preProcessors=lessCssImport
#List of postProcessors
postProcessors=less4j,jsMin
9 changes: 9 additions & 0 deletions spring/integration/pubsub/src/main/wro/wro.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,9 @@
<groups xmlns="http://www.isdc.ro/wro">
<group name="angular-bootstrap">
<css>webjar:bootstrap/3.2.0/less/bootstrap.less</css>
<css>file:${project.basedir}/src/main/wro/main.less</css>
<js>webjar:jquery/2.1.1/jquery.min.js</js>
<js>webjar:bootstrap/3.2.0/bootstrap.js</js>
<js>webjar:angularjs/1.3.8/angular.min.js</js>
</group>
</groups>