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

GH-2293: Cancel Subscription on Stop

Browse Source
This commit is contained in:
Gary Russell
2022-03-12 13:05:56 -05:00
parent 9b8ebe235e
commit 9ccf9ce48b
1 changed files with 12 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinder.java
View File

@@ -30,6 +30,7 @@ import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.serialization.ByteArrayDeserializer;
import org.apache.kafka.common.serialization.ByteArraySerializer;
import org.reactivestreams.Subscription;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.publisher.Sinks;
@@ -131,15 +132,26 @@ public class ReactorKafkaBinder
private final KafkaReceiver<Object, Object> receiver = KafkaReceiver.create(opts);
private volatile Subscription subscription;
@SuppressWarnings("unchecked")
@Override
protected void doStart() {
Flux<Message<Object>> flux = receiver
.receive()
.doOnSubscribe(subs -> this.subscription = subs)
.map(record -> (Message<Object>) converter.toMessage(record, null, null, null));
subscribeToPublisher(flux);
}
@Override
protected synchronized void doStop() {
if (this.subscription != null) {
this.subscription.cancel();
this.subscription = null;
}
}
};
}