You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@servicemix.apache.org by gn...@apache.org on 2006/02/01 16:09:18 UTC
svn commit: r374081 - in /incubator/servicemix/trunk/servicemix-core/src:
main/java/org/apache/servicemix/jbi/nmr/flow/cluster/
main/java/org/apache/servicemix/jbi/nmr/flow/jca/
main/java/org/apache/servicemix/jbi/nmr/flow/jms/ test/java/org/apache/ser...
Author: gnodet
Date: Wed Feb 1 07:09:06 2006
New Revision: 374081
URL: http://svn.apache.org/viewcvs?rev=374081&view=rev
Log:
SM-174: jms flow does not work anymore
Removed:
incubator/servicemix/trunk/servicemix-core/src/main/java/org/apache/servicemix/jbi/nmr/flow/cluster/
incubator/servicemix/trunk/servicemix-core/src/test/java/org/apache/servicemix/jbi/nmr/flow/cluster/
Modified:
incubator/servicemix/trunk/servicemix-core/src/main/java/org/apache/servicemix/jbi/nmr/flow/jca/JCAFlow.java
incubator/servicemix/trunk/servicemix-core/src/main/java/org/apache/servicemix/jbi/nmr/flow/jms/JMSFlow.java
incubator/servicemix/trunk/servicemix-core/src/test/java/org/apache/servicemix/jbi/messaging/AbstractPersistenceTest.java
incubator/servicemix/trunk/servicemix-core/src/test/java/org/apache/servicemix/jbi/nmr/flow/jms/MultipleJMSFlowTest.java
incubator/servicemix/trunk/servicemix-core/src/test/java/org/apache/servicemix/jbi/nmr/flow/jms/SimpleClusterSendSyncTest.java
incubator/servicemix/trunk/servicemix-core/src/test/resources/org/apache/servicemix/jbi/nmr/flow/jms/client.xml
Modified: incubator/servicemix/trunk/servicemix-core/src/main/java/org/apache/servicemix/jbi/nmr/flow/jca/JCAFlow.java
URL: http://svn.apache.org/viewcvs/incubator/servicemix/trunk/servicemix-core/src/main/java/org/apache/servicemix/jbi/nmr/flow/jca/JCAFlow.java?rev=374081&r1=374080&r2=374081&view=diff
==============================================================================
--- incubator/servicemix/trunk/servicemix-core/src/main/java/org/apache/servicemix/jbi/nmr/flow/jca/JCAFlow.java (original)
+++ incubator/servicemix/trunk/servicemix-core/src/main/java/org/apache/servicemix/jbi/nmr/flow/jca/JCAFlow.java Wed Feb 1 07:09:06 2006
@@ -44,6 +44,7 @@
import org.apache.activemq.advisory.AdvisorySupport;
import org.apache.activemq.command.ActiveMQDestination;
+import org.apache.activemq.command.ActiveMQMessage;
import org.apache.activemq.command.ActiveMQQueue;
import org.apache.activemq.command.ActiveMQTopic;
import org.apache.activemq.command.ConsumerId;
@@ -444,8 +445,11 @@
* @param message
*/
public void onMessage(Message message) {
+ if (message == null) {
+ return;
+ }
try {
- if (message != null && message instanceof ObjectMessage) {
+ if (message instanceof ObjectMessage) {
ObjectMessage objMsg = (ObjectMessage) message;
Object obj = objMsg.getObject();
if (obj != null) {
@@ -461,23 +465,26 @@
me.setTransactionContext(tm.getTransaction());
}
super.doRouting(me);
- }else if(obj instanceof ConsumerInfo){
- ConsumerInfo info=(ConsumerInfo) obj;
- subscriberSet.add(info.getConsumerId().getConnectionId());
- if(started.get()){
- for(Iterator i=broker.getRegistry().getLocalComponentConnectors().iterator();i.hasNext();){
- LocalComponentConnector lcc=(LocalComponentConnector) i.next();
- ComponentPacket packet=lcc.getPacket();
- ComponentPacketEvent cpe=new ComponentPacketEvent(packet,ComponentPacketEvent.ACTIVATED);
- onEvent(cpe);
- }
- }
- }else if(obj instanceof RemoveInfo){
- ConsumerId id=(ConsumerId) ((RemoveInfo) obj).getObjectId();
- subscriberSet.remove(id.getConnectionId());
- removeAllPackets(id.getConnectionId());
}
}
+ } else if (message instanceof ActiveMQMessage) {
+ Object obj = ((ActiveMQMessage) message).getDataStructure();
+ if(obj instanceof ConsumerInfo){
+ ConsumerInfo info=(ConsumerInfo) obj;
+ subscriberSet.add(info.getConsumerId().getConnectionId());
+ if(started.get()){
+ for(Iterator i=broker.getRegistry().getLocalComponentConnectors().iterator();i.hasNext();){
+ LocalComponentConnector lcc=(LocalComponentConnector) i.next();
+ ComponentPacket packet=lcc.getPacket();
+ ComponentPacketEvent cpe=new ComponentPacketEvent(packet,ComponentPacketEvent.ACTIVATED);
+ onEvent(cpe);
+ }
+ }
+ }else if(obj instanceof RemoveInfo){
+ ConsumerId id=(ConsumerId) ((RemoveInfo) obj).getObjectId();
+ subscriberSet.remove(id.getConnectionId());
+ removeAllPackets(id.getConnectionId());
+ }
}
}
catch (JMSException jmsEx) {
Modified: incubator/servicemix/trunk/servicemix-core/src/main/java/org/apache/servicemix/jbi/nmr/flow/jms/JMSFlow.java
URL: http://svn.apache.org/viewcvs/incubator/servicemix/trunk/servicemix-core/src/main/java/org/apache/servicemix/jbi/nmr/flow/jms/JMSFlow.java?rev=374081&r1=374080&r2=374081&view=diff
==============================================================================
--- incubator/servicemix/trunk/servicemix-core/src/main/java/org/apache/servicemix/jbi/nmr/flow/jms/JMSFlow.java (original)
+++ incubator/servicemix/trunk/servicemix-core/src/main/java/org/apache/servicemix/jbi/nmr/flow/jms/JMSFlow.java Wed Feb 1 07:09:06 2006
@@ -23,6 +23,7 @@
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.advisory.AdvisorySupport;
import org.apache.activemq.command.ActiveMQDestination;
+import org.apache.activemq.command.ActiveMQMessage;
import org.apache.activemq.command.ConsumerId;
import org.apache.activemq.command.ConsumerInfo;
import org.apache.activemq.command.RemoveInfo;
@@ -250,6 +251,11 @@
if (started.compareAndSet(true, false)) {
log.info(broker.getContainerName() + ": Stopping jms flow");
super.stop();
+ for (Iterator it = subscriberSet.iterator(); it.hasNext();) {
+ String id = (String) it.next();
+ removeAllPackets(id);
+ }
+ subscriberSet.clear();
try {
advisoryConsumer.close();
broadcastConsumer.close();
@@ -349,6 +355,9 @@
ComponentNameSpace id=me.getRole()==Role.PROVIDER?me.getDestinationId():me.getSourceId();
ComponentConnector cc=broker.getRegistry().getComponentConnector(id);
if(cc!=null){
+ if (me.isTransacted() && me.getMirror().getSyncState() != MessageExchangeImpl.SYNC_STATE_ASYNC) {
+ throw new IllegalStateException("transacted sendSync can not be used on jca flow with external components");
+ }
// let ActiveMQ do the routing ...
try{
String componentName=cc.getComponentNameSpace().getName();
@@ -376,8 +385,11 @@
* @param message
*/
public void onMessage(Message message) {
+ if (!started.get() || message == null) {
+ return;
+ }
try {
- if (started.get() && message != null && message instanceof ObjectMessage) {
+ if (message instanceof ObjectMessage) {
ObjectMessage objMsg = (ObjectMessage) message;
Object obj = objMsg.getObject();
if (obj != null) {
@@ -403,22 +415,23 @@
}
}
});
- }else if(obj instanceof ConsumerInfo){
- ConsumerInfo info=(ConsumerInfo) obj;
- subscriberSet.add(info.getConsumerId().getConnectionId());
- if(started.get()){
- for(Iterator i=broker.getRegistry().getLocalComponentConnectors().iterator();i.hasNext();){
- LocalComponentConnector lcc=(LocalComponentConnector) i.next();
- ComponentPacket packet=lcc.getPacket();
- ComponentPacketEvent cpe=new ComponentPacketEvent(packet,ComponentPacketEvent.ACTIVATED);
- onEvent(cpe);
- }
- }
- }else if(obj instanceof RemoveInfo){
- ConsumerId id=(ConsumerId) ((RemoveInfo) obj).getObjectId();
- subscriberSet.remove(id.getConnectionId());
- removeAllPackets(id.getConnectionId());
}
+ }
+ } else if (message instanceof ActiveMQMessage) {
+ Object obj = ((ActiveMQMessage) message).getDataStructure();
+ if(obj instanceof ConsumerInfo){
+ ConsumerInfo info=(ConsumerInfo) obj;
+ subscriberSet.add(info.getConsumerId().getConnectionId());
+ for(Iterator i=broker.getRegistry().getLocalComponentConnectors().iterator();i.hasNext();){
+ LocalComponentConnector lcc=(LocalComponentConnector) i.next();
+ ComponentPacket packet=lcc.getPacket();
+ ComponentPacketEvent cpe=new ComponentPacketEvent(packet,ComponentPacketEvent.ACTIVATED);
+ onEvent(cpe);
+ }
+ }else if(obj instanceof RemoveInfo){
+ ConsumerId id=(ConsumerId) ((RemoveInfo) obj).getObjectId();
+ subscriberSet.remove(id.getConnectionId());
+ removeAllPackets(id.getConnectionId());
}
}
}
Modified: incubator/servicemix/trunk/servicemix-core/src/test/java/org/apache/servicemix/jbi/messaging/AbstractPersistenceTest.java
URL: http://svn.apache.org/viewcvs/incubator/servicemix/trunk/servicemix-core/src/test/java/org/apache/servicemix/jbi/messaging/AbstractPersistenceTest.java?rev=374081&r1=374080&r2=374081&view=diff
==============================================================================
--- incubator/servicemix/trunk/servicemix-core/src/test/java/org/apache/servicemix/jbi/messaging/AbstractPersistenceTest.java (original)
+++ incubator/servicemix/trunk/servicemix-core/src/test/java/org/apache/servicemix/jbi/messaging/AbstractPersistenceTest.java Wed Feb 1 07:09:06 2006
@@ -41,7 +41,7 @@
public abstract class AbstractPersistenceTest extends AbstractTransactionTest {
protected JBIContainer createJbiContainer(String name) throws Exception {
- JBIContainer container = new JBIContainer();
+ JBIContainer container = new JBIContainer();
container.setTransactionManager(tm);
container.setName(name);
container.setFlow(createFlow());
@@ -51,68 +51,67 @@
container.start();
return container;
}
-
+
protected void runSimpleTest(final boolean syncSend, final boolean syncReceive) throws Exception {
- //final int numMessages = 1;
+ // final int numMessages = 1;
final int numMessages = NUM_MESSAGES;
final SenderComponent sender = new SenderComponent();
sender.setResolver(new ServiceNameEndpointResolver(ReceiverComponent.SERVICE));
final Receiver receiver;
final Map delivered = new ConcurrentHashMap();
if (syncReceive) {
- receiver = new ReceiverComponent() {
- public void onMessageExchange(MessageExchange exchange) throws MessagingException {
- try {
- if (delivered.get(exchange.getExchangeId()) == null) {
- System.err.println("Message delivery rolled back: " + exchange.getExchangeId());
- delivered.put(exchange.getExchangeId(), Boolean.TRUE);
- tm.setRollbackOnly();
- done(exchange);
- } else {
- System.err.println("Message delivery accepted: " + exchange.getExchangeId());
- super.onMessageExchange(exchange);
- }
- } catch (Exception e) {
- throw new MessagingException(e);
- }
- }
- };
+ receiver = new ReceiverComponent() {
+ public void onMessageExchange(MessageExchange exchange) throws MessagingException {
+ try {
+ if (delivered.get(exchange.getExchangeId()) == null) {
+ System.err.println("Message delivery rolled back: " + exchange.getExchangeId());
+ delivered.put(exchange.getExchangeId(), Boolean.TRUE);
+ tm.setRollbackOnly();
+ } else {
+ System.err.println("Message delivery accepted: " + exchange.getExchangeId());
+ super.onMessageExchange(exchange);
+ }
+ } catch (Exception e) {
+ throw new MessagingException(e);
+ }
+ }
+ };
} else {
- receiver = new AsyncReceiverPojo() {
- public void onMessageExchange(MessageExchange exchange) throws MessagingException {
- try {
- if (delivered.get(exchange.getExchangeId()) == null) {
- System.err.println("Message delivery rolled back: " + exchange.getExchangeId());
- delivered.put(exchange.getExchangeId(), Boolean.TRUE);
- tm.setRollbackOnly();
- exchange.setStatus(ExchangeStatus.DONE);
- getContext().getDeliveryChannel().send(exchange);
- } else {
- System.err.println("Message delivery accepted: " + exchange.getExchangeId());
- super.onMessageExchange(exchange);
- }
- } catch (Exception e) {
- throw new MessagingException(e);
- }
- }
- };
+ receiver = new AsyncReceiverPojo() {
+ public void onMessageExchange(MessageExchange exchange) throws MessagingException {
+ try {
+ if (delivered.get(exchange.getExchangeId()) == null) {
+ System.err.println("Message delivery rolled back: " + exchange.getExchangeId());
+ delivered.put(exchange.getExchangeId(), Boolean.TRUE);
+ tm.setRollbackOnly();
+ exchange.setStatus(ExchangeStatus.DONE);
+ getContext().getDeliveryChannel().send(exchange);
+ } else {
+ System.err.println("Message delivery accepted: " + exchange.getExchangeId());
+ super.onMessageExchange(exchange);
+ }
+ } catch (Exception e) {
+ throw new MessagingException(e);
+ }
+ }
+ };
}
senderContainer.activateComponent(new ActivationSpec("sender", sender));
senderContainer.activateComponent(new ActivationSpec("receiver", receiver));
-
+
tt.execute(new TransactionCallback() {
- public Object doInTransaction(TransactionStatus status) {
+ public Object doInTransaction(TransactionStatus status) {
try {
sender.sendMessages(numMessages, syncSend);
} catch (JBIException e) {
throw new RuntimeJBIException(e);
}
- return null;
- }
+ return null;
+ }
});
- //sender.sendMessages(NUM_MESSAGES, syncSend);
+ //sender.sendMessages(NUM_MESSAGES, syncSend);
receiver.getMessageList().assertMessagesReceived(numMessages);
}
-
+
}
Modified: incubator/servicemix/trunk/servicemix-core/src/test/java/org/apache/servicemix/jbi/nmr/flow/jms/MultipleJMSFlowTest.java
URL: http://svn.apache.org/viewcvs/incubator/servicemix/trunk/servicemix-core/src/test/java/org/apache/servicemix/jbi/nmr/flow/jms/MultipleJMSFlowTest.java?rev=374081&r1=374080&r2=374081&view=diff
==============================================================================
--- incubator/servicemix/trunk/servicemix-core/src/test/java/org/apache/servicemix/jbi/nmr/flow/jms/MultipleJMSFlowTest.java (original)
+++ incubator/servicemix/trunk/servicemix-core/src/test/java/org/apache/servicemix/jbi/nmr/flow/jms/MultipleJMSFlowTest.java Wed Feb 1 07:09:06 2006
@@ -54,28 +54,43 @@
containers[i].init();
}
long t0 = System.currentTimeMillis();
+ printNodes(containers);
for (int i = 0; i < containers.length; i++) {
containers[i].start();
+ printNodes(containers);
}
long t1 = System.currentTimeMillis();
System.err.println(t1 - t0);
for (int i = 0; i < containers.length; i++) {
containers[i].stop();
+ printNodes(containers);
}
for (int i = 0; i < containers.length; i++) {
containers[i].start();
+ printNodes(containers);
}
for (int i = 0; i < containers.length; i++) {
containers[i].stop();
+ printNodes(containers);
}
for (int i = 0; i < containers.length; i++) {
containers[i].shutDown();
+ printNodes(containers);
}
}
+ protected void printNodes(JBIContainer[] containers) {
+ String msg = "Nodes: ";
+ for (int j = 0; j < containers.length; j++) {
+ if (j > 0) msg += ", ";
+ msg += ((JMSFlow) containers[j].getFlow()).numberInNetwork();
+ }
+ System.err.println(msg);
+ }
+
protected Flow createFlow() {
JMSFlow flow = new JMSFlow();
- flow.setJmsURL("tcp://localhost:61216");
+ //flow.setJmsURL("tcp://localhost:61216");
return flow;
}
Modified: incubator/servicemix/trunk/servicemix-core/src/test/java/org/apache/servicemix/jbi/nmr/flow/jms/SimpleClusterSendSyncTest.java
URL: http://svn.apache.org/viewcvs/incubator/servicemix/trunk/servicemix-core/src/test/java/org/apache/servicemix/jbi/nmr/flow/jms/SimpleClusterSendSyncTest.java?rev=374081&r1=374080&r2=374081&view=diff
==============================================================================
--- incubator/servicemix/trunk/servicemix-core/src/test/java/org/apache/servicemix/jbi/nmr/flow/jms/SimpleClusterSendSyncTest.java (original)
+++ incubator/servicemix/trunk/servicemix-core/src/test/java/org/apache/servicemix/jbi/nmr/flow/jms/SimpleClusterSendSyncTest.java Wed Feb 1 07:09:06 2006
@@ -52,22 +52,18 @@
super.tearDown();
}
- public void testSendSync() {
- try {
- ApplicationContext ctx = new ClassPathXmlApplicationContext("org/apache/servicemix/jbi/nmr/flow/jms/client.xml");
- ServiceMixClient client = (ServiceMixClient) ctx.getBean("client");
- Thread.sleep(5000);
- InOut exchange = client.createInOutExchange();
- exchange.setService(new QName("http://www.habuma.com/foo", "pingService"));
- NormalizedMessage in = exchange.getInMessage();
- in.setContent(new StringSource("<ping>Pinging you</ping>"));
- System.out.println("SENDING; exchange.status=" + exchange.getStatus());
- client.sendSync(exchange);
- assertNotNull(exchange.getOutMessage());
- System.out.println("GOT RESPONSE; exchange.out=" + new SourceTransformer().toString(exchange.getOutMessage().getContent()));
- client.done(exchange);
- } catch (Exception e) {
- System.out.println("Could not connect to Service: " + e);
- }
+ public void testSendSync() throws Exception {
+ ApplicationContext ctx = new ClassPathXmlApplicationContext("org/apache/servicemix/jbi/nmr/flow/jms/client.xml");
+ ServiceMixClient client = (ServiceMixClient) ctx.getBean("client");
+ Thread.sleep(500);
+ InOut exchange = client.createInOutExchange();
+ exchange.setService(new QName("http://www.habuma.com/foo", "pingService"));
+ NormalizedMessage in = exchange.getInMessage();
+ in.setContent(new StringSource("<ping>Pinging you</ping>"));
+ System.out.println("SENDING; exchange.status=" + exchange.getStatus());
+ client.sendSync(exchange);
+ assertNotNull(exchange.getOutMessage());
+ System.out.println("GOT RESPONSE; exchange.out=" + new SourceTransformer().toString(exchange.getOutMessage().getContent()));
+ client.done(exchange);
}
}
Modified: incubator/servicemix/trunk/servicemix-core/src/test/resources/org/apache/servicemix/jbi/nmr/flow/jms/client.xml
URL: http://svn.apache.org/viewcvs/incubator/servicemix/trunk/servicemix-core/src/test/resources/org/apache/servicemix/jbi/nmr/flow/jms/client.xml?rev=374081&r1=374080&r2=374081&view=diff
==============================================================================
--- incubator/servicemix/trunk/servicemix-core/src/test/resources/org/apache/servicemix/jbi/nmr/flow/jms/client.xml (original)
+++ incubator/servicemix/trunk/servicemix-core/src/test/resources/org/apache/servicemix/jbi/nmr/flow/jms/client.xml Wed Feb 1 07:09:06 2006
@@ -13,7 +13,6 @@
name="jbi"/>
<bean id="client"
- class="org.apache.servicemix.client.DefaultServiceMixClient">
- <constructor-arg ref="jbi" />
- </bean>
+ class="org.apache.servicemix.client.RemoteServiceMixClient"
+ init-method="start"/>
</beans>