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

Support flexible bound element types

Browse Source 
Fixes #519

Introduces some internal changes to the framework allowing the use
of other types than MessageChannel/SubscribableChannel as bindable
types (e.g. Flux, Observable, KStream, etc.)

- Modify BinderFactory to allow the retrieval and lookup of a
  binder not only by name, but also by binding target type
- Subsequent changes to ChannelBindingService and tests to account
  for the modified signature
- Introduce BindingTargetFactory as the contract for creating bound
  elements
- Remove any references to chanels and bound elements and use
  'binding target' systematically across the board

Reinstate our own Checkstyle checks with a reduced set of rules so that header
  validation can be performed automatically at compile time.

Use ${project.version} for the checkstyle plugin configuration

Renaming some occurences of 'boundElement' to 'bindingTarget'

Renamed a stray occurence of 'channel'

Fix some occurences of String concatenation in the same line after reformatting
This commit is contained in:
Marius Bogoevici
2016-11-21 23:58:18 -05:00
committed by markfisher
parent 8e5689c298
commit 8ac71c7b57
99 changed files with 1997 additions and 1375 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
spring-cloud-stream-reactive/src/main/java/org/springframework/cloud/stream/reactive/FluxSender.java
View File

@@ -20,18 +20,19 @@ import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
/**
* Used for {@link org.springframework.cloud.stream.annotation.StreamListener} arguments annotated with {@link
* org.springframework.cloud.stream.annotation.Output}.
* Used for {@link org.springframework.cloud.stream.annotation.StreamListener} arguments
* annotated with {@link org.springframework.cloud.stream.annotation.Output}.
* @author Marius Bogoevici
*/
public interface FluxSender {
/**
* Streams the {@link reactor.core.publisher.Flux} through the bound
* element corresponding to the {@link org.springframework.cloud.stream.annotation.Output} annotation of the
* argument.
* @param flux a {@link Flux} that will be streamed through the bound element
* @return a {@link Mono} representing the result of sending the flux (completion or error)
* Streams the {@link reactor.core.publisher.Flux} through the binding target
* corresponding to the {@link org.springframework.cloud.stream.annotation.Output}
* annotation of the argument.
* @param flux a {@link Flux} that will be streamed through the binding target
* @return a {@link Mono} representing the result of sending the flux (completion or
* error)
*/
Mono<Void> send(Flux<?> flux);
}

8
spring-cloud-stream-reactive/src/main/java/org/springframework/cloud/stream/reactive/FluxToMessageChannelResultAdapter.java
View File

@@ -36,16 +36,16 @@ public class FluxToMessageChannelResultAdapter
private Log log = LogFactory.getLog(FluxToMessageChannelResultAdapter.class);
@Override
public boolean supports(Class<?> resultType, Class<?> boundType) {
return Flux.class.isAssignableFrom(resultType) && MessageChannel.class.isAssignableFrom(boundType);
public boolean supports(Class<?> resultType, Class<?> bindingTarget) {
return Flux.class.isAssignableFrom(resultType) && MessageChannel.class.isAssignableFrom(bindingTarget);
}
public void adapt(Flux<?> streamListenerResult, MessageChannel boundElement) {
public void adapt(Flux<?> streamListenerResult, MessageChannel bindingTarget) {
streamListenerResult
.doOnError(e -> this.log.error("Error while processing result", e))
.retry()
.subscribe(
result -> boundElement.send(result instanceof Message<?> ? (Message<?>) result
result -> bindingTarget.send(result instanceof Message<?> ? (Message<?>) result
: MessageBuilder.withPayload(result).build()));
}
}

8
spring-cloud-stream-reactive/src/main/java/org/springframework/cloud/stream/reactive/MessageChannelToFluxSenderParameterAdapter.java
View File

@@ -38,14 +38,14 @@ public class MessageChannelToFluxSenderParameterAdapter
private Log log = LogFactory.getLog(MessageChannelToFluxSenderParameterAdapter.class);
@Override
public boolean supports(Class<?> boundElementType, MethodParameter methodParameter) {
public boolean supports(Class<?> bindingTargetType, MethodParameter methodParameter) {
ResolvableType type = ResolvableType.forMethodParameter(methodParameter);
return MessageChannel.class.isAssignableFrom(boundElementType)
return MessageChannel.class.isAssignableFrom(bindingTargetType)
&& FluxSender.class.isAssignableFrom(type.getRawClass());
}
@Override
public FluxSender adapt(MessageChannel boundElement, MethodParameter parameter) {
public FluxSender adapt(MessageChannel bindingTarget, MethodParameter parameter) {
return resultPublisher -> {
MonoProcessor<Void> sendResult = MonoProcessor.create();
// add error handling and reconnect in the event of an error
@@ -53,7 +53,7 @@ public class MessageChannelToFluxSenderParameterAdapter
.doOnError(e -> this.log.error("Error during processing: ", e))
.retry()
.subscribe(
result -> boundElement.send(result instanceof Message<?> ? (Message<?>) result :
result -> bindingTarget.send(result instanceof Message<?> ? (Message<?>) result :
MessageBuilder.withPayload(result).build()), e -> sendResult.onError(e),
() -> sendResult.onComplete());
return sendResult;

14
spring-cloud-stream-reactive/src/main/java/org/springframework/cloud/stream/reactive/MessageChannelToInputFluxParameterAdapter.java
View File

@@ -45,13 +45,13 @@ public class MessageChannelToInputFluxParameterAdapter
}
@Override
public boolean supports(Class<?> boundElementType, MethodParameter methodParameter) {
return SubscribableChannel.class.isAssignableFrom(boundElementType)
public boolean supports(Class<?> bindingTargetType, MethodParameter methodParameter) {
return SubscribableChannel.class.isAssignableFrom(bindingTargetType)
&& Flux.class.isAssignableFrom(methodParameter.getParameterType());
}
@Override
public Flux<?> adapt(final SubscribableChannel boundElement, MethodParameter parameter) {
public Flux<?> adapt(final SubscribableChannel bindingTarget, MethodParameter parameter) {
ResolvableType resolvableType = ResolvableType.forMethodParameter(parameter);
final Class<?> argumentClass = (resolvableType.getGeneric(0).getRawClass() != null) ? (resolvableType
.getGeneric(0).getRawClass()) : Object.class;
@@ -63,8 +63,8 @@ public class MessageChannelToInputFluxParameterAdapter
emitter.next(message);
}
};
boundElement.subscribe(messageHandler);
emitter.setCancellation(() -> boundElement.unsubscribe(messageHandler));
bindingTarget.subscribe(messageHandler);
emitter.setCancellation(() -> bindingTarget.unsubscribe(messageHandler));
}).publish().autoConnect();
}
else {
@@ -80,8 +80,8 @@ public class MessageChannelToInputFluxParameterAdapter
}
}
};
boundElement.subscribe(messageHandler);
emitter.setCancellation(() -> boundElement.unsubscribe(messageHandler));
bindingTarget.subscribe(messageHandler);
emitter.setCancellation(() -> bindingTarget.unsubscribe(messageHandler));
}).publish().autoConnect();
}
}

8
spring-cloud-stream-reactive/src/main/java/org/springframework/cloud/stream/reactive/MessageChannelToInputObservableParameterAdapter.java
View File

@@ -41,14 +41,14 @@ public class MessageChannelToInputObservableParameterAdapter
this.messageChannelToInputFluxArgumentAdapter = messageChannelToInputFluxArgumentAdapter;
}
public boolean supports(Class<?> boundElementType, MethodParameter methodParameter) {
return SubscribableChannel.class.isAssignableFrom(boundElementType)
public boolean supports(Class<?> bindingTargetType, MethodParameter methodParameter) {
return SubscribableChannel.class.isAssignableFrom(bindingTargetType)
&& Observable.class.isAssignableFrom(methodParameter.getParameterType());
}
@Override
public Observable<?> adapt(final SubscribableChannel boundElement, MethodParameter parameter) {
public Observable<?> adapt(final SubscribableChannel bindingTarget, MethodParameter parameter) {
return RxJava1Adapter.publisherToObservable(
this.messageChannelToInputFluxArgumentAdapter.adapt(boundElement, parameter));
this.messageChannelToInputFluxArgumentAdapter.adapt(bindingTarget, parameter));
}
}

8
spring-cloud-stream-reactive/src/main/java/org/springframework/cloud/stream/reactive/MessageChannelToObservableSenderParameterAdapter.java
View File

@@ -43,19 +43,19 @@ public class MessageChannelToObservableSenderParameterAdapter implements
}
@Override
public boolean supports(Class<?> boundElementType, MethodParameter methodParameter) {
public boolean supports(Class<?> bindingTargetType, MethodParameter methodParameter) {
ResolvableType type = ResolvableType.forMethodParameter(methodParameter);
return MessageChannel.class.isAssignableFrom(boundElementType)
return MessageChannel.class.isAssignableFrom(bindingTargetType)
&& ObservableSender.class.isAssignableFrom(type.getRawClass());
}
@Override
public ObservableSender adapt(MessageChannel boundElement, MethodParameter parameter) {
public ObservableSender adapt(MessageChannel bindingTarget, MethodParameter parameter) {
return new ObservableSender() {
private FluxSender fluxSender = MessageChannelToObservableSenderParameterAdapter.this
.messageChannelToFluxSenderArgumentAdapter
.adapt(boundElement, parameter);
.adapt(bindingTarget, parameter);
@Override
public Single<Void> send(Observable<?> observable) {

13
spring-cloud-stream-reactive/src/main/java/org/springframework/cloud/stream/reactive/ObservableSender.java
View File

@@ -20,17 +20,20 @@ import rx.Observable;
import rx.Single;
/**
* Used for {@link org.springframework.cloud.stream.annotation.StreamListener} arguments annotated with {@link
* org.springframework.cloud.stream.annotation.Output}.
* Used for {@link org.springframework.cloud.stream.annotation.StreamListener} arguments
* annotated with {@link org.springframework.cloud.stream.annotation.Output}.
*
* @author Marius Bogoevici
*/
public interface ObservableSender {
/**
* Streams the {@link Observable} through the bound
* element corresponding to the {@link org.springframework.cloud.stream.annotation.Output} annotation of the
* Streams the {@link Observable} through the binding target corresponding to the
* {@link org.springframework.cloud.stream.annotation.Output} annotation of the
* argument.
* @param observable an {@link Observable} that will be streamed through the bound element
*
* @param observable an {@link Observable} that will be streamed through the bound
* element
* @return a {@link Single} representing the result of an operation
*/
Single<Void> send(Observable<?> observable);

8
spring-cloud-stream-reactive/src/main/java/org/springframework/cloud/stream/reactive/ObservableToMessageChannelResultAdapter.java
View File

@@ -40,13 +40,13 @@ public class ObservableToMessageChannelResultAdapter
}
@Override
public boolean supports(Class<?> resultType, Class<?> boundType) {
public boolean supports(Class<?> resultType, Class<?> bindingTarget) {
return Observable.class.isAssignableFrom(resultType)
&& MessageChannel.class.isAssignableFrom(boundType);
&& MessageChannel.class.isAssignableFrom(bindingTarget);
}
public void adapt(Observable<?> streamListenerResult, MessageChannel boundElement) {
public void adapt(Observable<?> streamListenerResult, MessageChannel bindingTarget) {
this.fluxToMessageChannelResultAdapter.adapt(RxJava1Adapter.observableToFlux(streamListenerResult),
boundElement);
bindingTarget);
}
}

1
spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveMethodTests.java
View File

@@ -13,6 +13,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.stream.reactive;
import org.junit.Test;

4
spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveReturnWithFailureTests.java
View File

@@ -16,8 +16,6 @@
package org.springframework.cloud.stream.reactive;
import static org.assertj.core.api.Assertions.assertThat;
import java.util.Arrays;
import java.util.Collection;
import java.util.UUID;
@@ -42,6 +40,8 @@ import org.springframework.messaging.Message;
import org.springframework.messaging.handler.annotation.SendTo;
import org.springframework.messaging.support.MessageBuilder;
import static org.assertj.core.api.Assertions.assertThat;
/**
* @author Marius Bogoevici
* @author Ilayaperumal Gopinathan

4
spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveReturnWithMessageTests.java
View File

@@ -16,8 +16,6 @@
package org.springframework.cloud.stream.reactive;
import static org.assertj.core.api.Assertions.assertThat;
import java.util.Arrays;
import java.util.Collection;
import java.util.UUID;
@@ -42,6 +40,8 @@ import org.springframework.messaging.Message;
import org.springframework.messaging.handler.annotation.SendTo;
import org.springframework.messaging.support.MessageBuilder;
import static org.assertj.core.api.Assertions.assertThat;
/**
* @author Marius Bogoevici
* @author Ilayaperumal Gopinathan

4
spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveReturnWithPojoTests.java
View File

@@ -16,8 +16,6 @@
package org.springframework.cloud.stream.reactive;
import static org.assertj.core.api.Assertions.assertThat;
import java.util.Arrays;
import java.util.Collection;
import java.util.concurrent.TimeUnit;
@@ -41,6 +39,8 @@ import org.springframework.messaging.Message;
import org.springframework.messaging.handler.annotation.SendTo;
import org.springframework.messaging.support.MessageBuilder;
import static org.assertj.core.api.Assertions.assertThat;
/**
* @author Marius Bogoevici
* @author Ilayaperumal Gopinathan