testpersistentsession.java

java开源的企业总线.xmlBlaster

/*------------------------------------------------------------------------------
Name:      TestPersistentSession.java
Project:   xmlBlaster.org
Copyright: xmlBlaster.org, see xmlBlaster-LICENSE file
------------------------------------------------------------------------------*/
package org.xmlBlaster.test.client;

import java.util.logging.Logger;
import org.xmlBlaster.util.Global;
import org.xmlBlaster.util.SessionName;
import org.xmlBlaster.util.XmlBlasterException;
import org.xmlBlaster.util.def.ErrorCode;
import org.xmlBlaster.util.def.Constants;
import org.xmlBlaster.util.property.PropString;
import org.xmlBlaster.util.EmbeddedXmlBlaster;
import org.xmlBlaster.util.qos.address.Address;
import org.xmlBlaster.util.qos.address.CallbackAddress;
import org.xmlBlaster.util.MsgUnit;
import org.xmlBlaster.util.dispatch.ConnectionStateEnum;
import org.xmlBlaster.client.qos.PublishQos;
import org.xmlBlaster.client.I_Callback;
import org.xmlBlaster.client.I_ConnectionStateListener;
import org.xmlBlaster.client.I_XmlBlasterAccess;
import org.xmlBlaster.client.key.SubscribeKey;
import org.xmlBlaster.client.key.UpdateKey;
import org.xmlBlaster.client.qos.*;

import org.xmlBlaster.test.Util;
import org.xmlBlaster.test.MsgInterceptor;

import junit.framework.*;


/**
 * Tests the persistent sessions .
 * <br />For a description of what this persistent sessions and subscriptions are
 * please read the requirement engine.persistence.session.
 * <p>
 * This is an interesting example, since it creates a XmlBlaster server instance
 * in the same JVM , but in a separate thread, talking over CORBA with it.
 * <p>
 * Invoke examples:<br />
 * <pre>
 *   java junit.textui.TestRunner -noloading org.xmlBlaster.test.client.TestPersistentSession
 *   java junit.swingui.TestRunner -noloading org.xmlBlaster.test.client.TestPersistentSession
 * </pre>
 * @see org.xmlBlaster.client.I_XmlBlasterAccess
 */
public class TestPersistentSession extends TestCase implements I_ConnectionStateListener, I_Callback
{
   private static String ME = "TestPersistentSession";
   private static final boolean TRANSIENT = false;
   private static final boolean PERSISTENT = true;
   
   private Global glob;
   private Global origGlobal;
   private Global serverGlobal;
   private static Logger log = Logger.getLogger(TestPersistentSession.class.getName());

   private int serverPort = 7604;
   private EmbeddedXmlBlaster serverThread;

   private MsgInterceptor[] updateInterceptors;
   //private I_XmlBlasterAccess con;
   private String senderName;

   private int numPublish = 8;
   private int numStop = 3;
   private int numStart = 5;
   private final String contentMime = "text/plain";

   private final long reconnectDelay = 2000L;
   private boolean failsafeCallback = true;
   /** the session is persistent from the beginning */
   private boolean persistent = true;
   private boolean exactSubscription = false;
   private boolean initialUpdates = true;
   private int numSubscribers = 4;

   public TestPersistentSession(String testName) {
      this(null, testName);
   }

   public TestPersistentSession(Global glob, String testName) {
      super(testName);
      this.origGlobal = glob;
      this.senderName = testName;
      this.updateInterceptors = new MsgInterceptor[this.numSubscribers];
   }

   /**
    * Sets up the fixture.
    * <p />
    * Connect to xmlBlaster and login
    */
   protected void setUp() {
      setup(false);
   }
   
   
   private void setup(boolean restrictedEntries) {
      this.origGlobal = (this.origGlobal == null) ? Global.instance() : this.origGlobal;

      
      this.origGlobal.init(Util.getOtherServerPorts(serverPort));
      this.glob = this.origGlobal.getClone(null);

      String[] args = null;
      if (restrictedEntries) {
         args = new String[] {"-persistence/session/maxEntriesCache", "1",
                       "-persistence/session/maxEntries","2",
                       "-persistence/subscribe/maxEntriesCache", "2",
                       "-persistence/subscribe/maxEntries","3",
                      };
      }
      this.serverGlobal = this.origGlobal.getClone(args);
      serverThread = EmbeddedXmlBlaster.startXmlBlaster(this.serverGlobal);
      log.info("XmlBlaster is ready for testing on bootstrapPort " + serverPort);

      try { // we just connect and disconnect to make sure all resources are really cleaned up
         Global tmpGlobal = this.origGlobal.getClone(null);
         I_XmlBlasterAccess con = tmpGlobal.getXmlBlasterAccess(); // Find orb

         String passwd = "secret";
         ConnectQos connectQos = new ConnectQos(tmpGlobal, senderName, passwd); // == "<qos>...</qos>";
         connectQos.setSessionName(new SessionName(tmpGlobal, "general/1"));
         // set the persistent connection 
         connectQos.setPersistent(this.persistent);
         // Setup fail save handling for connection ...
         Address addressProp = new Address(tmpGlobal);
         addressProp.setDelay(reconnectDelay); // retry connecting every 2 sec
         addressProp.setRetries(-1);       // -1 == forever
         addressProp.setPingInterval(-1L); // switched off
         con.registerConnectionListener(this);
         connectQos.setAddress(addressProp);
         
         // setup failsafe handling for callback ...
         if (this.failsafeCallback) {
            CallbackAddress cbAddress = new CallbackAddress(tmpGlobal);
            cbAddress.setRetries(-1);
            cbAddress.setPingInterval(-1);
            cbAddress.setDelay(1000L);
            cbAddress.setSecretCbSessionId("someSecredSessionId");
            connectQos.addCallbackAddress(cbAddress);
         }
         con.connect(connectQos, this);
         DisconnectQos disconnectQos = new DisconnectQos(tmpGlobal);
         con.disconnect(disconnectQos);
      }
      catch (XmlBlasterException e) {
          log.warning("setUp() - login failed: " + e.getMessage());
          fail("setUp() - login fail: " + e.getMessage());
      }
      catch (Exception e) {
          log.severe("setUp() - login failed: " + e.toString());
          e.printStackTrace();
          fail("setUp() - login fail: " + e.toString());
      }
      
      try {
         I_XmlBlasterAccess con = this.glob.getXmlBlasterAccess(); // Find orb

         String passwd = "secret";
         ConnectQos connectQos = new ConnectQos(this.glob, senderName, passwd); // == "<qos>...</qos>";
         connectQos.setSessionName(new SessionName(this.glob, "general/1"));
         // set the persistent connection 
         connectQos.setPersistent(this.persistent);
         // Setup fail save handling for connection ...
         Address addressProp = new Address(glob);
         addressProp.setDelay(reconnectDelay); // retry connecting every 2 sec
         addressProp.setRetries(-1);       // -1 == forever
         addressProp.setPingInterval(-1L); // switched off
         con.registerConnectionListener(this);
         connectQos.setAddress(addressProp);
         
         // setup failsafe handling for callback ...
         if (this.failsafeCallback) {
            CallbackAddress cbAddress = new CallbackAddress(this.glob);
            cbAddress.setRetries(-1);
            cbAddress.setPingInterval(-1);
            cbAddress.setDelay(1000L);
            cbAddress.setSecretCbSessionId("someSecredSessionId");
            connectQos.addCallbackAddress(cbAddress);
         }

         con.connect(connectQos, this);  // Login to xmlBlaster, register for updates
      }
      catch (XmlBlasterException e) {
          log.warning("setUp() - login failed: " + e.getMessage());
          fail("setUp() - login fail: " + e.getMessage());
      }
      catch (Exception e) {
          log.severe("setUp() - login failed: " + e.toString());
          e.printStackTrace();
          fail("setUp() - login fail: " + e.toString());
      }
   }

   /**
    * Tears down the fixture.
    * <p />
    * cleaning up .... erase() the previous message OID and logout
    */
   protected void tearDown() {
      log.info("Entering tearDown(), test is finished");
      String xmlKey = "<key oid='' queryType='XPATH'>\n" +
                      "   //TestPersistentSession-AGENT" +
                      "</key>";

      String qos = "<qos><forceDestroy>true</forceDestroy></qos>";
      I_XmlBlasterAccess con = this.glob.getXmlBlasterAccess();
      try {
         con.erase(xmlKey, qos);

         PropString defaultPlugin = new PropString("CACHE,1.0");
         String propName = defaultPlugin.setFromEnv(this.glob, glob.getStrippedId(), null, "persistence", Constants.RELATING_TOPICSTORE, "defaultPlugin");
         log.info("Lookup of propName=" + propName + " defaultValue=" + defaultPlugin.getValue());
      }
      catch(XmlBlasterException e) {
         log.severe("XmlBlasterException: " + e.getMessage());
      }
      finally {
         con.disconnect(null);
         EmbeddedXmlBlaster.stopXmlBlaster(this.serverThread);
         this.serverThread = null;
         // reset to default server bootstrapPort (necessary if other tests follow in the same JVM).
         Util.resetPorts(this.serverGlobal);
         Util.resetPorts(this.glob);
         Util.resetPorts(this.origGlobal);
         this.glob = null;
         this.serverGlobal = null;
         con = null;
         Global.instance().shutdown();
      }
   }

   /**
    * TEST: Subscribe to messages with XPATH.
    */
   private void doSubscribe(int num, boolean isExact, boolean isPersistent) {
      try {
         SubscribeKey key = null;
         if (isExact)  key = new SubscribeKey(this.glob, "Message-1");
         else key = new SubscribeKey(this.glob, "//TestPersistentSession-AGENT", "XPATH");

         SubscribeQos qos = new SubscribeQos(this.glob); // "<qos><persistent>true</persistent></qos>";
         qos.setPersistent(isPersistent);
         qos.setWantInitialUpdate(this.initialUpdates);
         qos.setWantNotify(false); // to avoig getting erased messages