From 503e749c19d0207a92a36648c2511d8bac2e7e48 Mon Sep 17 00:00:00 2001 From: Ilayaperumal Gopinathan Date: Mon, 9 Jan 2017 13:48:36 +0530 Subject: [PATCH] Fix compilation issue on ObservableToMessageChannelResultAdapter This resolves #756 --- .../reactive/ObservableToMessageChannelResultAdapter.java | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/spring-cloud-stream-reactive/src/main/java/org/springframework/cloud/stream/reactive/ObservableToMessageChannelResultAdapter.java b/spring-cloud-stream-reactive/src/main/java/org/springframework/cloud/stream/reactive/ObservableToMessageChannelResultAdapter.java index 42c459a06..1157bb71d 100644 --- a/spring-cloud-stream-reactive/src/main/java/org/springframework/cloud/stream/reactive/ObservableToMessageChannelResultAdapter.java +++ b/spring-cloud-stream-reactive/src/main/java/org/springframework/cloud/stream/reactive/ObservableToMessageChannelResultAdapter.java @@ -1,5 +1,5 @@ /* - * Copyright 2016 the original author or authors. + * Copyright 2016-2017 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. @@ -16,6 +16,7 @@ package org.springframework.cloud.stream.reactive; +import org.reactivestreams.Publisher; import reactor.core.publisher.Flux; import rx.Observable; import rx.RxReactiveStreams; @@ -27,6 +28,7 @@ import org.springframework.util.Assert; /** * A {@link StreamListenerResultAdapter} from an {@link Observable} * return type to a bound {@link MessageChannel}. + * * @author Marius Bogoevici */ public class ObservableToMessageChannelResultAdapter @@ -47,7 +49,7 @@ public class ObservableToMessageChannelResultAdapter } public void adapt(Observable streamListenerResult, MessageChannel bindingTarget) { - this.fluxToMessageChannelResultAdapter.adapt(Flux.from(RxReactiveStreams.toPublisher(streamListenerResult)), - bindingTarget); + Publisher adaptedPublisher = RxReactiveStreams.toPublisher(streamListenerResult); + this.fluxToMessageChannelResultAdapter.adapt(Flux.from(adaptedPublisher), bindingTarget); } }