i_maptest.java

java开源的企业总线.xmlBlaster

package org.xmlBlaster.test.classtest.msgstore;

import java.util.logging.Logger;
import java.util.logging.Level;
import org.xmlBlaster.engine.ServerScope;
import org.xmlBlaster.util.XmlBlasterException;
import org.xmlBlaster.util.MsgUnit;
import org.xmlBlaster.util.queue.StorageId;
import org.xmlBlaster.engine.msgstore.I_MapEntry;
import org.xmlBlaster.engine.msgstore.I_Map;
import org.xmlBlaster.util.qos.storage.MsgUnitStoreProperty;
import org.xmlBlaster.util.qos.storage.QueuePropertyBase;
import org.xmlBlaster.engine.qos.PublishQosServer;
import org.xmlBlaster.engine.MsgUnitWrapper;

import java.util.ArrayList;

import junit.framework.*;
import org.xmlBlaster.engine.msgstore.StoragePluginManager;
import org.xmlBlaster.util.plugin.PluginInfo;

/**
 * Test I_Map e.g. MapPlugin which allows to store randomly messages. 
 * <p>
 * Invoke: java -Djava.compiler= junit.textui.TestRunner org.xmlBlaster.test.classtest.msgstore.I_MapTest
 * </p>
 * @see org.xmlBlaster.engine.msgstore.I_Map
 * @see org.xmlBlaster.engine.msgstore.ram.MapPlugin
 * @see org.xmlBlaster.util.queue.jdbc.JdbcQueuePlugin
 */
public class I_MapTest extends TestCase {
   private String ME = "I_MapTest";
   protected ServerScope glob;
   private static Logger log = Logger.getLogger(I_MapTest.class.getName());

   private final boolean IS_DURABLE = true;
   private final boolean IS_TRANSIENT = false;

   private I_Map currMap;
   private int currImpl;
/*
   static I_Map[] IMPL = {
                   new org.xmlBlaster.engine.msgstore.ram.MapPlugin(),
                   new org.xmlBlaster.util.queue.jdbc.JdbcQueuePlugin(),
                   new org.xmlBlaster.engine.msgstore.cache.PersistenceCachePlugin()
                 };
*/
   static String[] PLUGIN_TYPES = { new String("RAM"),
                                    new String("JDBC"),
                                    new String("CACHE") };

   public I_MapTest(String name, int currImpl) {
      super(name);
      this.currImpl = currImpl;

      String[] args = { //configure the cache
         "-persistence.persistentQueue", "JDBC,1.0",
         "-persistence.transientQueue", "RAM,1.0" };

      this.glob = new ServerScope(args);

      //this.ME = "I_MapTest[" + this.currMap.getClass().getName() + "]";
   }

   protected void setUp() {
      try {
         glob.getProperty().set("topic.queue.persistent.tableNamePrefix", "TEST");

         String type = PLUGIN_TYPES[this.currImpl];
         StoragePluginManager pluginManager = this.glob.getStoragePluginManager();
         PluginInfo pluginInfo = new PluginInfo(glob, pluginManager, "JDBC", "1.0");
         java.util.Properties prop = (java.util.Properties)pluginInfo.getParameters();
         prop.put("tableNamePrefix", "TEST");
         prop.put("entriesTableName", "_entries");
         this.glob.getProperty().set("QueuePlugin[JDBC][1.0]", pluginInfo.dumpPluginParameters());

         pluginInfo = new PluginInfo(glob, pluginManager, type, "1.0");

         MsgUnitStoreProperty storeProp = new MsgUnitStoreProperty(glob, "/node/test");
         StorageId queueId = new StorageId("msgUnitStore", "SomeMapId");

         this.currMap = pluginManager.getPlugin(pluginInfo, queueId, storeProp);
         this.currMap.shutdown(); // to allow to initialize again
      }
      catch (Exception ex) {
         log.severe("setUp: error when setting the property 'topic.queue.persistent.tableNamePrefix' to 'TEST': " + ex.getMessage());
      }

      // cleaning up the database from previous runs ...
/*
      QueuePropertyBase prop = null;
      try {
         prop = new MsgUnitStoreProperty(glob, "/node/test");

         StorageId queueId = new StorageId("msgUnitStore", "SetupMap");
         JdbcMapPlugin jdbcMap = new JdbcMapPlugin();
         jdbcMap.initialize(queueId, prop);
         jdbcMap.destroy();
      }
      catch (Exception ex) {
         log.severe("could not propertly set up the database: " + ex.getMessage());
      }
*/
   }

   private MsgUnit createMsgUnit(boolean persistent) {
      return createMsgUnit(persistent, -1);
   }

   private MsgUnit createMsgUnit(boolean persistent, long contentLen_) {
      try {
         int contentLen = (int)contentLen_;
         PublishQosServer publishQosServer = new PublishQosServer(glob, "<qos/>");
         publishQosServer.getData().setPersistent(persistent);
         String contentStr = "content";
         if (contentLen >= 0) {
            StringBuffer content = new StringBuffer(contentLen);
            for (int i=0; i<contentLen; i++) {
               content.append("X");
            }
            contentStr = content.toString();
         }
         return new MsgUnit(glob, "<key oid='Hi'/>", contentStr.getBytes(), publishQosServer.toXml());
      }
      catch (XmlBlasterException ex) {
         fail("msgUnit not constructed: " + ex.getMessage());
      }
      return null;
   }


   /**
    * Tests QueuePropertyBase() and getStorageId()
    * @param queueTypeList A space separated list of names for the
    *        implementations to be tested. Valid names are:
    *        RamMapPlugin JdbcMapPlugin
    */
   public void testConfig() {
      config(this.currMap);
   }

   /**
    * Tests initialize(), getProperties(), setProperties() and capacity()
    * @param queue !!!Is not initialized in this case!!!!
    */
   private void config(I_Map i_map) {
      ME = "I_MapTest.config(" + i_map.getStorageId() + ")[" + i_map.getClass().getName() + "]";
      System.out.println("***" + ME);

      QueuePropertyBase prop1 = null;
      QueuePropertyBase prop = null;
      try {
         // test initialize()
         prop1 = new MsgUnitStoreProperty(glob, "/node/test");
         int max = 12;
         prop1.setMaxEntries(max);
         prop1.setMaxEntriesCache(max);
         assertEquals(ME+": Wrong capacity", max, prop1.getMaxEntries());
         assertEquals(ME+": Wrong cache capacity", max, prop1.getMaxEntriesCache());
         StorageId queueId = new StorageId("msgUnitStore", "SomeMapId");

         i_map.initialize(queueId, prop1);
         assertEquals(ME+": Wrong queue ID", queueId, i_map.getStorageId());

         try {
            prop = new MsgUnitStoreProperty(glob, "/node/test");
            prop.setMaxEntries(99);
            prop.setMaxEntriesCache(99);
            i_map.setProperties(prop);
         }
         catch(XmlBlasterException e) {
            fail("Changing properties failed: " + e.getMessage());
         }

      }
      catch(XmlBlasterException e) {
         fail(ME + ": Exception thrown: " + e.getMessage());
      }

      long len = prop.getMaxEntries();
      assertEquals(ME+": Wrong capacity", prop.getMaxEntries(), i_map.getMaxNumOfEntries());
      assertEquals(ME+": Wrong capacity", prop.getMaxEntries(), ((QueuePropertyBase)i_map.getProperties()).getMaxEntries());
      assertEquals(ME+": Wrong size", 0, i_map.getNumOfEntries());

      try {
         for (int ii=0; ii<len; ii++) {
            i_map.put(new MsgUnitWrapper(glob, createMsgUnit(false), i_map.getStorageId()));
         }
         assertEquals(ME+": Wrong total size", len, i_map.getNumOfEntries());

         try {
            MsgUnitWrapper queueEntry = new MsgUnitWrapper(glob, createMsgUnit(false), i_map.getStorageId());
            i_map.put(queueEntry);
            i_map.put(queueEntry);
            fail("Did expect an exception on overflow getMaxNumOfEntries=" + i_map.getMaxNumOfEntries() + " size=" + i_map.getNumOfEntries());
         }
         catch(XmlBlasterException e) {
            log.info("SUCCESS the exception is OK: " + e.getMessage());
         }

         log.info("toXml() test:" + i_map.toXml(""));
         log.info("usage() test:" + i_map.usage());

         assertEquals(ME+": should not be shutdown", false, i_map.isShutdown());
         i_map.shutdown();
         assertEquals(ME+": should be shutdown", true, i_map.isShutdown());

         log.info("#2 Success, filled " + i_map.getNumOfEntries() + " messages into queue");
         System.out.println("***" + ME + " [SUCCESS]");
         i_map.shutdown();
      }
      catch(XmlBlasterException e) {
         fail(ME + ": Exception thrown: " + e.getMessage());
      }
   }

//------------------------------------
   public void testPutMsg() {
      String queueType = "unknown";
      try {
         QueuePropertyBase prop = new MsgUnitStoreProperty(glob, "/node/test");
         queueType = this.currMap.toString();
         StorageId queueId = new StorageId("msgUnitStore", "MapPlugin/putMsg");
         this.currMap.initialize(queueId, prop);
         this.currMap.clear();
         assertEquals(ME + "wrong size before starting ", 0L, this.currMap.getNumOfEntries());
         putMsg(this.currMap);
      }
      catch (XmlBlasterException ex) {
         fail("Exception when testing PutMsg probably due to failed initialization of the queue of type " + queueType + ": " + ex.getMessage());
      }
   }

   /**
    * Tests put(MsgMapEntry[]) and put(MsgMapEntry) and clear()
    */
   private void putMsg(I_Map i_map) {
      ME = "I_MapTest.putMsg(" + i_map.getStorageId() + ")[" + i_map.getClass().getName() + "]";
      System.out.println("***" + ME);
      try {
         //========== Test 1: put(I_MapEntry[])
         int numLoop = 10;
         ArrayList list = new ArrayList();
         for (int ii=0; ii<numLoop; ii++) {
            MsgUnitWrapper[] queueEntries = {
                         new MsgUnitWrapper(glob, createMsgUnit(false), i_map.getStorageId()),
                         new MsgUnitWrapper(glob, createMsgUnit(false), i_map.getStorageId()),
                         new MsgUnitWrapper(glob, createMsgUnit(false), i_map.getStorageId())};

            for(int i=0; i<queueEntries.length; i++)
               i_map.put(queueEntries[i]);

            for (int i=0; i < 3; i++) list.add(queueEntries[i]);

            this.checkSizeAndEntries(" put(I_MapEntry[]) ", list, i_map);
            assertEquals(ME+": Wrong size", (ii+1)*queueEntries.length, i_map.getNumOfEntries());
         }
         int total = numLoop*3;
         assertEquals(ME+": Wrong total size", total, i_map.getNumOfEntries());
         log.info("#1 Success, filled " + i_map.getNumOfEntries() + " messages into queue");


         //========== Test 2: put(I_MapEntry)
         for (int ii=0; ii<numLoop; ii++) {
            MsgUnitWrapper queueEntry = new MsgUnitWrapper(glob, createMsgUnit(false), i_map.getStorageId());
            list.add(queueEntry);