Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

FluxMessageChannel.send: no subscribers assertion

Browse Source
This commit is contained in:
Artem Bilan
2017-06-21 11:10:41 -04:00
parent ecb7d03895
commit a834ce9ffb
2 changed files with 15 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
spring-integration-core/src/main/java/org/springframework/integration/channel/FluxMessageChannel.java
View File

@@ -25,6 +25,7 @@ import org.reactivestreams.Publisher;
import org.reactivestreams.Subscriber;
import org.springframework.messaging.Message;
import org.springframework.util.Assert;
import reactor.core.publisher.ConnectableFlux;
import reactor.core.publisher.Flux;
@@ -59,6 +60,7 @@ public class FluxMessageChannel extends AbstractMessageChannel
@Override
protected boolean doSend(Message<?> message, long timeout) {
Assert.state(subscribers.size() > 0, () -> "The [" + this + "] doesn't have subscribers to accept messages");
this.sink.next(message);
return true;
}

13
spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/ReactiveStreamsConsumerTests.java
View File

@@ -16,6 +16,7 @@
package org.springframework.integration.channel.reactive;
import static org.hamcrest.Matchers.containsString;
import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.instanceOf;
import static org.junit.Assert.assertSame;
@@ -259,4 +260,16 @@ public class ReactiveStreamsConsumerTests {
assertThat(result, Matchers.<Message<?>>contains(testMessage, testMessage2, testMessage2));
}
@Test
public void testFluxMessageChannelSendWithoutSubscription() {
try {
new FluxMessageChannel().send(new GenericMessage<>("foo"));
}
catch (Exception e) {
assertThat(e, instanceOf(MessageDeliveryException.class));
assertThat(e.getCause(), instanceOf(IllegalStateException.class));
assertThat(e.getMessage(), containsString("doesn't have subscribers to accept messages"));
}
}
}