SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Make discovery work for channels

Browse Source 
Here's how it works. User adds a bean of type ChannelLocator
qualified as @InputChannel (or @OutputChannel) and it gets
used instead of the default. The discovery version needs a
service id. What it does is invert the naming convention in
channel binding - "input" locally requires "output" remotely,
"input.topic:foo" requires "output.topic:foo" remotely etc.
This commit is contained in:
Dave Syer
2015-06-05 16:58:45 +01:00
parent 538083b656
commit cee7917c49
8 changed files with 288 additions and 73 deletions
Download Patch File Download Diff File Expand all files Collapse all files

73
spring-bus-core/src/main/java/org/springframework/bus/runner/adapter/DiscoveryChannelLocator.java
View File

@@ -1,73 +0,0 @@
/*
* 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 org.springframework.bus.runner.adapter;
import java.net.URI;
import java.util.List;
import java.util.Random;
import org.springframework.cloud.client.ServiceInstance;
import org.springframework.cloud.client.discovery.DiscoveryClient;
import org.springframework.web.client.RestTemplate;
/**
* @author Dave Syer
*
*/
public class DiscoveryChannelLocator implements ChannelLocator {
private DiscoveryClient discovery;
private RestTemplate restTemplate = new RestTemplate();
private String serviceId;
public DiscoveryChannelLocator(DiscoveryClient discovery, String serviceId) {
this.discovery = discovery;
this.serviceId = serviceId;
}
@Override
public String locate(String name) {
List<ServiceInstance> instances = discovery.getInstances(serviceId);
if (instances==null || instances.isEmpty()) {
return null;
}
URI uri = pickUrl(instances);
try {
ChannelsMetadata channels = restTemplate.getForObject(uri, ChannelsMetadata.class);
for (OutputChannelSpec spec : channels.getOutputChannels()) {
if (name.equals(spec.getLocalName())) {
return spec.getName();
}
}
for (InputChannelSpec spec : channels.getInputChannels()) {
if (name.equals(spec.getLocalName())) {
return spec.getName();
}
}
} catch (Exception e) {
return null;
}
return null;
}
private URI pickUrl(List<ServiceInstance> instances) {
return instances.get(new Random().nextInt(instances.size())).getUri();
}
}

98
spring-bus-core/src/main/java/org/springframework/bus/runner/adapter/DiscoveryClientChannelLocator.java Normal file
View File

@@ -0,0 +1,98 @@
/*
* 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 org.springframework.bus.runner.adapter;
import java.net.URI;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.Random;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.cloud.client.ServiceInstance;
import org.springframework.cloud.client.discovery.DiscoveryClient;
import org.springframework.web.client.RestOperations;
import org.springframework.web.client.RestTemplate;
import org.springframework.web.util.UriComponentsBuilder;
/**
* @author Dave Syer
*
*/
public class DiscoveryClientChannelLocator implements ChannelLocator {
private Log logger = LogFactory.getLog(DiscoveryClientChannelLocator.class);
private DiscoveryClient discovery;
private RestOperations restTemplate = new RestTemplate();
private String serviceId;
public DiscoveryClientChannelLocator(DiscoveryClient discovery, String serviceId) {
this.discovery = discovery;
this.serviceId = serviceId;
}
public void setRestTemplate(RestOperations restTemplate) {
this.restTemplate = restTemplate;
}
@Override
public String locate(String name) {
List<ServiceInstance> instances = this.discovery.getInstances(this.serviceId);
if (instances == null || instances.isEmpty()) {
return null;
}
URI uri = pickUrl(instances);
try {
ChannelsMetadata channels = this.restTemplate.getForObject(uri,
ChannelsMetadata.class);
Collection<? extends InputChannelSpec> specs = Collections.emptySet();
if (name.startsWith("input")) {
name = name.replace("input", "output");
specs = channels.getOutputChannels();
}
else if (name.startsWith("output")) {
name = name.replace("output", "input");
specs = channels.getInputChannels();
}
for (InputChannelSpec spec : specs) {
if (name.equals(spec.getLocalName())) {
this.logger.debug("Discovered channel for '" + this.serviceId + "' ("
+ name + "=" + spec.getName() + ")");
return spec.getName();
}
}
}
catch (Exception e) {
this.logger.warn("Could not discover channel for '" + this.serviceId + "' ("
+ e.getClass() + ": " + e.getMessage() + ")");
return null;
}
this.logger.warn("No channel disccovered for '" + this.serviceId + "' (" + name + ")");
return null;
}
private URI pickUrl(List<ServiceInstance> instances) {
return UriComponentsBuilder
.fromUri(instances.get(new Random().nextInt(instances.size())).getUri())
.path("channels").build().toUri();
}
}

41
spring-bus-core/src/main/java/org/springframework/bus/runner/adapter/InputChannel.java Normal file
View File

@@ -0,0 +1,41 @@
/*
* Copyright 2012-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 org.springframework.bus.runner.adapter;
import java.lang.annotation.Documented;
import java.lang.annotation.ElementType;
import java.lang.annotation.Inherited;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;
import org.springframework.beans.factory.annotation.Qualifier;
/**
* Qualifier annotation for a bean relating input channels.
*
* @author Dave Syer
*/
@Qualifier
@Target({ ElementType.FIELD, ElementType.METHOD, ElementType.PARAMETER, ElementType.TYPE,
ElementType.ANNOTATION_TYPE })
@Retention(RetentionPolicy.RUNTIME)
@Inherited
@Documented
public @interface InputChannel {
}

4
spring-bus-core/src/main/java/org/springframework/bus/runner/adapter/InputChannelSpec.java
View File

@@ -25,6 +25,10 @@ public class InputChannelSpec {
private String name;
private String localName;
protected InputChannelSpec() {
this(null);
}
public InputChannelSpec(String localName) {
this.localName = localName;
}

41
spring-bus-core/src/main/java/org/springframework/bus/runner/adapter/OutputChannel.java Normal file
View File

@@ -0,0 +1,41 @@
/*
* Copyright 2012-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 org.springframework.bus.runner.adapter;
import java.lang.annotation.Documented;
import java.lang.annotation.ElementType;
import java.lang.annotation.Inherited;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;
import org.springframework.beans.factory.annotation.Qualifier;
/**
* Qualifier annotation for a bean relating input channels.
*
* @author Dave Syer
*/
@Qualifier
@Target({ ElementType.FIELD, ElementType.METHOD, ElementType.PARAMETER, ElementType.TYPE,
ElementType.ANNOTATION_TYPE })
@Retention(RetentionPolicy.RUNTIME)
@Inherited
@Documented
public @interface OutputChannel {
}

4
spring-bus-core/src/main/java/org/springframework/bus/runner/adapter/OutputChannelSpec.java
View File

@@ -26,6 +26,10 @@ public class OutputChannelSpec extends InputChannelSpec {
private boolean tapped = false;
private String tapChannelName;
protected OutputChannelSpec() {
this(null);
}
public OutputChannelSpec(String localName) {
super(localName);
}

17
spring-bus-core/src/main/java/org/springframework/bus/runner/config/MessageBusAdapterConfiguration.java
View File

@@ -29,8 +29,11 @@ import org.springframework.beans.factory.BeanFactoryUtils;
import org.springframework.beans.factory.ListableBeanFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.bus.runner.adapter.ChannelLocator;
import org.springframework.bus.runner.adapter.InputChannel;
import org.springframework.bus.runner.adapter.InputChannelSpec;
import org.springframework.bus.runner.adapter.MessageBusAdapter;
import org.springframework.bus.runner.adapter.OutputChannel;
import org.springframework.bus.runner.adapter.OutputChannelSpec;
import org.springframework.bus.runner.endpoint.ChannelsEndpoint;
import org.springframework.context.annotation.Bean;
@@ -56,12 +59,26 @@ public class MessageBusAdapterConfiguration {
@Autowired
private ListableBeanFactory beanFactory;
@Autowired(required=false)
@InputChannel
private ChannelLocator inputChannelLocator;
@Autowired(required=false)
@OutputChannel
private ChannelLocator outputChannelLocator;
@Bean
public MessageBusAdapter messageBusAdapter(MessageBusProperties module,
MessageBus messageBus) {
MessageBusAdapter adapter = new MessageBusAdapter(module, messageBus);
adapter.setOutputChannels(getOutputChannels());
adapter.setInputChannels(getInputChannels());
if (this.inputChannelLocator!=null) {
adapter.setInputChannelLocator(this.inputChannelLocator);
}
if (this.outputChannelLocator!=null) {
adapter.setOutputChannelLocator(this.outputChannelLocator);
}
return adapter;
}

83
spring-bus-core/src/test/java/org/springframework/bus/runner/adapter/DiscoveryClientChannelLocatorTests.java Normal file
View File

@@ -0,0 +1,83 @@
/*
* 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 org.springframework.bus.runner.adapter;
import static org.junit.Assert.assertEquals;
import static org.mockito.Matchers.any;
import java.net.URI;
import java.util.Arrays;
import java.util.HashSet;
import org.junit.Before;
import org.junit.Test;
import org.mockito.Mockito;
import org.springframework.bus.runner.config.MessageBusProperties;
import org.springframework.cloud.client.DefaultServiceInstance;
import org.springframework.cloud.client.discovery.DiscoveryClient;
import org.springframework.web.client.RestOperations;
/**
* @author Dave Syer
*
*/
public class DiscoveryClientChannelLocatorTests {
private DiscoveryClient client = Mockito.mock(DiscoveryClient.class);
private RestOperations restTemplate = Mockito.mock(RestOperations.class);
private DiscoveryClientChannelLocator locator = new DiscoveryClientChannelLocator(
this.client, "service");
private ChannelsMetadata metadata = new ChannelsMetadata();
@Before
public void init() {
this.locator.setRestTemplate(this.restTemplate);
this.metadata.setModule(new MessageBusProperties());
this.metadata.setInputChannels(new HashSet<InputChannelSpec>());
this.metadata.setOutputChannels(new HashSet<OutputChannelSpec>());
Mockito.when(
this.restTemplate.getForObject(Mockito.any(URI.class), anyChannels()))
.thenReturn(this.metadata);
Mockito.when(this.client.getInstances(Mockito.anyString())).thenReturn(
Arrays.asList(new DefaultServiceInstance("service", "example.com", 888,
false)));
}
@Test
public void locateInputFromOutput() {
OutputChannelSpec output = new OutputChannelSpec("output");
output.setName("foo.0");
this.metadata.getOutputChannels().add(output);
assertEquals("foo.0", this.locator.locate("input"));
}
@Test
public void locateOutputFromInput() {
InputChannelSpec input = new InputChannelSpec("input");
input.setName("foo.0");
this.metadata.getInputChannels().add(input);
assertEquals("foo.0", this.locator.locate("output"));
}
@SuppressWarnings({ "unchecked" })
private Class<ChannelsMetadata> anyChannels() {
return any(Class.class);
}
}