mirror of https://github.com/apache/activemq.git
Removed not use class, which neither has ever been documentated, or used by end users. Also they can implement their own spring JMS message converter if they want.
This commit is contained in:
Claus Ibsen
parent
5469d806e8
commit
65be281c16
|
|
@ -1,56 +0,0 @@
|
||||||
/**
|
|
||||||
*
|
|
||||||
* 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.activemq.camel.converter;
|
|
||||||
|
|
||||||
import javax.jms.JMSException;
|
|
||||||
import javax.jms.Message;
|
|
||||||
import javax.jms.Session;
|
|
||||||
|
|
||||||
import org.apache.activemq.command.ActiveMQMessage;
|
|
||||||
import org.springframework.jms.support.converter.MessageConversionException;
|
|
||||||
import org.springframework.jms.support.converter.MessageConverter;
|
|
||||||
import org.springframework.util.ObjectUtils;
|
|
||||||
|
|
||||||
/**
|
|
||||||
* Identity conversion, return the original ActiveMQMessage as is, useful when camel does message
|
|
||||||
* redelivery routing. ReadOnlyPropertes flag inverted to allow
|
|
||||||
* additional properties to be appended or existing properties to be modified
|
|
||||||
*/
|
|
||||||
public class IdentityMessageReuseConverter implements MessageConverter {
|
|
||||||
|
|
||||||
/* (non-Javadoc)
|
|
||||||
* @see org.springframework.jms.support.converter.MessageConverter#fromMessage(javax.jms.Message)
|
|
||||||
*/
|
|
||||||
public Object fromMessage(Message message) throws JMSException, MessageConversionException {
|
|
||||||
return message;
|
|
||||||
}
|
|
||||||
|
|
||||||
/* (non-Javadoc)
|
|
||||||
* @see org.springframework.jms.support.converter.MessageConverter#toMessage(java.lang.Object, javax.jms.Session)
|
|
||||||
*/
|
|
||||||
public Message toMessage(Object object, Session session) throws JMSException, MessageConversionException {
|
|
||||||
if (object instanceof ActiveMQMessage) {
|
|
||||||
// allow setting additional properties
|
|
||||||
((ActiveMQMessage)object).setReadOnlyProperties(false);
|
|
||||||
return (Message)object;
|
|
||||||
} else {
|
|
||||||
throw new MessageConversionException("Cannot reuse object of type [" +
|
|
||||||
ObjectUtils.nullSafeClassName(object) + "] as ActiveMQMessage message. Message must already be an ActiveMQMessage.");
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
@ -1,109 +0,0 @@
|
||||||
/**
|
|
||||||
*
|
|
||||||
* 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.activemq.camel;
|
|
||||||
|
|
||||||
import javax.jms.MessageConsumer;
|
|
||||||
import javax.jms.MessageProducer;
|
|
||||||
import javax.jms.Session;
|
|
||||||
import javax.jms.TextMessage;
|
|
||||||
|
|
||||||
import org.apache.activemq.ActiveMQConnection;
|
|
||||||
import org.apache.activemq.ActiveMQConnectionFactory;
|
|
||||||
import org.apache.activemq.RedeliveryPolicy;
|
|
||||||
import org.apache.activemq.command.ActiveMQDestination;
|
|
||||||
import org.apache.activemq.command.ActiveMQMessage;
|
|
||||||
import org.apache.activemq.command.ActiveMQQueue;
|
|
||||||
import org.apache.camel.CamelContext;
|
|
||||||
import org.apache.camel.Handler;
|
|
||||||
import org.apache.camel.RecipientList;
|
|
||||||
import org.junit.Test;
|
|
||||||
import org.junit.runner.RunWith;
|
|
||||||
import org.slf4j.Logger;
|
|
||||||
import org.slf4j.LoggerFactory;
|
|
||||||
import org.springframework.beans.factory.annotation.Autowired;
|
|
||||||
import org.springframework.context.ApplicationContext;
|
|
||||||
import org.springframework.test.context.ContextConfiguration;
|
|
||||||
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
|
|
||||||
|
|
||||||
import static org.junit.Assert.*;
|
|
||||||
|
|
||||||
/**
|
|
||||||
*
|
|
||||||
*/
|
|
||||||
@ContextConfiguration
|
|
||||||
@RunWith(SpringJUnit4ClassRunner.class)
|
|
||||||
public class CamelRedeliveryTest {
|
|
||||||
private static final transient Logger LOG = LoggerFactory.getLogger(CamelRedeliveryTest.class);
|
|
||||||
|
|
||||||
@Autowired
|
|
||||||
protected CamelContext camelContext;
|
|
||||||
|
|
||||||
@Autowired
|
|
||||||
protected ApplicationContext applicationContext;
|
|
||||||
|
|
||||||
@Test
|
|
||||||
public void testRedeliveryViaCamel() throws Exception {
|
|
||||||
|
|
||||||
|
|
||||||
ActiveMQConnectionFactory factory = applicationContext.getBean("connectionFactory", ActiveMQConnectionFactory.class);
|
|
||||||
ActiveMQConnection connection = (ActiveMQConnection) factory.createConnection();
|
|
||||||
|
|
||||||
// send message to dlq immediately
|
|
||||||
RedeliveryPolicy policy = connection.getRedeliveryPolicy();
|
|
||||||
policy.setMaximumRedeliveries(0);
|
|
||||||
connection.start();
|
|
||||||
|
|
||||||
Session session = connection.createSession(true, Session.SESSION_TRANSACTED);
|
|
||||||
ActiveMQQueue destination = new ActiveMQQueue("camelRedeliveryQ");
|
|
||||||
MessageProducer producer = session.createProducer(destination);
|
|
||||||
|
|
||||||
MessageConsumer consumer = session.createConsumer(destination);
|
|
||||||
// Send the messages
|
|
||||||
producer.send(session.createTextMessage("1st"));
|
|
||||||
session.commit();
|
|
||||||
LOG.info("sent 1st message");
|
|
||||||
|
|
||||||
TextMessage m;
|
|
||||||
m = (TextMessage)consumer.receive(1000);
|
|
||||||
assertNotNull(m);
|
|
||||||
assertEquals("1st", m.getText());
|
|
||||||
session.rollback();
|
|
||||||
|
|
||||||
LOG.info("received and rolledback 1st message: " + m);
|
|
||||||
m = (TextMessage)consumer.receive(1);
|
|
||||||
assertNull("no immediate redelivery", m);
|
|
||||||
|
|
||||||
m = (TextMessage)consumer.receive(20000);
|
|
||||||
LOG.info("received redelivery on second wait attempt, message: " + m);
|
|
||||||
|
|
||||||
assertNotNull("got redelivery on second attempt", m);
|
|
||||||
assertEquals("text matches original", "1st", m.getText());
|
|
||||||
|
|
||||||
// came from camel
|
|
||||||
assertTrue("redelivery marker header set, so came from camel", m.getBooleanProperty("CamelRedeliveryMarker"));
|
|
||||||
}
|
|
||||||
|
|
||||||
public static class DestinationExtractor {
|
|
||||||
|
|
||||||
@RecipientList @Handler
|
|
||||||
public String routeTo(ActiveMQMessage body) throws Exception {
|
|
||||||
ActiveMQDestination originalDestination = body.getOriginalDestination();
|
|
||||||
return "activemq:" + originalDestination.getPhysicalName() + "?explicitQosEnabled=true&messageConverter=#messageConverter";
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
@ -1,48 +0,0 @@
|
||||||
<?xml version="1.0" encoding="UTF-8"?>
|
|
||||||
<!--
|
|
||||||
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.
|
|
||||||
-->
|
|
||||||
<!-- START SNIPPET: example -->
|
|
||||||
<beans xmlns="http://www.springframework.org/schema/beans"
|
|
||||||
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
|
|
||||||
xsi:schemaLocation="
|
|
||||||
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.0.xsd
|
|
||||||
http://camel.apache.org/schema/spring http://camel.apache.org/schema/spring/camel-spring.xsd
|
|
||||||
">
|
|
||||||
|
|
||||||
<camelContext xmlns="http://camel.apache.org/schema/spring" trace="true">
|
|
||||||
<route>
|
|
||||||
<from uri="activemq:ActiveMQ.DLQ?mapJmsMessage=false"/>
|
|
||||||
<setHeader headerName="CamelRedeliveryMarker"><constant>true</constant></setHeader>
|
|
||||||
<!-- delay redelivery by 1 second using schedualler, the determination
|
|
||||||
of the delay can come form a processor to make it as complicated as needed -->
|
|
||||||
<setHeader headerName="AMQ_SCHEDULED_DELAY"><constant>1000</constant></setHeader>
|
|
||||||
|
|
||||||
<!-- dynamically configure the destination based on the message -->
|
|
||||||
<to pattern="InOnly" uri="recipientListDestinationGenerator" />
|
|
||||||
<!-- to pattern="InOnly" uri="activemq:camelRedeliveryQ?explicitQosEnabled=true&messageConverter=#messageConverter"/ -->
|
|
||||||
</route>
|
|
||||||
</camelContext>
|
|
||||||
|
|
||||||
<bean id="connectionFactory" class="org.apache.activemq.spring.ActiveMQConnectionFactory">
|
|
||||||
<property name="brokerURL" value="vm://localhost?broker.persistent=false"/>
|
|
||||||
</bean>
|
|
||||||
|
|
||||||
<bean id="messageConverter" class="org.apache.activemq.camel.converter.IdentityMessageReuseConverter" />
|
|
||||||
<bean id="recipientListDestinationGenerator" class="org.apache.activemq.camel.CamelRedeliveryTest$DestinationExtractor" />
|
|
||||||
|
|
||||||
</beans>
|
|
||||||
<!-- END SNIPPET: example -->
|
|
||||||
Loading…
Reference in New Issue