From b6b4e25efc1c8fff703f1cdaa29a6d0e669f6f01 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Wed, 22 Feb 2012 17:16:20 -0500 Subject: [PATCH] INT-2449 Fix PubSub Subscriber Accounting PubSub channel subscriber accounting was incorrect; the counter was not decremented when unsubscribing. The handler count, reflecting the current number of subscriptions, was maintained in the AbstractSubscribableChannel. The count really belongs in the dispatcher and calls should be delegated to it, particularly for a custom ASC, which might have a custom AbstractDispatcher. However, it is possible (although perhaps rare) that someone could implement a custom channel with a custom dispatcher that is *not* a subclass of AbstractDispatcher. Therefore, we now delegate to the dispatcher if we can, and revert to a counter in the channel if we can't. Also, the asymmetric accounting in this case is resolved; the counter is decremented when a subscriber unsubscribes. Polishing - factor out common code. --- .../channel/AbstractSubscribableChannel.java | 31 +++++--- .../dispatcher/AbstractDispatcher.java | 10 ++- .../integration/channel/P2pChannelTests.java | 73 ++++++++++++++++--- 3 files changed, 92 insertions(+), 22 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractSubscribableChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractSubscribableChannel.java index 0e72ae3995..1fb17e354c 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractSubscribableChannel.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractSubscribableChannel.java @@ -24,8 +24,8 @@ import org.springframework.integration.MessageDeliveryException; import org.springframework.integration.MessageDispatchingException; import org.springframework.integration.core.MessageHandler; import org.springframework.integration.core.SubscribableChannel; +import org.springframework.integration.dispatcher.AbstractDispatcher; import org.springframework.integration.dispatcher.MessageDispatcher; -import org.springframework.integration.dispatcher.UnicastingDispatcher; import org.springframework.util.Assert; import org.springframework.util.StringUtils; @@ -44,20 +44,31 @@ public abstract class AbstractSubscribableChannel extends AbstractMessageChannel public boolean subscribe(MessageHandler handler) { MessageDispatcher dispatcher = this.getRequiredDispatcher(); boolean added = dispatcher.addHandler(handler); - if (added) { - int counter = handlerCounter.incrementAndGet(); - if (logger.isInfoEnabled()) { - logger.info("Channel '" + this.getComponentName() + "' has " + counter + " subscriber(s)."); - } - } + this.adjustCounterIfNecessary(dispatcher, added ? 1 : 0); return added; } public boolean unsubscribe(MessageHandler handle) { - if (this.getRequiredDispatcher() instanceof UnicastingDispatcher){ - handlerCounter.getAndDecrement(); + MessageDispatcher dispatcher = this.getRequiredDispatcher(); + boolean removed = dispatcher.removeHandler(handle); + this.adjustCounterIfNecessary(dispatcher, removed ? -1 : 0); + return removed; + } + + private void adjustCounterIfNecessary(MessageDispatcher dispatcher, int delta) { + if (delta != 0) { + int counter = 0; + if (dispatcher instanceof AbstractDispatcher) { + counter = ((AbstractDispatcher) dispatcher).getHandlerCount(); + } + else { + // some other dispatcher - hand-roll the counter + counter = handlerCounter.addAndGet(delta); + } + if (logger.isInfoEnabled()) { + logger.info("Channel '" + this.getComponentName() + "' has " + counter + " subscriber(s)."); + } } - return this.getRequiredDispatcher().removeHandler(handle); } @Override diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/AbstractDispatcher.java b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/AbstractDispatcher.java index 70e711d9b9..7ea5887958 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/AbstractDispatcher.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/AbstractDispatcher.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2011 the original author or authors. + * Copyright 2002-2012 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. @@ -39,6 +39,7 @@ import org.springframework.util.Assert; * @author Mark Fisher * @author Iwein Fuld * @author Oleg Zhurakousky + * @author Gary Russell */ public abstract class AbstractDispatcher implements MessageDispatcher { @@ -79,4 +80,11 @@ public abstract class AbstractDispatcher implements MessageDispatcher { public String toString() { return this.getClass().getSimpleName() + " with handlers: " + this.handlers.toString(); } + + /** + * @return The current number of handlers + */ + public int getHandlerCount() { + return this.handlers.size(); + } } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/P2pChannelTests.java b/spring-integration-core/src/test/java/org/springframework/integration/channel/P2pChannelTests.java index 89a744215c..3de3057e76 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/channel/P2pChannelTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/channel/P2pChannelTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2011 the original author or authors. + * Copyright 2002-2012 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. @@ -15,24 +15,31 @@ */ package org.springframework.integration.channel; +import static org.junit.Assert.assertEquals; +import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; import java.lang.reflect.Field; +import java.util.ArrayList; +import java.util.List; import java.util.concurrent.Executor; import org.apache.commons.logging.Log; import org.junit.Test; import org.mockito.Mockito; - +import org.mockito.invocation.InvocationOnMock; +import org.mockito.stubbing.Answer; import org.springframework.integration.core.MessageHandler; +import org.springframework.integration.dispatcher.MessageDispatcher; import org.springframework.util.ReflectionUtils; import org.springframework.util.ReflectionUtils.FieldCallback; /** * @author Oleg Zhurakousky + * @author Gary Russell * */ public class P2pChannelTests { @@ -41,10 +48,40 @@ public class P2pChannelTests { public void testDirectChannelLoggingWithMoreThenOneSubscriber() { final DirectChannel channel = new DirectChannel(); channel.setBeanName("directChannel"); - + + verifySubscriptions(channel); + } + + @Test + public void testCustomChannelLoggingWithMoreThenOneSubscriberNotAbstractDispatcher() { + final MessageDispatcher mockDispatcher = mock(MessageDispatcher.class); + when(mockDispatcher.addHandler(Mockito.any(MessageHandler.class))).thenReturn(true); + when(mockDispatcher.removeHandler(Mockito.any(MessageHandler.class))).thenReturn(true).thenReturn(false).thenReturn(true); + + final AbstractSubscribableChannel channel = new AbstractSubscribableChannel() { + @Override + protected MessageDispatcher getDispatcher() { + return mockDispatcher; + } + }; + channel.setBeanName("customChannel"); + + verifySubscriptions(channel); + } + + /** + * @param channel + */ + private void verifySubscriptions(final AbstractSubscribableChannel channel) { final Log logger = mock(Log.class); when(logger.isInfoEnabled()).thenReturn(true); - ReflectionUtils.doWithFields(AbstractMessageChannel.class, new FieldCallback() { + final List logs = new ArrayList(); + doAnswer(new Answer() { + public Object answer(InvocationOnMock invocation) throws Throwable { + logs.add((String) invocation.getArguments()[0]); + return null; + }}).when(logger).info(Mockito.anyString()); + ReflectionUtils.doWithFields(AbstractMessageChannel.class, new FieldCallback() { public void doWith(Field field) throws IllegalArgumentException, IllegalAccessException { if ("logger".equals(field.getName())){ @@ -53,21 +90,35 @@ public class P2pChannelTests { } } }); - - channel.subscribe(mock(MessageHandler.class)); - channel.subscribe(mock(MessageHandler.class)); - verify(logger, times(2)).info(Mockito.anyString()); + String log = "Channel '" + + channel.getComponentName() + + "' has " + + "%d subscriber(s)."; + + MessageHandler handler1 = mock(MessageHandler.class); + channel.subscribe(handler1); + assertEquals(String.format(log, 1), logs.remove(0)); + MessageHandler handler2 = mock(MessageHandler.class); + channel.subscribe(handler2); + assertEquals(String.format(log, 2), logs.remove(0)); + channel.unsubscribe(handler1); + assertEquals(String.format(log, 1), logs.remove(0)); + channel.unsubscribe(handler1); + assertEquals(0, logs.size()); + channel.unsubscribe(handler2); + assertEquals(String.format(log, 0), logs.remove(0)); + verify(logger, times(4)).info(Mockito.anyString()); } @Test public void testExecutorChannelLoggingWithMoreThenOneSubscriber() { final ExecutorChannel channel = new ExecutorChannel(mock(Executor.class)); channel.setBeanName("executorChannel"); - + final Log logger = mock(Log.class); when(logger.isInfoEnabled()).thenReturn(true); ReflectionUtils.doWithFields(AbstractMessageChannel.class, new FieldCallback() { - + public void doWith(Field field) throws IllegalArgumentException, IllegalAccessException { if ("logger".equals(field.getName())){ @@ -85,7 +136,7 @@ public class P2pChannelTests { public void testPubSubChannelLoggingWithMoreThenOneSubscriber() { final PublishSubscribeChannel channel = new PublishSubscribeChannel(); channel.setBeanName("pubSubChannel"); - + final Log logger = mock(Log.class); when(logger.isInfoEnabled()).thenReturn(true); ReflectionUtils.doWithFields(AbstractMessageChannel.class, new FieldCallback() {