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

GH-2212 Additional fixes related to lifecycle control over components that rely on SPCA

Browse Source
This commit is contained in:
Oleg Zhurakousky
2021-10-27 19:27:09 +02:00
parent 2c47c6de89
commit 5e0ad40113
2 changed files with 18 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinding.java
View File

@@ -131,12 +131,12 @@ public class DefaultBinding<T> implements Binding<T> {
@Override
public synchronized void start() {
if (this.companion != null) {
this.companion.start();
}
if (!this.isRunning()) {
if (this.lifecycle != null && this.restartable) {
this.lifecycle.start();
if (this.companion != null) {
this.companion.start();
}
}
else {
this.logger.warn("Can not re-bind an anonymous binding");
@@ -146,11 +146,11 @@ public class DefaultBinding<T> implements Binding<T> {
@Override
public synchronized void stop() {
if (this.companion != null) {
this.companion.stop();
}
if (this.isRunning()) {
this.lifecycle.stop();
if (this.companion != null) {
this.companion.stop();
}
}
}

15
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/SourceToFunctionsSupportTests.java
View File

@@ -115,7 +115,7 @@ public class SourceToFunctionsSupportTests {
}
@Test
public void testImperativeSupplier() {
public void testImperativeSupplier() throws Exception {
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
TestChannelBinderConfiguration.getCompleteConfiguration(
FunctionsConfiguration.class, SupplierConfiguration.class)).web(WebApplicationType.NONE).run(
@@ -126,8 +126,14 @@ public class SourceToFunctionsSupportTests {
String result = new String(target.receive(5000).getPayload(), StandardCharsets.UTF_8);
assertThat(result).isEqualTo("1");
result = new String(target.receive(1000).getPayload(), StandardCharsets.UTF_8);
lifecycle.stop("number-out-0");
assertThat(result).isEqualTo("2");
lifecycle.stop("number-out-0");
for (int i = 0; i < 2; i++) { //drain
target.receive(1000);
}
Thread.sleep(2000);
assertThat(target.receive(1000)).isNull();
}
}
@@ -320,7 +326,10 @@ public class SourceToFunctionsSupportTests {
@Bean
public Supplier<String> number() {
return () -> String.valueOf(this.counter.incrementAndGet());
return () -> {
System.out.println("Supplying");
return String.valueOf(this.counter.incrementAndGet());
};
}
@Bean