1 /*------------------------------------------------------------------------------
2 Name: TestPtDQueueRedeliver.java
3 Project: xmlBlaster.org
4 Copyright: xmlBlaster.org, see xmlBlaster-LICENSE file
5 ------------------------------------------------------------------------------*/
6 package org.xmlBlaster.test.qos;
7
8 import java.util.logging.Logger;
9
10 import junit.framework.Test;
11 import junit.framework.TestCase;
12 import junit.framework.TestSuite;
13
14 import org.xmlBlaster.client.I_XmlBlasterAccess;
15 import org.xmlBlaster.client.key.PublishKey;
16 import org.xmlBlaster.client.qos.ConnectQos;
17 import org.xmlBlaster.client.qos.ConnectReturnQos;
18 import org.xmlBlaster.client.qos.PublishQos;
19 import org.xmlBlaster.client.qos.PublishReturnQos;
20 import org.xmlBlaster.test.MsgInterceptor;
21 import org.xmlBlaster.test.Util;
22 import org.xmlBlaster.util.EmbeddedXmlBlaster;
23 import org.xmlBlaster.util.Global;
24 import org.xmlBlaster.util.MsgUnit;
25 import org.xmlBlaster.util.SessionName;
26 import org.xmlBlaster.util.XmlBlasterException;
27 import org.xmlBlaster.util.def.Constants;
28 import org.xmlBlaster.util.def.PriorityEnum;
29 import org.xmlBlaster.util.qos.TopicProperty;
30 import org.xmlBlaster.util.qos.address.CallbackAddress;
31 import org.xmlBlaster.util.qos.address.Destination;
32
33
34 /**
35 * This client does test if a subscriber can reconnect to its session and
36 * its callback queue holded the messages during downtime.
37 * <p>
38 * See method testPersistentPtp() for a description.
39 * </p>
40 * <p>
41 * This client may be invoked multiple time on the same xmlBlaster server,
42 * as it cleans up everything after his tests are done.
43 * </p>
44 * <p>
45 * Invoke examples:
46 * </p>
47 * <pre>
48 * java junit.textui.TestRunner org.xmlBlaster.test.qos.TestPtDQueueRedeliver
49 * java junit.swingui.TestRunner -noloading org.xmlBlaster.test.qos.TestPtDQueueRedeliver
50 * </pre>
51 */
52 public class TestPtDQueueRedeliver extends TestCase
53 {
54 private static String ME = "TestPtDQueueRedeliver";
55 private final Global glob;
56 private static Logger log = Logger.getLogger(TestPtDQueueRedeliver.class.getName());
57 private String passwd = "secret";
58 private int serverPort = 7615;
59 private String oid = "TestPtDQueueRedeliver.Msg";
60 private EmbeddedXmlBlaster serverThread = null;
61 private String sessionNameRcv = "TestPtDQueueRedeliverReceiver";
62 private I_XmlBlasterAccess conRcv;
63 private boolean connectedRcv = false;
64 private MsgInterceptor updateInterceptorRcv;
65
66 private String sessionNameSnd = "TestPtDQueueRedeliverSender";
67 private I_XmlBlasterAccess conSnd;
68 private MsgInterceptor updateInterceptorSnd;
69
70 /** For Junit */
71 public TestPtDQueueRedeliver() {
72 this(new Global(), "TestPtDQueueRedeliver");
73 }
74
75 /**
76 * Constructs the TestPtDQueueRedeliver object.
77 * <p />
78 * @param testName The name used in the test suite and to login to xmlBlaster
79 */
80 public TestPtDQueueRedeliver(Global glob, String testName) {
81 super(testName);
82 this.glob = glob;
83
84 }
85
86 /**
87 * Sets up the fixture.
88 * <p />
89 * Connect to xmlBlaster and login
90 */
91 protected void setUp() {
92 glob.init(Util.getOtherServerPorts(serverPort));
93 serverThread = EmbeddedXmlBlaster.startXmlBlaster(glob);
94 log.info("XmlBlaster is ready for testing");
95 }
96
97 /**
98 * Cleaning up.
99 */
100 protected void tearDown() {
101 try { Thread.sleep(1000);} catch(Exception ex) {}
102 if (serverThread != null)
103 serverThread.stopServer(true);
104 // reset to default server port (necessary if other tests follow in the same JVM).
105 Util.resetPorts();
106 }
107
108 /**
109 * <p>
110 * 1. xmlBlaster starts and sender sends persistent and forceQueuing PtP message.
111 * </p>
112 * <p>
113 * 2. xmlBlaster stops and starts again
114 * </p>
115 * <p>
116 * 3. receiver start and should receive the message
117 * </p>
118 */
119 public void testPersistentPtp() {
120 log.info("testPersistentPtp("+sessionNameRcv+") ...");
121
122 try {
123
124 log.info("============ STEP 1: Start publisher");
125 conSnd = glob.getXmlBlasterAccess();
126 ConnectQos qosPub = new ConnectQos(glob);
127 ConnectReturnQos crqPub = conSnd.connect(qosPub, null); // Login to xmlBlaster, no updates
128 log.info("Connect success as " + crqPub.getSessionName());
129
130 int numPub = 8;
131 log.info("============ STEP 2: Publish " + numPub + " PtP messages");
132 MsgUnit[] sentArr = new MsgUnit[numPub];
133 PublishReturnQos[] sentQos = new PublishReturnQos[numPub];
134 for(int i=0; i<numPub; i++) {
135 PublishKey pk = new PublishKey(glob, oid, "text/xml", "1.0");
136 pk.setClientTags("<org.xmlBlaster><demo/></org.xmlBlaster>");
137
138 PublishQos pq = new PublishQos(glob);
139 pq.setPriority(PriorityEnum.NORM_PRIORITY);
140 pq.setPersistent(true);
141 Destination dest = new Destination(glob, new SessionName(glob, sessionNameRcv));
142 dest.forceQueuing(true);
143 pq.addDestination(dest);
144 pq.setForceUpdate(true);
145 pq.setSubscribable(false);
146 pq.setLifeTime(60000L);
147
148 if (i == 0) {
149 TopicProperty topicProperty = new TopicProperty(glob);
150 topicProperty.setDestroyDelay(60000L);
151 topicProperty.setCreateDomEntry(true);
152 topicProperty.setReadonly(false);
153 topicProperty.getHistoryQueueProperty().setMaxEntries(numPub+5);
154 pq.setTopicProperty(topicProperty);
155 log.info("Added TopicProperty on first publish: " + topicProperty.toXml());
156 }
157
158 byte[] content = "Hello".getBytes();
159 MsgUnit msgUnit = new MsgUnit(pk, content, pq);
160 sentArr[i] = msgUnit;
161 PublishReturnQos prq = conSnd.publish(msgUnit);
162 sentQos[i] = prq;
163 log.info("Got status='" + prq.getState() + "' rcvTimestamp=" + prq.getRcvTimestamp().toString() +
164 " for published message '" + prq.getKeyOid() + "'");
165 }
166
167 log.info("============ STEP 3: Stop xmlBlaster");
168 this.serverThread.stopServer(true);
169
170 log.info("============ STEP 4: Start xmlBlaster");
171 glob.init(Util.getOtherServerPorts(serverPort));
172 serverThread = EmbeddedXmlBlaster.startXmlBlaster(glob);
173 log.info("XmlBlaster is ready for testing");
174
175 log.info("============ STEP 5: Start subscriber");
176 // A testsuite helper to collect update messages
177 this.updateInterceptorRcv = new MsgInterceptor(glob, log, null);
178
179 Global globRcv = glob.getClone(null);
180 conRcv = globRcv.getXmlBlasterAccess();
181
182 ConnectQos qosSub = new ConnectQos(globRcv, sessionNameRcv, passwd);
183
184 CallbackAddress addr = new CallbackAddress(globRcv);
185 addr.setRetries(-1);
186 String secretCbSessionId = "TrustMeSub";
187 addr.setSecretCbSessionId(secretCbSessionId);
188 qosSub.getSessionCbQueueProperty().setCallbackAddress(addr);
189
190 ConnectReturnQos crqSub = conRcv.connect(qosSub, this.updateInterceptorRcv); // Login to xmlBlaster
191 log.info("Connect as subscriber '" + crqSub.getSessionName() + "' success");
192 log.info(this.updateInterceptorRcv.toString());
193
194 /*
195 SubscribeKey sk = new SubscribeKey(globRcv, oid);
196 SubscribeQos sq = new SubscribeQos(globRcv);
197 sq.setWantInitialUpdate(false);
198 sq.setWantLocal(true);
199 sq.setWantContent(true);
200
201 HistoryQos historyQos = new HistoryQos(globRcv);
202 historyQos.setNumEntries(1);
203 sq.setHistoryQos(historyQos);
204
205 SubscribeReturnQos srq = conRcv.subscribe(sk.toXml(), sq.toXml());
206 log.info("Subscription to '" + oid + "' done");
207 */
208
209 log.info("============ STEP 6: Check if messages arrived");
210 assertEquals("", numPub, this.updateInterceptorRcv.waitOnUpdate(4000L, oid, Constants.STATE_OK));
211 this.updateInterceptorRcv.compareToReceived(sentArr, secretCbSessionId);
212 this.updateInterceptorRcv.compareToReceived(sentQos);
213
214 this.updateInterceptorRcv.clear();
215 }
216 catch (XmlBlasterException e) {
217 log.severe(e.toString());
218 e.printStackTrace();
219 fail(e.toString());
220 }
221 finally { // clean up
222 log.info("Disconnecting '" + sessionNameRcv + "'");
223 if (conRcv != null) conRcv.disconnect(null);
224 }
225 log.info("Success in testPersistentPtp()");
226 }
227
228 /**
229 * Method is used by TestRunner to load these tests
230 */
231 public static Test suite() {
232 TestSuite suite= new TestSuite();
233 String loginName = "TestPtDQueueRedeliver";
234 suite.addTest(new TestPtDQueueRedeliver(Global.instance(), "testPersistentPtp"));
235 return suite;
236 }
237
238 /**
239 * Invoke:
240 * <pre>
241 * java org.xmlBlaster.test.qos.TestPtDQueueRedeliver
242 * java -Djava.compiler= junit.textui.TestRunner org.xmlBlaster.test.qos.TestPtDQueueRedeliver
243 * <pre>
244 */
245 public static void main(String args[]) {
246 TestPtDQueueRedeliver testSub = new TestPtDQueueRedeliver(new Global(args), "TestPtDQueueRedeliver");
247 testSub.setUp();
248 testSub.testPersistentPtp();
249 testSub.tearDown();
250 }
251 }
syntax highlighted by Code2HTML, v. 0.9.1