diff --git a/spring-cloud-streams/src/main/java/org/springframework/cloud/streams/adapter/discovery/DiscoveryClientAutoConfiguration.java b/spring-cloud-streams/src/main/java/org/springframework/cloud/streams/adapter/discovery/DiscoveryClientAutoConfiguration.java deleted file mode 100644 index c606960cb..000000000 --- a/spring-cloud-streams/src/main/java/org/springframework/cloud/streams/adapter/discovery/DiscoveryClientAutoConfiguration.java +++ /dev/null @@ -1,74 +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.cloud.streams.adapter.discovery; - -import javax.annotation.PostConstruct; - -import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.boot.autoconfigure.condition.ConditionalOnClass; -import org.springframework.cloud.client.discovery.DiscoveryClient; -import org.springframework.cloud.client.discovery.event.HeartbeatEvent; -import org.springframework.cloud.client.discovery.event.HeartbeatMonitor; -import org.springframework.cloud.streams.adapter.ChannelLocator; -import org.springframework.cloud.streams.adapter.Input; -import org.springframework.cloud.streams.adapter.MessageBusAdapter; -import org.springframework.cloud.streams.adapter.Output; -import org.springframework.context.annotation.Configuration; -import org.springframework.context.event.EventListener; - -/** - * Autoconfiguration for use when user has provided a - * {@link DiscoveryClientChannelLocator}. Listens for changes in the service registry and - * rebinds the external channels as needed. - * - * @author Dave Syer - */ -@Configuration -@ConditionalOnClass(DiscoveryClient.class) -public class DiscoveryClientAutoConfiguration { - - private HeartbeatMonitor monitor = new HeartbeatMonitor(); - - @Autowired - private MessageBusAdapter adapter; - - @Autowired(required = false) - @Input - private ChannelLocator inputChannelLocator; - - @Autowired(required = false) - @Output - private ChannelLocator outputChannelLocator; - - private boolean enabled = false; - - @PostConstruct - public void init() { - if (this.inputChannelLocator instanceof DiscoveryClientChannelLocator - || this.outputChannelLocator instanceof DiscoveryClientChannelLocator) { - this.enabled = true; - } - } - - @EventListener - public void discoveryHeartbeat(HeartbeatEvent event) { - if (this.enabled && this.monitor.update(event.getValue())) { - this.adapter.rebind(); - } - } - -} diff --git a/spring-cloud-streams/src/main/java/org/springframework/cloud/streams/adapter/discovery/DiscoveryClientChannelLocator.java b/spring-cloud-streams/src/main/java/org/springframework/cloud/streams/adapter/discovery/DiscoveryClientChannelLocator.java deleted file mode 100644 index 56bfb2ebe..000000000 --- a/spring-cloud-streams/src/main/java/org/springframework/cloud/streams/adapter/discovery/DiscoveryClientChannelLocator.java +++ /dev/null @@ -1,99 +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.cloud.streams.adapter.discovery; - -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.cloud.streams.adapter.ChannelBinding; -import org.springframework.cloud.streams.adapter.ChannelLocator; -import org.springframework.cloud.streams.adapter.ChannelsMetadata; -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 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 bindings = Collections.emptySet(); - if (name.startsWith("input")) { - name = name.replace("input", "output"); - bindings = channels.getOutputChannels(); - } - else if (name.startsWith("output")) { - name = name.replace("output", "input"); - bindings = channels.getInputChannels(); - } - for (ChannelBinding binding : bindings) { - if (name.equals(binding.getLocalName())) { - this.logger.debug("Discovered channel for '" + this.serviceId + "' (" - + name + "=" + binding.getRemoteName() + ")"); - return binding.getRemoteName(); - } - } - } - 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 instances) { - return UriComponentsBuilder - .fromUri(instances.get(new Random().nextInt(instances.size())).getUri()) - .path("channels").build().toUri(); - } - -} diff --git a/spring-cloud-streams/src/main/resources/META-INF/spring.factories b/spring-cloud-streams/src/main/resources/META-INF/spring.factories deleted file mode 100644 index 072c4fa18..000000000 --- a/spring-cloud-streams/src/main/resources/META-INF/spring.factories +++ /dev/null @@ -1,3 +0,0 @@ -# AutoConfiguration -org.springframework.boot.autoconfigure.EnableAutoConfiguration=\ -org.springframework.cloud.streams.adapter.discovery.DiscoveryClientAutoConfiguration \ No newline at end of file diff --git a/spring-cloud-streams/src/test/java/org/springframework/cloud/streams/adapter/DiscoveryClientChannelLocatorTests.java b/spring-cloud-streams/src/test/java/org/springframework/cloud/streams/adapter/DiscoveryClientChannelLocatorTests.java deleted file mode 100644 index 857ee21ac..000000000 --- a/spring-cloud-streams/src/test/java/org/springframework/cloud/streams/adapter/DiscoveryClientChannelLocatorTests.java +++ /dev/null @@ -1,84 +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.cloud.streams.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.cloud.client.DefaultServiceInstance; -import org.springframework.cloud.client.discovery.DiscoveryClient; -import org.springframework.cloud.streams.adapter.ChannelsMetadata; -import org.springframework.cloud.streams.adapter.InputChannelBinding; -import org.springframework.cloud.streams.adapter.OutputChannelBinding; -import org.springframework.cloud.streams.adapter.discovery.DiscoveryClientChannelLocator; -import org.springframework.cloud.streams.config.MessageBusProperties; -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()); - this.metadata.setOutputChannels(new HashSet()); - 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() { - OutputChannelBinding output = new OutputChannelBinding("output"); - output.setRemoteName("foo.0"); - this.metadata.getOutputChannels().add(output); - assertEquals("foo.0", this.locator.locate("input")); - } - - @Test - public void locateOutputFromInput() { - InputChannelBinding input = new InputChannelBinding("input"); - input.setRemoteName("foo.0"); - this.metadata.getInputChannels().add(input); - assertEquals("foo.0", this.locator.locate("output")); - } - - @SuppressWarnings({ "unchecked" }) - private Class anyChannels() { - return any(Class.class); - } -}