1 /*------------------------------------------------------------------------------
2 Name: TestPtPSubscribable.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 import java.util.logging.Level;
10 import org.xmlBlaster.util.Global;
11 import org.xmlBlaster.util.SessionName;
12 import org.xmlBlaster.util.qos.address.Destination;
13 import org.xmlBlaster.util.def.Constants;
14 import org.xmlBlaster.util.qos.address.CallbackAddress;
15 import org.xmlBlaster.client.qos.ConnectQos;
16 import org.xmlBlaster.client.qos.ConnectReturnQos;
17 import org.xmlBlaster.client.qos.DisconnectQos;
18 import org.xmlBlaster.util.XmlBlasterException;
19 import org.xmlBlaster.util.EmbeddedXmlBlaster;
20 import org.xmlBlaster.client.key.PublishKey;
21 import org.xmlBlaster.client.key.SubscribeKey;
22 import org.xmlBlaster.client.qos.PublishQos;
23 import org.xmlBlaster.client.qos.PublishReturnQos;
24 import org.xmlBlaster.client.qos.SubscribeQos;
25 import org.xmlBlaster.client.qos.SubscribeReturnQos;
26 import org.xmlBlaster.client.I_XmlBlasterAccess;
27 import org.xmlBlaster.util.MsgUnit;
28
29 import org.xmlBlaster.test.Util;
30 import org.xmlBlaster.test.Msg;
31 import org.xmlBlaster.test.MsgInterceptor;
32
33 import junit.framework.*;
34
35
36 /**
37 * Here we test how to make PtP messages invisible to subscribers using the <i>subscribable</i> QoS.
38 * <p>
39 * </p>
40 * <p>
41 * Invoke examples:
42 * </p>
43 * <pre>
44 * java junit.textui.TestRunner org.xmlBlaster.test.qos.TestPtPSubscribable
45 * java junit.swingui.TestRunner -noloading org.xmlBlaster.test.qos.TestPtPSubscribable
46 * </pre>
47 */
48 public class TestPtPSubscribable extends TestCase
49 {
50 private static String ME = "TestPtPSubscribable";
51 private final Global glob;
52 private static Logger log = Logger.getLogger(TestPtPSubscribable.class.getName());
53 private String passwd = "secret";
54 private int serverPort = 7615;
55 private String oid = "TestPtPSubscribable.Msg";
56 private EmbeddedXmlBlaster serverThread = null;
57 private String sessionNameRcv = "TestPtPSubscribableReceiver";
58 private I_XmlBlasterAccess conRcv;
59 private boolean connectedRcv = false;
60 private MsgInterceptor updateInterceptorRcv;
61
62 private String sessionNameSnd = "TestPtPSubscribableSender";
63 private I_XmlBlasterAccess conSnd;
64 private MsgInterceptor updateInterceptorSnd;
65
66 /** For Junit */
67 public TestPtPSubscribable() {
68 this(new Global(), "TestPtPSubscribable");
69 }
70
71 /**
72 * Constructs the TestPtPSubscribable object.
73 * <p />
74 * @param testName The name used in the test suite and to login to xmlBlaster
75 */
76 public TestPtPSubscribable(Global glob, String testName) {
77 super(testName);
78 this.glob = glob;
79
80 }
81
82 /**
83 * Sets up the fixture.
84 * <p />
85 * Connect to xmlBlaster and login
86 */
87 protected void setUp() {
88 glob.init(Util.getOtherServerPorts(serverPort));
89 serverThread = EmbeddedXmlBlaster.startXmlBlaster(glob);
90 log.info("XmlBlaster is ready for testing");
91 }
92
93 /**
94 * Cleaning up.
95 */
96 protected void tearDown() {
97 try { Thread.sleep(1000);} catch(Exception ex) {}
98 if (serverThread != null)
99 serverThread.stopServer(true);
100 // reset to default server port (necessary if other tests follow in the same JVM).
101 Util.resetPorts();
102 }
103
104 /**
105 * <p>
106 * 1. xmlBlaster starts and sender sends persistent and forceQueuing PtP message.
107 * </p>
108 * <p>
109 * 2. xmlBlaster stops and starts again
110 * </p>
111 * <p>
112 * 3. receiver start and should receive the message
113 * </p>
114 */
115 public void testSubscribable() {
116 log.info("testSubscribable("+sessionNameRcv+") ...");
117
118 try {
119 log.info("============ STEP 1: Start publisher client");
120 Global globSnd = glob.getClone(null);
121 conSnd = globSnd.getXmlBlasterAccess();
122 ConnectQos qosSnd = new ConnectQos(globSnd);
123 String secretSndCbSessionId = "TrustMeSubSnd";
124 CallbackAddress addrSnd = new CallbackAddress(globSnd);
125 addrSnd.setSecretCbSessionId(secretSndCbSessionId);
126 qosSnd.getSessionCbQueueProperty().setCallbackAddress(addrSnd);
127 this.updateInterceptorSnd = new MsgInterceptor(globSnd, log, null);
128 ConnectReturnQos crqPub = conSnd.connect(qosSnd, this.updateInterceptorSnd);
129 log.info("Connect success as " + crqPub.getSessionName());
130
131 log.info("============ STEP 2: Subscribe in Pub/Sub mode");
132 SubscribeKey sk = new SubscribeKey(globSnd, oid);
133 SubscribeQos sq = new SubscribeQos(globSnd);
134 sq.setWantInitialUpdate(false);
135 sq.setWantLocal(true);
136 SubscribeReturnQos srq = conSnd.subscribe(sk, sq);
137 log.info("Subscription to '" + oid + "' done");
138 assertEquals("", 0, this.updateInterceptorSnd.waitOnUpdate(1000L, oid, Constants.STATE_OK));
139
140 log.info("============ STEP 3: Start receiver");
141 Global globRcv = glob.getClone(null);
142 conRcv = globRcv.getXmlBlasterAccess();
143 ConnectQos qosRcv = new ConnectQos(globRcv, sessionNameRcv, passwd);
144 CallbackAddress addr = new CallbackAddress(globRcv);
145 addr.setRetries(-1);
146 String secretCbSessionId = "TrustMeSub";
147 addr.setSecretCbSessionId(secretCbSessionId);
148 qosRcv.getSessionCbQueueProperty().setCallbackAddress(addr);
149 this.updateInterceptorRcv = new MsgInterceptor(globRcv, log, null);
150 ConnectReturnQos crqRcv = conRcv.connect(qosRcv, this.updateInterceptorRcv); // Login to xmlBlaster
151 log.info("Connect as subscriber '" + crqRcv.getSessionName() + "' success");
152
153 {
154 log.info("============ STEP 4: Publish PtP message which is NOT subscribable");
155 PublishKey pk = new PublishKey(globSnd, oid, "text/xml", "1.0");
156 PublishQos pq = new PublishQos(globSnd);
157 Destination dest = new Destination(globSnd, new SessionName(globSnd, sessionNameRcv));
158 pq.addDestination(dest);
159 pq.setSubscribable(false);
160 byte[] content = "Hello".getBytes();
161 MsgUnit msgUnit = new MsgUnit(pk, content, pq);
162 PublishReturnQos prq = conSnd.publish(msgUnit);
163 log.info("Got status='" + prq.getState() + "' rcvTimestamp=" + prq.getRcvTimestamp().toString() +
164 " for published message '" + prq.getKeyOid() + "'");
165 assertEquals("", 1, this.updateInterceptorRcv.waitOnUpdate(1000L, oid, Constants.STATE_OK));
166 assertEquals("", secretCbSessionId, this.updateInterceptorRcv.getMsg(oid, Constants.STATE_OK).getCbSessionId());
167 assertEquals("", 0, this.updateInterceptorSnd.waitOnUpdate(1000L, oid, Constants.STATE_OK));
168
169 this.updateInterceptorRcv.clear();
170 this.updateInterceptorSnd.clear();
171 }
172
173 {
174 log.info("============ STEP 5: Publish PtP message which IS subscribable");
175 PublishKey pk = new PublishKey(globSnd, oid, "text/xml", "1.0");
176 PublishQos pq = new PublishQos(globSnd);
177 Destination dest = new Destination(globSnd, new SessionName(globSnd, sessionNameRcv));
178 pq.addDestination(dest);
179 pq.setSubscribable(true);
180 byte[] content = "Hello".getBytes();
181 MsgUnit msgUnit = new MsgUnit(pk, content, pq);
182 PublishReturnQos prq = conSnd.publish(msgUnit);
183 log.info("Got status='" + prq.getState() + "' rcvTimestamp=" + prq.getRcvTimestamp().toString() +
184 " for published message '" + prq.getKeyOid() + "'");
185
186 assertEquals("", 1, this.updateInterceptorRcv.waitOnUpdate(1000L, oid, Constants.STATE_OK));
187 assertEquals("", secretCbSessionId, this.updateInterceptorRcv.getMsg(oid, Constants.STATE_OK).getCbSessionId());
188 assertEquals("", 1, this.updateInterceptorSnd.waitOnUpdate(1000L, oid, Constants.STATE_OK));
189 assertEquals("", secretSndCbSessionId, this.updateInterceptorSnd.getMsg(oid, Constants.STATE_OK).getCbSessionId());
190
191 this.updateInterceptorRcv.clear();
192 this.updateInterceptorSnd.clear();
193 }
194 }
195 catch (XmlBlasterException e) {
196 log.severe(e.toString());
197 fail(e.toString());
198 }
199 finally { // clean up
200 log.info("Disconnecting '" + sessionNameRcv + "'");
201 if (conRcv != null) conRcv.disconnect(null);
202 if (conSnd != null) conSnd.disconnect(null);
203 }
204 log.info("Success in testSubscribable()");
205 }
206
207 /**
208 * Method is used by TestRunner to load these tests
209 */
210 public static Test suite() {
211 TestSuite suite= new TestSuite();
212 String loginName = "TestPtPSubscribable";
213 suite.addTest(new TestPtPSubscribable(Global.instance(), "testSubscribable"));
214 return suite;
215 }
216
217 /**
218 * Invoke:
219 * <pre>
220 * java org.xmlBlaster.test.qos.TestPtPSubscribable
221 * java -Djava.compiler= junit.textui.TestRunner org.xmlBlaster.test.qos.TestPtPSubscribable
222 * <pre>
223 */
224 public static void main(String args[]) {
225 TestPtPSubscribable testSub = new TestPtPSubscribable(new Global(args), "TestPtPSubscribable");
226 testSub.setUp();
227 testSub.testSubscribable();
228 testSub.tearDown();
229 }
230 }
syntax highlighted by Code2HTML, v. 0.9.1