1 /*------------------------------------------------------------------------------
  2 Name:      TestTailback.java
  3 Project:   xmlBlaster.org
  4 Copyright: xmlBlaster.org, see xmlBlaster-LICENSE file
  5 ------------------------------------------------------------------------------*/
  6 package org.xmlBlaster.test.client;
  7 
  8 
  9 import java.util.logging.Logger;
 10 import java.util.logging.Level;
 11 import org.xmlBlaster.util.Global;
 12 import org.xmlBlaster.util.XmlBlasterException;
 13 import org.xmlBlaster.util.def.ErrorCode;
 14 import org.xmlBlaster.util.EmbeddedXmlBlaster;
 15 import org.xmlBlaster.util.SessionName;
 16 import org.xmlBlaster.client.I_Callback;
 17 import org.xmlBlaster.client.I_XmlBlasterAccess;
 18 import org.xmlBlaster.client.key.UpdateKey;
 19 import org.xmlBlaster.client.key.PublishKey;
 20 import org.xmlBlaster.client.key.GetKey;
 21 import org.xmlBlaster.client.key.SubscribeKey;
 22 import org.xmlBlaster.client.key.UnSubscribeKey;
 23 import org.xmlBlaster.client.key.EraseKey;
 24 import org.xmlBlaster.client.qos.ConnectQos;
 25 import org.xmlBlaster.client.qos.GetQos;
 26 import org.xmlBlaster.client.qos.GetReturnQos;
 27 import org.xmlBlaster.client.qos.PublishQos;
 28 import org.xmlBlaster.client.qos.PublishReturnQos;
 29 import org.xmlBlaster.client.qos.UpdateQos;
 30 import org.xmlBlaster.client.qos.UpdateReturnQos;
 31 import org.xmlBlaster.client.qos.SubscribeQos;
 32 import org.xmlBlaster.client.qos.SubscribeReturnQos;
 33 import org.xmlBlaster.client.qos.EraseQos;
 34 import org.xmlBlaster.client.qos.EraseReturnQos;
 35 import org.xmlBlaster.client.qos.UnSubscribeQos;
 36 import org.xmlBlaster.client.qos.UnSubscribeReturnQos;
 37 import org.xmlBlaster.client.I_ConnectionStateListener;
 38 import org.xmlBlaster.util.dispatch.ConnectionStateEnum;
 39 import org.xmlBlaster.util.property.Property;
 40 import org.xmlBlaster.util.qos.address.Address;
 41 import org.xmlBlaster.util.qos.address.CallbackAddress;
 42 import org.xmlBlaster.util.MsgUnit;
 43 
 44 import org.xmlBlaster.test.Util;
 45 import org.xmlBlaster.test.Msg;
 46 import org.xmlBlaster.test.MsgInterceptor;
 47 import junit.framework.*;
 48 
 49 
 50 /**
 51  * Tests the correct reference counting for persistent messages
 52  * after recovery. 
 53  * <pre>
 54  *   1. Start xmlBlaster server, configure cache size to 2
 55  *   2. Publish two messages
 56  *   3. Subscribe to message in fail save mode and block in callback
 57  *   4. Kill client subscriber
 58  *   5. Publish two more message to swap the first two
 59  *   6. Start same subscriber without initial update - it will receive the messages from 3.
 60  *   7. Start another subscriber with history = 10: we expect 4 updates
 61  *   If the reference counter is not properly swapped and resored test 6. will fail.
 62  * </pre>
 63  * Invoke examples:<br />
 64  * <pre>
 65  *   java junit.textui.TestRunner org.xmlBlaster.test.client.TestTailback
 66  *   java junit.swingui.TestRunner -noloading org.xmlBlaster.test.client.TestTailback
 67  * </pre>
 68  */
 69 public class TestTailback extends TestCase implements I_ConnectionStateListener
 70 {
 71    private static String ME = "TestTailback";
 72    private Global glob;
 73    private static Logger log = Logger.getLogger(TestTailback.class.getName());
 74 
 75    private int serverPort = 7694;
 76    private EmbeddedXmlBlaster serverThread;
 77 
 78    private String oid = "tailbackMsg";
 79 
 80    class Client {
 81       I_XmlBlasterAccess con;
 82       MsgInterceptor updateInterceptor;
 83    }
 84 
 85    public TestTailback(String testName) {
 86       this(null, testName);
 87    }
 88 
 89    /**
 90     * Constructs the TestTailback object.
 91     * <p />
 92     * @param testName  The name used in the test suite
 93     */
 94    public TestTailback(Global glob, String testName) {
 95       super(testName);
 96       this.glob = glob;
 97    }
 98 
 99    /**
100     * Sets up the fixture.
101     * <p />
102     * Connect to xmlBlaster and login
103     */
104    protected void setUp() {
105       this.glob = (this.glob == null) ? new Global() : this.glob;
106 
107       this.glob.init(Util.getOtherServerPorts(serverPort));
108    }
109 
110    /**
111     * Tears down the fixture.
112     * <p />
113     * cleaning up .... erase() the previous message OID and logout
114     */
115    protected void tearDown() {
116       log.info("Entering tearDown(), test is finished");
117 
118       if (this.serverThread != null) {
119          EmbeddedXmlBlaster.stopXmlBlaster(this.serverThread);
120       }
121 
122       // reset to default server port (necessary if other tests follow in the same JVM).
123       Util.resetPorts(this.glob);
124       Global.instance().shutdown();
125       this.glob = null;
126      
127       this.serverThread = null;
128    }
129 
130    /**
131     * Create a new connection
132     * @param loginName The login name
133     * @param cb The callback handle or null
134     */
135    private Client doConnect(String loginName, I_Callback cb) {
136       try {
137          Client client = new Client();
138          Global gg = this.glob.getClone(null);
139          ConnectQos connectQos = new ConnectQos(gg);
140          SessionName sessionName = new SessionName(gg, loginName);
141          connectQos.setSessionName(sessionName);
142          client.con = gg.getXmlBlasterAccess();
143          client.con.registerConnectionListener(this);
144          client.updateInterceptor = new MsgInterceptor(gg, log, cb); // Collect received msgs
145          client.con.connect(connectQos, client.updateInterceptor); // Login to xmlBlaster
146          return client;
147       }
148       catch (XmlBlasterException e) {
149          log.warning("doConnect() - login failed: " + e.getMessage());
150          fail(ME+".doConnect() failed: " + e.getMessage());
151       }
152       return null;
153    }
154 
155    /**
156     * Subscribe to message. 
157     */
158    private void doSubscribe(I_XmlBlasterAccess con) {
159       if (log.isLoggable(Level.FINE)) log.fine("Subscribing using EXACT oid syntax ...");
160       try {
161          SubscribeKey subscribeKey = new SubscribeKey(con.getGlobal(), this.oid);
162          SubscribeQos subscribeQos = new SubscribeQos(con.getGlobal());
163          String subscribeOid = con.subscribe(subscribeKey, subscribeQos).getSubscriptionId();
164          log.info("Success: Subscribe on " + subscribeOid + " done");
165          assertTrue("returned null subscribeOid", subscribeOid != null);
166       } catch(XmlBlasterException e) {
167          log.severe("XmlBlasterException: " + e.getMessage());
168          fail(ME+".doSubscribe() failed: " + e.getMessage());
169       }
170    }
171 
172    /**
173     * Construct a message and publish it persistent. 
174     */
175    private void doPublish(I_XmlBlasterAccess con) {
176       if (log.isLoggable(Level.FINE)) log.fine("Publishing a message");
177       try {
178          PublishKey publishKey = new PublishKey(con.getGlobal(), this.oid);
179          PublishQos publishQos = new PublishQos(con.getGlobal());
180          publishQos.setPersistent(true);
181          String content = "Hi";
182          MsgUnit msgUnit = new MsgUnit(publishKey, content.getBytes(), publishQos);
183          con.publish(msgUnit);
184          log.info("Success: Publishing of " + this.oid + " done");
185       } catch(XmlBlasterException e) {
186          log.severe("XmlBlasterException: " + e.getMessage());
187          fail(ME+".doPublish() failed: " + e.getMessage());
188       }
189    }
190 
191    /**
192     * Erase the message. 
193     */
194    private void doErase(I_XmlBlasterAccess con) {
195       log.info("Erasing " + this.oid + " ...");
196       try {
197          EraseKey eraseKey = new EraseKey(con.getGlobal(), this.oid);
198          EraseQos eraseQos = new EraseQos(con.getGlobal());
199          eraseQos.setForceDestroy(true);
200          EraseReturnQos[] arr = con.erase(eraseKey, eraseQos);
201       }
202       catch(XmlBlasterException e) {
203          log.severe("XmlBlasterException: " + e.getMessage());
204          fail(ME+".doErase() failed: " + e.getMessage());
205       }
206    }
207 
208    /**
209     * Test as described in class javadoc. 
210     */
211    public void testTailback() {
212       log.info("testTailback START");
213 
214       log.info("STEP1: Publish a persistent message twice, will be tailed back as no xmlBlaster runs");
215       Client pub = doConnect("publisher/2", null); // with public session ID
216       doPublish(pub.con);
217       doPublish(pub.con);
218       pub.con.disconnect(null);
219 
220       log.info("STEP2: Start xmlBlaster server");
221       this.serverThread = EmbeddedXmlBlaster.startXmlBlaster(Util.getOtherServerPorts(serverPort));
222 
223       log.info("STEP3: Start subscriber and subscribe");
224       Client sub1 = doConnect("subscribe/1", null);
225       doSubscribe(sub1.con);
226       assertEquals("", 0, sub1.updateInterceptor.waitOnUpdate(1000L));
227       sub1.updateInterceptor.clear();
228 
229       log.info("STEP4: Start same publisher again");
230       pub = doConnect("publisher/2", null); // with public session ID
231       assertEquals("", 2, sub1.updateInterceptor.waitOnUpdate(1000L, 2));
232 
233       log.info("STEP5: Success, cleanup");
234       doErase(pub.con);
235       pub.con.disconnect(null);
236       sub1.con.disconnect(null);
237       EmbeddedXmlBlaster.stopXmlBlaster(this.serverThread);
238       this.serverThread = null;
239    }
240 
241    /**
242     * This is the callback method invoked from I_XmlBlasterAccess
243     * informing the client in an asynchronous mode if the connection was established.
244     * <p />
245     * This method is enforced through interface I_ConnectionStateListener
246     */
247    public void reachedAlive(ConnectionStateEnum oldState, I_XmlBlasterAccess connection) {
248       log.info("I_ConnectionStateListener-"+connection.getId()+": We were lucky, reconnected to xmlBlaster");
249    }
250 
251    public void reachedAliveSync(ConnectionStateEnum oldState, I_XmlBlasterAccess connection) {
252    }
253 
254    public void reachedPolling(ConnectionStateEnum oldState, I_XmlBlasterAccess connection) {
255       if (log!=null) log.warning("I_ConnectionStateListener-"+connection.getId()+": Lost connection to xmlBlaster");
256    }
257 
258    public void reachedDead(ConnectionStateEnum oldState, I_XmlBlasterAccess connection) {
259       if (log!=null) log.severe("DEBUG ONLY: Changed from connection state " + oldState + " to " + ConnectionStateEnum.DEAD);
260    }
261 
262    /**
263     * Method is used by TestRunner to load these tests
264     */
265    public static Test suite() {
266        TestSuite suite= new TestSuite();
267        suite.addTest(new TestTailback(null, "testTailback"));
268        return suite;
269    }
270 
271    /**
272     * Invoke: java org.xmlBlaster.test.client.TestTailback
273     * @deprecated Use the TestRunner from the testsuite to run it:<p />
274     * <pre>   java -Djava.compiler= junit.textui.TestRunner org.xmlBlaster.test.client.TestTailback</pre>
275     */
276    public static void main(String args[]) {
277       Global glob = new Global();
278       if (glob.init(args) != 0) {
279          System.err.println(ME + ": Init failed");
280          System.exit(1);
281       }
282       TestTailback testSub = new TestTailback(glob, "TestTailback");
283       testSub.setUp();
284       testSub.testTailback();
285       testSub.tearDown();
286    }
287 }


syntax highlighted by Code2HTML, v. 0.9.1