1818
1919import java .util .function .Consumer ;
2020
21+ import org .reactivestreams .Subscription ;
2122import org .springframework .beans .factory .BeanFactory ;
2223import org .springframework .beans .factory .annotation .Autowired ;
2324import org .springframework .cloud .stream .binder .AbstractMessageChannelBinder ;
2829import org .springframework .cloud .stream .binder .test .TestChannelBinderProvisioner .SpringIntegrationProducerDestination ;
2930import org .springframework .cloud .stream .provisioning .ConsumerDestination ;
3031import org .springframework .cloud .stream .provisioning .ProducerDestination ;
32+ import org .springframework .context .ApplicationEvent ;
33+ import org .springframework .context .ApplicationListener ;
3134import org .springframework .core .retry .RetryException ;
3235import org .springframework .integration .IntegrationMessageHeaderAccessor ;
3336import org .springframework .integration .acks .AcknowledgmentCallback ;
@@ -157,7 +160,7 @@ protected MessageProducer createConsumerEndpoint(ConsumerDestination destination
157160 .getChannel ();
158161
159162 IntegrationMessageListeningContainer messageListenerContainer = new IntegrationMessageListeningContainer ();
160- IntegrationBinderInboundChannelAdapter adapter = new IntegrationBinderInboundChannelAdapter (
163+ IntegrationBinderInboundChannelAdapter adapter = new IntegrationBinderInboundChannelAdapter (siBinderInputChannel ,
161164 messageListenerContainer );
162165 adapter .setBeanFactory (this .beanFactory );
163166 String groupName = StringUtils .hasText (group ) ? group : "anonymous" ;
@@ -173,7 +176,6 @@ protected MessageProducer createConsumerEndpoint(ConsumerDestination destination
173176 }
174177
175178 siBinderInputChannel .subscribe (messageListenerContainer );
176-
177179 return adapter ;
178180 }
179181
@@ -225,9 +227,12 @@ private static class IntegrationBinderInboundChannelAdapter
225227
226228 private RecoveryCallback <?> recoveryCallback ;
227229
228- IntegrationBinderInboundChannelAdapter (
230+ private final SubscribableChannel subscribableChannel ;
231+
232+ IntegrationBinderInboundChannelAdapter (SubscribableChannel subscribableChannel ,
229233 IntegrationMessageListeningContainer listenerContainer ) {
230234 this .listenerContainer = listenerContainer ;
235+ this .subscribableChannel = subscribableChannel ;
231236 }
232237
233238 // Temporarily unused until DLQ strategy for this binder becomes a requirement
@@ -239,6 +244,12 @@ public void setRetryTemplate(RetryTemplate retryTemplate) {
239244 this .retryTemplate = retryTemplate ;
240245 }
241246
247+ @ Override
248+ protected void doStop () {
249+ super .doStop ();
250+ this .subscribableChannel .unsubscribe (listenerContainer );
251+ }
252+
242253 @ Override
243254 protected void onInit () {
244255 if (this .retryTemplate != null ) {
0 commit comments