diff --git a/spring-integration-xmpp/src/main/java/org/springframework/integration/xmpp/messages/XmppMessageDrivenEndpoint.java b/spring-integration-xmpp/src/main/java/org/springframework/integration/xmpp/messages/XmppMessageDrivenEndpoint.java index 0f2ac98a55..fda5e3989d 100644 --- a/spring-integration-xmpp/src/main/java/org/springframework/integration/xmpp/messages/XmppMessageDrivenEndpoint.java +++ b/spring-integration-xmpp/src/main/java/org/springframework/integration/xmpp/messages/XmppMessageDrivenEndpoint.java @@ -29,6 +29,7 @@ import org.springframework.integration.core.MessagingTemplate; import org.springframework.integration.endpoint.AbstractEndpoint; import org.springframework.integration.support.MessageBuilder; import org.springframework.integration.xmpp.XmppHeaders; +import org.springframework.util.Assert; /** * This component logs in as a user and forwards any messages to that @@ -76,7 +77,10 @@ public class XmppMessageDrivenEndpoint extends AbstractEndpoint { private volatile XMPPConnection xmppConnection; private volatile boolean extractPayload = true; - + + private volatile PacketListener packetListener; + + private volatile boolean initialized; /** * This will be injected or configured via a xmpp-connection-factory element. @@ -107,26 +111,26 @@ public class XmppMessageDrivenEndpoint extends AbstractEndpoint { @Override protected void doStart() { + Assert.isTrue(this.initialized, this.getComponentType() + " must be initialized"); logger.debug("start: " + xmppConnection.isConnected() + ":" + xmppConnection.isAuthenticated()); - xmppConnection.addPacketListener(new PacketListener() { - public void processPacket(final Packet packet) { - org.jivesoftware.smack.packet.Message message = (org.jivesoftware.smack.packet.Message) packet; - forwardXmppMessage(xmppConnection.getChatManager().getThreadChat(message.getThread()), message); - } - }, null); + xmppConnection.addPacketListener(this.packetListener, null); } @Override protected void doStop() { - if (xmppConnection.isConnected()) { - logger.debug("shutting down " + XmppMessageDrivenEndpoint.class.getName() + "."); - xmppConnection.disconnect(); - } + xmppConnection.removePacketListener(this.packetListener); } @Override protected void onInit() throws Exception { messagingTemplate.afterPropertiesSet(); + this.packetListener = new PacketListener() { + public void processPacket(final Packet packet) { + org.jivesoftware.smack.packet.Message message = (org.jivesoftware.smack.packet.Message) packet; + forwardXmppMessage(xmppConnection.getChatManager().getThreadChat(message.getThread()), message); + } + }; + this.initialized = true; } private void forwardXmppMessage(Chat chat, Message xmppMessage) { diff --git a/spring-integration-xmpp/src/test/java/org/springframework/integration/xmpp/messages/XmppMessageDrivenEndpointTests.java b/spring-integration-xmpp/src/test/java/org/springframework/integration/xmpp/messages/XmppMessageDrivenEndpointTests.java new file mode 100644 index 0000000000..c19a496405 --- /dev/null +++ b/spring-integration-xmpp/src/test/java/org/springframework/integration/xmpp/messages/XmppMessageDrivenEndpointTests.java @@ -0,0 +1,79 @@ +/* + * Copyright 2002-2010 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.integration.xmpp.messages; + +import static junit.framework.Assert.assertEquals; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.mock; + +import java.util.HashSet; +import java.util.Set; + +import org.jivesoftware.smack.PacketListener; +import org.jivesoftware.smack.XMPPConnection; +import org.jivesoftware.smack.filter.PacketFilter; +import org.junit.Test; +import org.mockito.Mockito; +import org.mockito.invocation.InvocationOnMock; +import org.mockito.stubbing.Answer; + +/** + * @author Oleg Zhurakousky + * + */ +public class XmppMessageDrivenEndpointTests { + + + @Test + /** + * Should add/remove PacketListener when endpoint started/stopped + */ + public void testLifecycle(){ + final Set packetListSet = new HashSet(); + XmppMessageDrivenEndpoint endpoint = new XmppMessageDrivenEndpoint(); + + XMPPConnection connection = mock(XMPPConnection.class); + doAnswer(new Answer() { + @Override + public Object answer(InvocationOnMock invocation) throws Throwable { + packetListSet.add((PacketListener) invocation.getArguments()[0]); + return null; + } + }).when(connection).addPacketListener(Mockito.any(PacketListener.class), (PacketFilter) Mockito.any()); + + doAnswer(new Answer() { + @Override + public Object answer(InvocationOnMock invocation) throws Throwable { + packetListSet.remove((PacketListener) invocation.getArguments()[0]); + return null; + } + }).when(connection).removePacketListener(Mockito.any(PacketListener.class)); + + endpoint.setXmppConnection(connection); + assertEquals(0, packetListSet.size()); + endpoint.afterPropertiesSet(); + endpoint.start(); + assertEquals(1, packetListSet.size()); + endpoint.stop(); + assertEquals(0, packetListSet.size()); + } + + @Test(expected=IllegalArgumentException.class) + public void testNonInitializationFailure(){ + XmppMessageDrivenEndpoint endpoint = new XmppMessageDrivenEndpoint(); + endpoint.start(); + } +}