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
|
/**
* User: Robert Greig
* Date: 12-Dec-2006
******************************************************************************
* (c) Copyright JP Morgan Chase Ltd 2006. All rights reserved. No part of
* this program may be photocopied reproduced or translated to another
* program language without prior written consent of JP Morgan Chase Ltd
******************************************************************************/
package org.apache.qpid.test.unit.message;
import junit.framework.TestCase;
import org.apache.log4j.Logger;
import org.apache.qpid.client.AMQConnection;
import org.apache.qpid.client.AMQHeadersExchange;
import org.apache.qpid.client.AMQSession;
import org.apache.qpid.client.AMQQueue;
import org.apache.qpid.client.transport.TransportConnection;
import org.apache.qpid.exchange.ExchangeDefaults;
import org.apache.qpid.framing.FieldTable;
import org.apache.qpid.framing.PropertyFieldTable;
import org.apache.qpid.url.AMQBindingURL;
import org.apache.qpid.url.BindingURL;
import javax.jms.*;
/**
* @author Apache Software Foundation
*/
public class StreamMessageTest extends TestCase
{
private static final Logger _logger = Logger.getLogger(StreamMessageTest.class);
public String _connectionString = "vm://:1";
protected void setUp() throws Exception
{
super.setUp();
TransportConnection.createVMBroker(1);
}
protected void tearDown() throws Exception
{
super.tearDown();
TransportConnection.killAllVMBrokers();
}
public void testStreamMessageEOF() throws Exception
{
Connection con = new AMQConnection("vm://:1", "guest", "guest", "consumer1", "/test");
AMQSession consumerSession = (AMQSession) con.createSession(false, Session.CLIENT_ACKNOWLEDGE);
AMQHeadersExchange queue = new AMQHeadersExchange(new AMQBindingURL(ExchangeDefaults.HEADERS_EXCHANGE_CLASS+"://"+ExchangeDefaults.HEADERS_EXCHANGE_NAME+"/test/queue1?"+ BindingURL.OPTION_ROUTING_KEY+"='F0000=1'"));
FieldTable ft = new PropertyFieldTable();
ft.setString("F1000","1");
MessageConsumer consumer = consumerSession.createConsumer(queue, AMQSession.DEFAULT_PREFETCH_LOW_MARK, AMQSession.DEFAULT_PREFETCH_HIGH_MARK, false, false, (String)null, ft);
//force synch to ensure the consumer has resulted in a bound queue
((AMQSession) consumerSession).declareExchangeSynch(ExchangeDefaults.HEADERS_EXCHANGE_NAME, ExchangeDefaults.HEADERS_EXCHANGE_CLASS);
Connection con2 = new AMQConnection("vm://:1", "guest", "guest", "producer1", "/test");
AMQSession producerSession = (AMQSession) con2.createSession(false, Session.CLIENT_ACKNOWLEDGE);
// Need to start the "producer" connection in order to receive bounced messages
_logger.info("Starting producer connection");
con2.start();
MessageProducer mandatoryProducer = producerSession.createProducer(queue);
// Third test - should be routed
_logger.info("Sending isBound message");
StreamMessage msg = producerSession.createStreamMessage();
msg.setStringProperty("F1000","1");
msg.writeByte((byte)42);
mandatoryProducer.send(msg);
_logger.info("Starting consumer connection");
con.start();
StreamMessage msg2 = (StreamMessage) consumer.receive();
byte b1 = msg2.readByte();
try
{
byte b2 = msg2.readByte();
}
catch (Exception e)
{
assertTrue("Expected MessageEOFException: " + e, e instanceof MessageEOFException);
}
}
public void testModifyReceivedMessageExpandsBuffer() throws Exception
{
Connection con = new AMQConnection("vm://:1", "guest", "guest", "consumer1", "/test");
AMQSession consumerSession = (AMQSession) con.createSession(false, Session.CLIENT_ACKNOWLEDGE);
AMQQueue queue = new AMQQueue("testQ");
MessageConsumer consumer = consumerSession.createConsumer(queue);
consumer.setMessageListener(new MessageListener()
{
public void onMessage(Message message)
{
StreamMessage sm = (StreamMessage) message;
try
{
sm.clearBody();
sm.writeString("dfgjshfslfjshflsjfdlsjfhdsljkfhdsljkfhsd");
}
catch (JMSException e)
{
_logger.error("Error when writing large string to received msg: " + e, e);
fail("Error when writing large string to received msg" + e);
}
}
});
Connection con2 = new AMQConnection("vm://:1", "guest", "guest", "producer1", "/test");
AMQSession producerSession = (AMQSession) con2.createSession(false, Session.CLIENT_ACKNOWLEDGE);
MessageProducer mandatoryProducer = producerSession.createProducer(queue);
con.start();
StreamMessage sm = producerSession.createStreamMessage();
sm.writeInt(42);
mandatoryProducer.send(sm);
Thread.sleep(2000);
}
}
|
