massivesubtest.java

java开源的企业总线.xmlBlaster

/*------------------------------------------------------------------------------
Name:      MassiveSubTest.java
Project:   xmlBlaster.org
Copyright: xmlBlaster.org, see xmlBlaster-LICENSE file
Comment:   Load test for xmlBlaster
------------------------------------------------------------------------------*/
package org.xmlBlaster.test.stress;

import java.util.logging.Logger;
import java.util.logging.Level;

import org.xmlBlaster.util.Global;
import org.xmlBlaster.util.ThreadLister;
import org.xmlBlaster.util.def.Constants;
import org.xmlBlaster.util.StopWatch;
import org.xmlBlaster.client.qos.ConnectQos;
import org.xmlBlaster.client.qos.ConnectReturnQos;
import org.xmlBlaster.util.XmlBlasterException;
import org.xmlBlaster.client.I_XmlBlasterAccess;
import org.xmlBlaster.client.I_Callback;
import org.xmlBlaster.client.key.UpdateKey;
import org.xmlBlaster.client.qos.UpdateQos;
import org.xmlBlaster.client.qos.EraseReturnQos;
import org.xmlBlaster.client.key.SubscribeKey;
import org.xmlBlaster.client.qos.SubscribeQos;
import org.xmlBlaster.util.MsgUnit;
import org.xmlBlaster.util.qos.storage.CbQueueProperty;
import org.xmlBlaster.util.EmbeddedXmlBlaster;
import org.xmlBlaster.j2ee.util.GlobalUtil;

import org.xmlBlaster.test.Util;
import org.xmlBlaster.test.MsgInterceptor;
import junit.framework.*;
/**
 * Test differents scenarios for a massive ammount of subscibers.
 *
 * <p>Test 5000 subscribers (or numSubscribers) on one connection.</p>
 * <p>Test 5000 subscribers (or numSubscribers) with maxSubPerCon per connection</p>
 * <p>Test 5000 subscribers (or numSubscribers) on one connection each.</p>
 * <p>Do it for IOP, RMI</p>
 *
 * <p>If withEmbedded is set to false will run without an embedded server.</p>
 * <pre>
 *  java -Xms18M -Xmx256M org.xmlBlaster.test.stress.MassiveSubTest
 *  java -Xms18M -Xmx256M junit.swingui.TestRunner -noloading org.xmlBlaster.test.stress.MassiveSubTest
 * </pre>
 * @author Peter Antman
 */

public class MassiveSubTest extends TestCase implements I_Callback {
   private int numSubscribers = 5000;
   private int maxSubPerCon = 0;
   private boolean useOneConnection = false;
   private boolean withEmbedded = true;
   private int noToPub = 1;
   private int numToRec = numSubscribers * noToPub;
   private String ME = "MassiveSubTest";
   private Global glob;
   private static Logger log = Logger.getLogger(MassiveSubTest.class.getName());
   private int serverPort = 7615;
   private EmbeddedXmlBlaster serverThread;
   private boolean messageArrived = false;
   private MsgInterceptor updateInterceptor;

   private final String publishOid1 = "dummy1";
   private I_XmlBlasterAccess oneConnection;
   private String oneName;

   private int numReceived = 0;         // error checking
   private final String contentMime = "text/xml";
   private final String contentMimeExtended = "1.0";
   private GlobalUtil globalUtil;

   class Client {
      String loginName;
      I_XmlBlasterAccess connection;
      String subscribeOid;
      boolean oneConnection;
   }
   private Client[] manyClients;
   private I_XmlBlasterAccess[] manyConnections;
   private StopWatch stopWatch = new StopWatch();


   public MassiveSubTest(String testName) {
      super(testName);
      Global glob_ = Global.instance();
      setProtoMax(glob_, "IOR", "500");
      init(glob_, testName, "testManyClients", true);
   }

   public MassiveSubTest(Global glob, String testName, String loginName, boolean useOneConnection) {
      super(testName);
      init(glob, testName, loginName, useOneConnection);
   }

   public void init(Global glob, String testName, String loginName, boolean useOneConnection) {
      this.glob = glob;

      this.oneName = loginName;
      numSubscribers = glob.getProperty().get("numSubscribers", numSubscribers);
      maxSubPerCon = glob.getProperty().get("maxSubPerCon", maxSubPerCon);
      withEmbedded = glob.getProperty().get("withEmbedded", withEmbedded);
      noToPub = glob.getProperty().get("noToPub", noToPub);
      this.useOneConnection = useOneConnection;
      String clientProtocol = glob.getProperty().get("client.protocol", "IOR");
      try {
         glob.getProperty().set("client.protocol",clientProtocol);



      }catch(XmlBlasterException ex) {
         assertTrue("Could not setup test: " + ex, false);
      }
      ME = ME+":"+clientProtocol+(useOneConnection ? ":oneCon":":manyCon")+":"+numSubscribers + (maxSubPerCon>0?"/"+maxSubPerCon:"");

      numToRec = numSubscribers * noToPub;

   }
   
   /**
    * Sets up the fixture.
    * <p />
    * Connect to xmlBlaster and login
    */
   protected void setUp()
   {
      String[] args = {
         "-ClientProtocolPlugin[LOCAL][1.0]",
         "org.xmlBlaster.client.protocol.local.LocalConnection",
         "-ClientCbServerProtocolPlugin[LOCAL][1.0]",
         "org.xmlBlaster.client.protocol.local.LocalCallbackImpl",
         "-CbProtocolPlugin[LOCAL][1.0]",
         "org.xmlBlaster.protocol.local.CallbackLocalDriver"
      };
      glob.init(args);

      log.info("Setting up test ...");
      if (withEmbedded) {
         glob.init(Util.getOtherServerPorts(serverPort));
         serverThread = EmbeddedXmlBlaster.startXmlBlaster(glob);
         log.info("XmlBlaster is ready for testing a lots of subscribers");
         globalUtil = new GlobalUtil( serverThread.getMain().getGlobal() );
      } else {
         globalUtil = new GlobalUtil( );
      } // end of else
      glob = globalUtil.getClone(glob);


      numReceived = 0;
      try {
         oneConnection = glob.getXmlBlasterAccess(); // Find orb
         ConnectQos connectQos = new ConnectQos(glob, oneName, "secret"); // "<qos></qos>"; During login this is manipulated (callback address added)
         // If we have many subs on one con, we must raise the max size of the callback queue!
         CbQueueProperty cbProp =connectQos.getSessionCbQueueProperty();
         cbProp.setMaxEntries(numSubscribers+1000);
         cbProp.setMaxEntriesCache(numSubscribers+1000);
         this.updateInterceptor = new MsgInterceptor(this.glob, log, this); // Collect received msgs
         ConnectReturnQos connectReturnQos = oneConnection.connect(connectQos, this.updateInterceptor);
         log.info("Connected: " + connectReturnQos.toXml());
      }
      catch (Exception e) {
          log.severe("Login failed: " + e.toString());
          e.printStackTrace();
          assertTrue("Login failed: " + e.toString(), false);
      }
   }
   
   /**
    * Tears down the fixture.
    * <p />
    * cleaning up .... erase() the previous message OID and logout
    */
   protected void tearDown()
   {
      log.info("Tearing down");
      if (numReceived != numToRec) {
         log.severe("numToRec=" + numToRec + " but numReceived=" + numReceived);
         assertEquals("numToRec=" + numToRec + " but numReceived=" + numReceived, numSubscribers, numReceived);
      }
      

      if (manyClients != null) {
         for (int ii=0; ii<numSubscribers; ii++) {
            Client sub = manyClients[ii];
            if (sub.oneConnection) {
               try {
                  if ( sub.connection != null) {
                     sub.connection.unSubscribe( "<key oid='"+sub.subscribeOid+"'/>",
                                                 "<qos/>");
                  } else {
                     oneConnection.unSubscribe( "<key oid='"+sub.subscribeOid+"'/>",
                                                "<qos/>");
                  } // end of else
                  
               }catch(XmlBlasterException ex) {
                  log.severe("Could not unsubscribe: " +sub.subscribeOid+": " + ex);
               }
            }else {
               sub.connection.disconnect(null);
            }
         }
      }
      if ( manyConnections != null) {
         for ( int ii = 0;ii<manyConnections.length;ii++) {
            manyConnections[ii].disconnect(null);
         } // end of for ()
         
      } // end of if ()
      


      
      {
         String xmlKey = "<?xml version='1.0' encoding='ISO-8859-1' ?>\n" +
            "<key oid='" + publishOid1 + "' queryType='EXACT'>\n" +
            "</key>";
         String qos = "<qos></qos>";
         try {
            EraseReturnQos[] arr = oneConnection.erase(xmlKey, qos);
            assertEquals("Erase", 1, arr.length);
         } catch(XmlBlasterException e) { fail("Erase-XmlBlasterException: " + e.getMessage()); }
      }
      
      oneConnection.disconnect(null);
      oneConnection = null;
      log.info("Logout done");
      if (withEmbedded) {
         try { Thread.sleep(100L); } catch( InterruptedException i) {}
         EmbeddedXmlBlaster.stopXmlBlaster(this.serverThread);
         this.serverThread = null;
         
         // reset to default server port (necessary if other tests follow in the same JVM).
         Util.resetPorts();
      }

      this.glob = null;
     
      this.updateInterceptor = null;
      this.oneConnection = null;
      this.manyClients = null;
      this.manyConnections = null;
      this.stopWatch = null;
   }

   /**
    * helper
    */
   public void subcribeMany()
   {
      int ci=-1;
      try {

         if (log.isLoggable(Level.FINE)) log.fine("Subscribing ...");
         
         String passwd = "secret";
         
         SubscribeKey subKeyW = new SubscribeKey(glob, publishOid1);
         String subKey = subKeyW.toXml(); // "<key oid='" + publishOid1 + "' queryType='EXACT'></key>";
         
         SubscribeQos subQosW = new SubscribeQos(glob); // "<qos></qos>";
         String subQos = subQosW.toXml();
         
         manyClients = new Client[numSubscribers];
         if (maxSubPerCon >0 ) {
            // Check if reasonably
            if (  numSubscribers %  maxSubPerCon!= 0) {