mirror of https://github.com/apache/activemq.git
Fix for https://issues.apache.org/jira/browse/AMQ-4305 Support destination lookup by JNDI name in resource adapter, patch applied, thanks Harald Wellmann. Also added unit tests
git-svn-id: https://svn.apache.org/repos/asf/activemq/trunk@1445642 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
parent
b2fca26213
commit
7bc7178899
|
@ -74,6 +74,7 @@ public class ActiveMQActivationSpec implements MessageActivationSpec, Serializab
|
|||
private String enableBatch = "false";
|
||||
private String maxMessagesPerBatch = "10";
|
||||
private RedeliveryPolicy redeliveryPolicy;
|
||||
private boolean useJndi;
|
||||
|
||||
/**
|
||||
* @see javax.resource.spi.ActivationSpec#validate()
|
||||
|
@ -452,7 +453,7 @@ public class ActiveMQActivationSpec implements MessageActivationSpec, Serializab
|
|||
public String toString() {
|
||||
return "ActiveMQActivationSpec{" + "acknowledgeMode='" + acknowledgeMode + "'" + ", destinationType='" + destinationType + "'" + ", messageSelector='" + messageSelector + "'"
|
||||
+ ", destination='" + destination + "'" + ", clientId='" + clientId + "'" + ", subscriptionName='" + subscriptionName + "'" + ", subscriptionDurability='" + subscriptionDurability
|
||||
+ "'" + "}";
|
||||
+ "'" + ", useJndi='"+ useJndi + "'" +"}";
|
||||
}
|
||||
|
||||
public int getAcknowledgeModeForSession() {
|
||||
|
@ -665,4 +666,12 @@ public class ActiveMQActivationSpec implements MessageActivationSpec, Serializab
|
|||
}
|
||||
return redeliveryPolicy;
|
||||
}
|
||||
|
||||
public void setUseJndi(boolean useJndi) {
|
||||
this.useJndi = useJndi;
|
||||
}
|
||||
|
||||
public boolean isUseJndi() {
|
||||
return useJndi;
|
||||
}
|
||||
}
|
||||
|
|
|
@ -28,6 +28,8 @@ import javax.jms.Message;
|
|||
import javax.jms.MessageListener;
|
||||
import javax.jms.Session;
|
||||
import javax.jms.Topic;
|
||||
import javax.naming.InitialContext;
|
||||
import javax.naming.NamingException;
|
||||
import javax.resource.ResourceException;
|
||||
import javax.resource.spi.endpoint.MessageEndpointFactory;
|
||||
import javax.resource.spi.work.Work;
|
||||
|
@ -200,14 +202,28 @@ public class ActiveMQEndpointWorker {
|
|||
};
|
||||
|
||||
MessageActivationSpec activationSpec = endpointActivationKey.getActivationSpec();
|
||||
if ("javax.jms.Queue".equals(activationSpec.getDestinationType())) {
|
||||
dest = new ActiveMQQueue(activationSpec.getDestination());
|
||||
} else if ("javax.jms.Topic".equals(activationSpec.getDestinationType())) {
|
||||
dest = new ActiveMQTopic(activationSpec.getDestination());
|
||||
} else {
|
||||
throw new ResourceException("Unknown destination type: " + activationSpec.getDestinationType());
|
||||
if (activationSpec.isUseJndi()) {
|
||||
try {
|
||||
InitialContext initialContext = new InitialContext();
|
||||
dest = (ActiveMQDestination) initialContext.lookup(activationSpec.getDestination());
|
||||
}
|
||||
catch (NamingException exc) {
|
||||
throw new ResourceException("JNDI lookup failed for "
|
||||
+ activationSpec.getDestination());
|
||||
}
|
||||
}
|
||||
else {
|
||||
if ("javax.jms.Queue".equals(activationSpec.getDestinationType())) {
|
||||
dest = new ActiveMQQueue(activationSpec.getDestination());
|
||||
}
|
||||
else if ("javax.jms.Topic".equals(activationSpec.getDestinationType())) {
|
||||
dest = new ActiveMQTopic(activationSpec.getDestination());
|
||||
}
|
||||
else {
|
||||
throw new ResourceException("Unknown destination type: "
|
||||
+ activationSpec.getDestinationType());
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
|
@ -128,4 +128,7 @@ public interface MessageActivationSpec extends ActivationSpec {
|
|||
RedeliveryPolicy redeliveryPolicy();
|
||||
|
||||
RedeliveryPolicy lazyCreateRedeliveryPolicy();
|
||||
|
||||
boolean isUseJndi();
|
||||
|
||||
}
|
||||
|
|
|
@ -143,6 +143,70 @@ public class MDBTest extends TestCase {
|
|||
|
||||
}
|
||||
|
||||
public void testDestinationInJndi() throws Exception{
|
||||
ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("vm://localhost?broker.persistent=false");
|
||||
Connection connection = factory.createConnection();
|
||||
connection.start();
|
||||
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
|
||||
|
||||
MessageConsumer advisory = session.createConsumer(AdvisorySupport.getConsumerAdvisoryTopic(new ActiveMQQueue("TEST")));
|
||||
|
||||
ActiveMQResourceAdapter adapter = new ActiveMQResourceAdapter();
|
||||
adapter.setServerUrl("vm://localhost?broker.persistent=false");
|
||||
adapter.setQueuePrefetch(1);
|
||||
adapter.start(new StubBootstrapContext());
|
||||
|
||||
final CountDownLatch messageDelivered = new CountDownLatch(1);
|
||||
|
||||
final StubMessageEndpoint endpoint = new StubMessageEndpoint() {
|
||||
public void onMessage(Message message) {
|
||||
super.onMessage(message);
|
||||
messageDelivered.countDown();
|
||||
};
|
||||
};
|
||||
|
||||
ActiveMQActivationSpec activationSpec = new ActiveMQActivationSpec();
|
||||
activationSpec.setDestinationType(Queue.class.getName());
|
||||
activationSpec.setDestination("MyQueue");
|
||||
activationSpec.setUseJndi(true);
|
||||
activationSpec.setResourceAdapter(adapter);
|
||||
activationSpec.validate();
|
||||
|
||||
MessageEndpointFactory messageEndpointFactory = new MessageEndpointFactory() {
|
||||
public MessageEndpoint createEndpoint(XAResource resource) throws UnavailableException {
|
||||
endpoint.xaresource = resource;
|
||||
return endpoint;
|
||||
}
|
||||
|
||||
public boolean isDeliveryTransacted(Method method) throws NoSuchMethodException {
|
||||
return true;
|
||||
}
|
||||
};
|
||||
|
||||
// Activate an Endpoint
|
||||
adapter.endpointActivation(messageEndpointFactory, activationSpec);
|
||||
|
||||
ActiveMQMessage msg = (ActiveMQMessage)advisory.receive(1000);
|
||||
if (msg != null) {
|
||||
assertEquals("Prefetch size hasn't been set", 1, ((ConsumerInfo)msg.getDataStructure()).getPrefetchSize());
|
||||
} else {
|
||||
fail("Consumer hasn't been created");
|
||||
}
|
||||
|
||||
// Send the broker a message to that endpoint
|
||||
MessageProducer producer = session.createProducer(new ActiveMQQueue("TEST"));
|
||||
producer.send(session.createTextMessage("Hello!"));
|
||||
|
||||
connection.close();
|
||||
|
||||
// Wait for the message to be delivered.
|
||||
assertTrue(messageDelivered.await(5000, TimeUnit.MILLISECONDS));
|
||||
|
||||
// Shut the Endpoint down.
|
||||
adapter.endpointDeactivation(messageEndpointFactory, activationSpec);
|
||||
adapter.stop();
|
||||
}
|
||||
|
||||
public void testMessageDelivery() throws Exception {
|
||||
|
||||
ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("vm://localhost?broker.persistent=false");
|
||||
|
|
|
@ -0,0 +1,38 @@
|
|||
## ---------------------------------------------------------------------------
|
||||
## 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: jndi
|
||||
|
||||
java.naming.factory.initial = org.apache.activemq.jndi.ActiveMQInitialContextFactory
|
||||
|
||||
# use the following property to configure the default connector
|
||||
java.naming.provider.url = vm://localhost
|
||||
|
||||
# use the following property to specify the JNDI name the connection factory
|
||||
# should appear as.
|
||||
#connectionFactoryNames = connectionFactory, queueConnectionFactory, topicConnectionFactry
|
||||
|
||||
# register some queues in JNDI using the form
|
||||
# queue.[jndiName] = [physicalName]
|
||||
queue.MyQueue = TEST
|
||||
|
||||
|
||||
# register some topics in JNDI using the form
|
||||
# topic.[jndiName] = [physicalName]
|
||||
topic.MyTopic = example.MyTopic
|
||||
|
||||
# END SNIPPET: jndi
|
Loading…
Reference in New Issue