added a test case and a fix for the use of temporary destinations in NMS. Fixes AMQ-843

git-svn-id: https://svn.apache.org/repos/asf/incubator/activemq/trunk@428235 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
James Strachan 2006-08-03 04:44:17 +00:00
parent df09548bde
commit b62e5cd91e
5 changed files with 120 additions and 7 deletions

View File

@ -128,14 +128,37 @@ namespace ActiveMQ
public ITemporaryQueue CreateTemporaryQueue()
{
return new ActiveMQTempQueue(connection.CreateTemporaryDestinationName());
ActiveMQTempQueue answer = new ActiveMQTempQueue(connection.CreateTemporaryDestinationName());
CreateTemporaryDestination(answer);
return answer;
}
public ITemporaryTopic CreateTemporaryTopic()
{
return new ActiveMQTempTopic(connection.CreateTemporaryDestinationName());
ActiveMQTempTopic answer = new ActiveMQTempTopic(connection.CreateTemporaryDestinationName());
CreateTemporaryDestination(answer);
return answer;
}
protected void CreateTemporaryDestination(ActiveMQDestination tempDestination)
{
DestinationInfo command = new DestinationInfo();
command.ConnectionId = connection.ConnectionId;
command.OperationType = 0; // 0 is add
command.Destination = tempDestination;
connection.SyncRequest(command);
}
protected void DestroyTemporaryDestination(ActiveMQDestination tempDestination)
{
DestinationInfo command = new DestinationInfo();
command.ConnectionId = connection.ConnectionId;
command.OperationType = 1; // 1 is remove
command.Destination = tempDestination;
connection.SyncRequest(command);
}
public IMessage CreateMessage()

View File

@ -22,6 +22,6 @@ using System.Runtime.InteropServices;
[assembly: AssemblyCopyrightAttribute("Copyright (C) 2005-2006 Apache Software Foundation")]
[assembly: AssemblyTrademarkAttribute("")]
[assembly: AssemblyCultureAttribute("")]
[assembly: AssemblyVersionAttribute("4.0.2281.0")]
[assembly: AssemblyVersionAttribute("4.0.2406.0")]
[assembly: AssemblyInformationalVersionAttribute("4.0")]

View File

@ -22,6 +22,6 @@ using System.Runtime.InteropServices;
[assembly: AssemblyCopyrightAttribute("Copyright (C) 2005-2006 Apache Software Foundation")]
[assembly: AssemblyTrademarkAttribute("")]
[assembly: AssemblyCultureAttribute("")]
[assembly: AssemblyVersionAttribute("4.0.2281.0")]
[assembly: AssemblyVersionAttribute("4.0.2406.0")]
[assembly: AssemblyInformationalVersionAttribute("4.0")]

View File

@ -155,7 +155,7 @@ namespace NMS
protected virtual IConnectionFactory CreateConnectionFactory()
{
return new ActiveMQ.ConnectionFactory(new Uri("tcp://localhost:61616"));
return new ActiveMQ.ConnectionFactory(new Uri("tcp://localhost:61616?logging=true"));
}
protected virtual IConnection CreateConnection()
@ -169,13 +169,13 @@ namespace NMS
protected virtual IMessageProducer CreateProducer()
{
IMessageProducer producer = Session.CreateProducer(destination);
IMessageProducer producer = Session.CreateProducer(Destination);
return producer;
}
protected virtual IMessageConsumer CreateConsumer()
{
IMessageConsumer consumer = Session.CreateConsumer(destination);
IMessageConsumer consumer = Session.CreateConsumer(Destination);
return consumer;
}

View File

@ -0,0 +1,90 @@
/*
* 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.
*/
using NMS;
using NUnit.Framework;
using System;
using System.Threading;
namespace NMS {
[ TestFixture ]
public class TemporaryQueueTest : JMSTestSupport {
protected Object semaphore = new Object();
protected bool received;
[ SetUp ]
override public void SetUp() {
base.SetUp();
}
[ TearDown ]
override public void TearDown() {
base.TearDown();
}
[ Test ]
public void TestAsynchronousConsume() {
// lets consume to a regular queue
IMessageConsumer consumer = CreateConsumer();
consumer.Listener += new MessageListener(OnQueueMessage);
// lets create a temporary queue and a consumer on it
ITemporaryQueue tempQ = Session.CreateTemporaryQueue();
IMessageConsumer tempQueueConsumer = Session.CreateConsumer(tempQ);
tempQueueConsumer.Listener += new MessageListener(OnTempQueueMessage);
// Send message to queue which has a listener to reply to the temporary queue
IMessageProducer producer = CreateProducer();
IMessage request = CreateMessage();
request.NMSCorrelationID = "abc";
request.NMSReplyTo = tempQ;
request.NMSPersistent = false;
producer.Send(request);
// now lets wait for the message to arrive on the temporary queue
WaitForMessageToArrive();
}
protected void OnQueueMessage(IMessage message) {
Console.WriteLine("First message received: " + message + " so about to reply to temporary queue");
ITextMessage response = Session.CreateTextMessage("this is a response!!");
response.NMSCorrelationID = message.NMSCorrelationID;
IMessageProducer producerTempQ = Session.CreateProducer(message.NMSReplyTo);
//Write msg to temp q.
producerTempQ.Send(response);
}
protected void OnTempQueueMessage(IMessage message) {
Console.WriteLine("Received message on temporary queue: " + message);
lock (semaphore) {
received = true;
Monitor.PulseAll(semaphore);
}
}
protected void WaitForMessageToArrive() {
lock (semaphore) {
if (!received) {
Monitor.Wait(semaphore, receiveTimeout);
}
Assert.AreEqual(true, received, "Should have received a message by now!");
}
}
}
}