1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
|
/*
*
* 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.qpid.test.unit.ack;
import org.apache.qpid.client.AMQDestination;
import org.apache.qpid.client.AMQSession;
import org.apache.qpid.test.utils.QpidBrokerTestCase;
import javax.jms.Connection;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.Queue;
import javax.jms.Session;
import javax.jms.MessageProducer;
/**
* Test the various JMS Acknowledge Modes the single testAcking method does all
* the work of receiving and validation of acking.
*
* The ack mode is provided from the various test methods.
*/
public class AcknowledgeTest extends QpidBrokerTestCase
{
protected int NUM_MESSAGES;
protected Connection _connection;
protected Queue _queue;
protected Session _consumerSession;
protected MessageConsumer _consumer;
protected MessageProducer _producer;
@Override
protected void setUp() throws Exception
{
super.setUp();
NUM_MESSAGES = 5;
_queue = getTestQueue();
_logger.info("AT: setup");
//Create Producer put some messages on the queue
_connection = getConnection();
}
protected void init(boolean transacted, int mode) throws Exception
{
_consumerSession = _connection.createSession(transacted, mode);
_consumer = _consumerSession.createConsumer(_queue);
_producer = _consumerSession.createProducer(_queue);
// These should all end up being prefetched by session
sendMessage(_consumerSession, _queue, 1);
if(!transacted)
{
((AMQSession)_consumerSession).sync();
}
assertEquals("Wrong number of messages on queue", 1,
((AMQSession) _consumerSession).getQueueDepth((AMQDestination) _queue));
}
/**
* The main test method.
*
* Receive the initial message and then proceed to send and ack messages
* until we have processed NUM_MESSAGES worth of messages.
*
* Each message is tagged with an INDEX value and these are used to check
* that the messages are received in the correct order.
*
* The test concludes by validating that the queue depth is 0 as expected.
*
* @param transacted
* @param mode
*
* @throws Exception
*/
protected void testAcking(boolean transacted, int mode) throws Exception
{
init(transacted, mode);
_connection.start();
Message msg = _consumer.receive(1500);
int count = 0;
while (count < NUM_MESSAGES)
{
assertNotNull("Message " + count + " not correctly received.", msg);
assertEquals("Incorrect message received", count, msg.getIntProperty(INDEX));
count++;
if (count < NUM_MESSAGES)
{
//Send the next message
_producer.send(createNextMessage(_consumerSession, count));
}
doAcknowlegement(msg);
msg = _consumer.receive(1500);
}
if (_consumerSession.getTransacted())
{
//Acknowledge the last msg if we are testing transacted otherwise queueDepth will be 1
doAcknowlegement(msg);
}
assertEquals("Wrong number of messages on queue", 0,
((AMQSession) _consumerSession).getQueueDepth((AMQDestination) _queue));
}
/**
* Perform the acknowledgement of messages if additionally required.
*
* @param msg
*
* @throws JMSException
*/
protected void doAcknowlegement(Message msg) throws JMSException
{
if (_consumerSession.getTransacted())
{
_consumerSession.commit();
}
if (_consumerSession.getAcknowledgeMode() == Session.CLIENT_ACKNOWLEDGE)
{
msg.acknowledge();
}
}
public void testClientAck() throws Exception
{
testAcking(false, Session.CLIENT_ACKNOWLEDGE);
}
public void testAutoAck() throws Exception
{
testAcking(false, Session.AUTO_ACKNOWLEDGE);
}
public void testTransacted() throws Exception
{
testAcking(true, Session.SESSION_TRANSACTED);
}
public void testDupsOk() throws Exception
{
testAcking(false, Session.DUPS_OK_ACKNOWLEDGE);
}
public void testNoAck() throws Exception
{
testAcking(false, AMQSession.NO_ACKNOWLEDGE);
}
public void testPreAck() throws Exception
{
testAcking(false, AMQSession.PRE_ACKNOWLEDGE);
}
}
|
