Author: ningjiang
Date: Fri Aug 7 03:58:03 2009
New Revision: 801874
URL: http://svn.apache.org/viewvc?rev=801874&view=rev
Log:
CAMEL-1884 applied patch with thanks to Stan
Added:
camel/trunk/components/camel-xmpp/src/test/java/org/apache/camel/component/xmpp/XmppRouteChatTest.java
(with props)
Modified:
camel/trunk/components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppComponent.java
camel/trunk/components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppConsumer.java
camel/trunk/components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppEndpoint.java
camel/trunk/components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppGroupChatProducer.java
camel/trunk/components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppPrivateChatProducer.java
Modified:
camel/trunk/components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppComponent.java
URL:
http://svn.apache.org/viewvc/camel/trunk/components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppComponent.java?rev=801874&r1=801873&r2=801874&view=diff
==============================================================================
---
camel/trunk/components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppComponent.java
(original)
+++
camel/trunk/components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppComponent.java
Fri Aug 7 03:58:03 2009
@@ -17,18 +17,32 @@
package org.apache.camel.component.xmpp;
import java.net.URI;
+import java.util.HashMap;
import java.util.Map;
import org.apache.camel.Endpoint;
import org.apache.camel.impl.DefaultComponent;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
/**
* @version $Revision:520964 $
*/
public class XmppComponent extends DefaultComponent {
+ private static final transient Log LOG =
LogFactory.getLog(XmppComponent.class);
+
+ //keep a cache of endpoints so they can be properly cleaned up
+ Map<String, XmppEndpoint> endpointCache = new HashMap<String,
XmppEndpoint>();
@Override
protected Endpoint createEndpoint(String uri, String remaining, Map
parameters) throws Exception {
+
+ if ( endpointCache.containsKey(uri) ) {
+ LOG.debug("Using cached endpoint for URI " + uri);
+ return endpointCache.get(uri);
+ }
+
+ LOG.debug("Creating new endpoint for URI " + uri);
XmppEndpoint endpoint = new XmppEndpoint(uri, this);
URI u = new URI(uri);
@@ -48,8 +62,16 @@
endpoint.setParticipant(remainingPath);
}
}
+
+ endpointCache.put(uri, endpoint);
return endpoint;
}
+ @Override
+ protected synchronized void doStop() throws Exception {
+ for (Map.Entry<String, XmppEndpoint> entry : endpointCache.entrySet())
{
+ entry.getValue().destroy();
+ }
+ }
}
Modified:
camel/trunk/components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppConsumer.java
URL:
http://svn.apache.org/viewvc/camel/trunk/components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppConsumer.java?rev=801874&r1=801873&r2=801874&view=diff
==============================================================================
---
camel/trunk/components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppConsumer.java
(original)
+++
camel/trunk/components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppConsumer.java
Fri Aug 7 03:58:03 2009
@@ -56,10 +56,19 @@
connection = endpoint.createConnection();
if (endpoint.getRoom() == null) {
- Chat privateChat =
connection.getChatManager().createChat(endpoint.getParticipant(), this);
- if (LOG.isInfoEnabled()) {
- LOG.info("Open private chat to: " +
privateChat.getParticipant());
+
+ // if an existing chat session has been opened (for example by a
producer) let's
+ // just add a listener to that chat
+ Chat privateChat =
connection.getChatManager().getThreadChat(endpoint.getParticipant());
+
+ if (privateChat != null) {
+ LOG.debug("Adding listener to existing chat opened to " +
privateChat.getParticipant());
+ privateChat.addMessageListener(this);
+ } else {
+ privateChat =
connection.getChatManager().createChat(endpoint.getParticipant(),
endpoint.getParticipant(), this);
+ LOG.debug("Opening private chat to " +
privateChat.getParticipant());
}
+
} else {
// add the presence packet listener to the connection so we only
get packets that concers us
// we must add the listener before creating the muc
@@ -92,13 +101,7 @@
muc.leave();
muc = null;
}
- if (connection != null) {
- if (LOG.isDebugEnabled()) {
- LOG.debug("Disconnecting from: " +
XmppEndpoint.getConnectionMessage(connection));
- }
- connection.disconnect();
- connection = null;
- }
+ //the endpoint will clean up the connection
}
public void processPacket(Packet packet) {
Modified:
camel/trunk/components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppEndpoint.java
URL:
http://svn.apache.org/viewvc/camel/trunk/components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppEndpoint.java?rev=801874&r1=801873&r2=801874&view=diff
==============================================================================
---
camel/trunk/components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppEndpoint.java
(original)
+++
camel/trunk/components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppEndpoint.java
Fri Aug 7 03:58:03 2009
@@ -17,6 +17,8 @@
package org.apache.camel.component.xmpp;
import java.util.Iterator;
+import java.util.HashMap;
+import java.util.Map;
import org.apache.camel.Consumer;
import org.apache.camel.Exchange;
@@ -58,6 +60,7 @@
private String participant;
private String nickname;
private String serviceName;
+ private XMPPConnection connection;
public XmppEndpoint() {
}
@@ -117,15 +120,17 @@
public boolean isSingleton() {
return true;
}
-
+
public XMPPConnection createConnection() throws XMPPException {
- XMPPConnection connection;
+
+ if ( connection != null )
+ return connection;
if (port > 0) {
if (getServiceName() == null) {
connection = new XMPPConnection(new
ConnectionConfiguration(host, port));
} else {
- connection = new XMPPConnection(new
ConnectionConfiguration(host, port, getServiceName()));
+ connection = new XMPPConnection(new
ConnectionConfiguration(host, port, serviceName));
}
} else {
connection = new XMPPConnection(host);
@@ -188,8 +193,13 @@
return room + "@" + chatServer;
}
- public static String getConnectionMessage(XMPPConnection connetion) {
- return connetion.getHost() + ":" + connetion.getPort() + "/" +
connetion.getServiceName();
+ public static String getConnectionMessage(XMPPConnection connection) {
+ return connection.getHost() + ":" + connection.getPort() + "/" +
connection.getServiceName();
+ }
+
+ protected synchronized void destroy() throws Exception {
+ if ( connection != null )
+ connection.disconnect();
}
// Properties
Modified:
camel/trunk/components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppGroupChatProducer.java
URL:
http://svn.apache.org/viewvc/camel/trunk/components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppGroupChatProducer.java?rev=801874&r1=801873&r2=801874&view=diff
==============================================================================
---
camel/trunk/components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppGroupChatProducer.java
(original)
+++
camel/trunk/components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppGroupChatProducer.java
Fri Aug 7 03:58:03 2009
@@ -99,13 +99,6 @@
chat.leave();
chat = null;
}
- if (connection != null) {
- if (LOG.isDebugEnabled()) {
- LOG.debug("Disconnecting from: " +
XmppEndpoint.getConnectionMessage(connection));
- }
- connection.disconnect();
- connection = null;
- }
super.doStop();
}
Modified:
camel/trunk/components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppPrivateChatProducer.java
URL:
http://svn.apache.org/viewvc/camel/trunk/components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppPrivateChatProducer.java?rev=801874&r1=801873&r2=801874&view=diff
==============================================================================
---
camel/trunk/components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppPrivateChatProducer.java
(original)
+++
camel/trunk/components/camel-xmpp/src/main/java/org/apache/camel/component/xmpp/XmppPrivateChatProducer.java
Fri Aug 7 03:58:03 2009
@@ -43,11 +43,10 @@
this.endpoint = endpoint;
this.participant = participant;
ObjectHelper.notEmpty(participant, "participant");
+ LOG.debug("Creating XmppPrivateChatProducer to participant " +
participant);
}
public void process(Exchange exchange) {
- String threadId = exchange.getExchangeId();
-
try {
if (connection == null) {
connection = endpoint.createConnection();
@@ -66,9 +65,9 @@
}
ChatManager chatManager = connection.getChatManager();
- Chat chat = chatManager.getThreadChat(threadId);
+ Chat chat = chatManager.getThreadChat(getParticipant());
if (chat == null) {
- chat = chatManager.createChat(getParticipant(), threadId, new
MessageListener() {
+ chat = chatManager.createChat(getParticipant(), getParticipant(),
new MessageListener() {
public void processMessage(Chat chat, Message message) {
// not here to do conversation
}
@@ -78,8 +77,8 @@
Message message = null;
try {
message = new Message();
- message.setTo(participant);
- message.setThread(threadId);
+ message.setTo(getParticipant());
+ message.setThread(getParticipant());
message.setType(Message.Type.normal);
endpoint.getBinding().populateXmppMessage(message, exchange);
@@ -88,24 +87,15 @@
LOG.debug("Sending XMPP message: " + message.getBody());
}
chat.sendMessage(message);
- } catch (XMPPException e) {
+ } catch (XMPPException xmppe) {
+ throw new RuntimeExchangeException("Cannot send XMPP message: " +
message
+ + " to: " + XmppEndpoint.getConnectionMessage(connection),
exchange, xmppe);
+ } catch (Exception e) {
throw new RuntimeExchangeException("Cannot send XMPP message: " +
message
+ " to: " + XmppEndpoint.getConnectionMessage(connection),
exchange, e);
}
}
- @Override
- protected void doStop() throws Exception {
- if (connection != null) {
- if (LOG.isDebugEnabled()) {
- LOG.debug("Disconnecting from: " +
XmppEndpoint.getConnectionMessage(connection));
- }
- connection.disconnect();
- connection = null;
- }
- super.doStop();
- }
-
// Properties
//
-------------------------------------------------------------------------
Added:
camel/trunk/components/camel-xmpp/src/test/java/org/apache/camel/component/xmpp/XmppRouteChatTest.java
URL:
http://svn.apache.org/viewvc/camel/trunk/components/camel-xmpp/src/test/java/org/apache/camel/component/xmpp/XmppRouteChatTest.java?rev=801874&view=auto
==============================================================================
---
camel/trunk/components/camel-xmpp/src/test/java/org/apache/camel/component/xmpp/XmppRouteChatTest.java
(added)
+++
camel/trunk/components/camel-xmpp/src/test/java/org/apache/camel/component/xmpp/XmppRouteChatTest.java
Fri Aug 7 03:58:03 2009
@@ -0,0 +1,93 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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.apache.camel.component.xmpp;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.Processor;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.component.xmpp.XmppMessage;
+import org.apache.camel.test.junit4.CamelTestSupport;
+
+import org.junit.Test;
+
+import org.jivesoftware.smack.packet.Message;
+
+/**
+ * @version $Revision$
+ */
+public class XmppRouteChatTest extends CamelTestSupport {
+ protected MockEndpoint consumerEndpoint;
+ protected MockEndpoint producerEndpoint;
+ protected String body1 = "the first message";
+ protected String body2 = "the second message";
+
+
+ @Test
+ public void testXmppChat() {
+ producerEndpoint =
(MockEndpoint)context.getEndpoint("mock:fromProducer");
+ consumerEndpoint =
(MockEndpoint)context.getEndpoint("mock:fromConsumer");
+
+ //will send chat messages to the consumer
+ template.sendBody("direct:toProducer", body1);
+ template.sendBody("direct:toProducer", body2);
+ consumerEndpoint.expectedBodiesReceived(body1, body2);
+
+ template.sendBody("direct:toConsumer", body1);
+ template.sendBody("direct:toConsumer", body2);
+ consumerEndpoint.expectedBodiesReceived(body1, body2);
+ }
+
+ protected RouteBuilder createRouteBuilder() throws Exception {
+ return new RouteBuilder() {
+ public void configure() throws Exception {
+
+ from("direct:toConsumer")
+ .to(getConsumerUri());
+
+ from("direct:toProducer")
+ .to(getProducerUri());
+
+ from(getConsumerUri())
+ .to("mock:fromConsumer");
+
+ from(getProducerUri())
+ .to("mock:fromProducer");
+
+ from("direct:getString")
+ .process(new Processor() {
+ public void process(Exchange exchange) throws
Exception {
+ XmppMessage xmppMessage =
(XmppMessage)exchange.getIn();
+ Message message = xmppMessage.getXmppMessage();
+ exchange.getIn().setBody(message.getBody());
+ }
+ })
+ .to("direct:getStringResult");
+
+ }
+ };
+ }
+
+ protected String getProducerUri() {
+ return
"xmpp://[email protected]:5222/[email protected]?user=camel_producer&password=secret&serviceName=jabber.org";
+ }
+
+ protected String getConsumerUri() {
+ return
"xmpp://[email protected]:5222/[email protected]?user=camel_consumer&password=secret&serviceName=jabber.org";
+ }
+
+}
Propchange:
camel/trunk/components/camel-xmpp/src/test/java/org/apache/camel/component/xmpp/XmppRouteChatTest.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange:
camel/trunk/components/camel-xmpp/src/test/java/org/apache/camel/component/xmpp/XmppRouteChatTest.java
------------------------------------------------------------------------------
svn:keywords = Rev Date
