cachequeuetest.java

java开源的企业总线.xmlBlaster

               refQueue.initialize(queueId, prop);

               for (int it=0; it < numOfTransientEntries.length; it++) {
                  // entry.setPrio(4+(it%3));
                  for (int id=0; id < numOfPersistentEntries.length; id++) {

                     log.info("**** SUB-TEST maxNumOfBytesCache["+ic+"]=" + maxNumOfBytesCache[ic] + " maxNumOfBytes["+is+"]=" + maxNumOfBytes[is] +
                                   " -> numOfTransientEntries["+it+"]=" + numOfTransientEntries[it] + " numOfPersistentEntries["+id+"]=" + numOfPersistentEntries[id]);
                     if (!refQueue.isShutdown()) refQueue.shutdown();
                     refQueue.initialize(queueId, prop);
                     refQueue.clear();

                     assertEquals(ME + " the number of bytes of the queue should be zero ", 0L, refQueue.getNumOfBytes());
                     assertEquals(ME + " the number of entries in the queue should be zero ", 0L, refQueue.getNumOfEntries());
                     assertEquals(ME + " the number of bytes of the persistent entries in the queue should be zero ", 0L, refQueue.getNumOfPersistentBytes());
                     assertEquals(ME + " the number of persistent entries in the queue should be zero ", 0L, refQueue.getNumOfPersistentEntries());

                     assertEquals(ME + " the maximum number of entries is wrong ", maxNumOfBytes[is], refQueue.getMaxNumOfBytes());

                     try {

                        refQueue.clear();
                        // prepare the inputs .
                        Hashtable[] inputTable = new Hashtable[3];
                        for (int i=0; i < 3; i++) inputTable[i] = new Hashtable();

                        DummyEntry[] transients = new DummyEntry[numOfTransientEntries[it]];
                        DummyEntry[] persistentEntries    = new DummyEntry[numOfPersistentEntries[id]];

                        log.info("putPeekRemove " + queueId + " persistent: " + persistentEntries.length + " transient: " + transients.length);

                        boolean persistent = false;
                        for (int i=0; i < transients.length; i++) {
                           int prio = i % 3;
                           PriorityEnum enumer = PriorityEnum.toPriorityEnum(prio+4);
                           DummyEntry entry = new DummyEntry(glob, enumer, refQueue.getStorageId(), entrySize, persistent);
                           transients[i] = entry;
                           inputTable[prio].put(new Long(entry.getUniqueId()), entry);
                        }
                        persistent = true;
                        for (int i=0; i < persistentEntries.length; i++) {
                           int prio = i % 3;
                           PriorityEnum enumer = PriorityEnum.toPriorityEnum(prio+4);
                           DummyEntry entry = new DummyEntry(glob, enumer, refQueue.getStorageId(), entrySize, persistent);
                           persistentEntries[i] = entry;
                           inputTable[prio].put(new Long(entry.getUniqueId()), entry);
                        }

                        // do the test here ....
                        assertEquals(ME + " number of persistent entries is wrong ", 0L, refQueue.getNumOfPersistentEntries());
                        assertEquals(ME + " number of entries is wrong ", 0L, refQueue.getNumOfEntries());
                        for (int i=0; i < transients.length; i++) {
                           lastSuccessfulLocation = "transientEntries put #" + i;
                           refQueue.put(transients[i], false);
                        }
                        assertEquals(ME + " number of entries after putting transients is wrong ", transients.length, refQueue.getNumOfEntries());
                        for (int i=0; i < persistentEntries.length; i++) {
                           lastSuccessfulLocation = "persistentEntries put #" + i;
                           refQueue.put(persistentEntries[i], false);
                        }
                        assertEquals(ME + " number of entries after putting transients is wrong ", persistentEntries.length + transients.length, refQueue.getNumOfEntries());
                        long nPersistents  = refQueue.getNumOfPersistentEntries();
                        long nTransient = refQueue.getNumOfEntries() - nPersistents;

                        assertEquals(ME + " number of persistent entries is wrong ", persistentEntries.length, nPersistents);
                        assertEquals(ME + " number of transient entries is wrong ", transients.length, nTransient);

                        ArrayList total = new ArrayList();
                        ArrayList ret = refQueue.peekSamePriority(-1, -1L);
                        refQueue.removeRandom((I_QueueEntry[])ret.toArray(new I_QueueEntry[ret.size()]));
                        while (ret.size() > 0) {
                           total.addAll(ret);
                           ret = refQueue.peekSamePriority(-1, -1L);
                           if (ret.size() > 0)
                              refQueue.removeRandom((I_QueueEntry[])ret.toArray(new I_QueueEntry[ret.size()]));
                        }
                        int mustEntries = inputTable[0].size() + inputTable[1].size() + inputTable[2].size();


                        long totNumOfBytes = entrySize * (numOfPersistentEntries[id]+numOfTransientEntries[it]);
                        log.fine("total number of bytes: " + totNumOfBytes + " maxNumOfBytes: " + maxNumOfBytes[is]);
                        log.fine("entries must be: " + mustEntries);

                        assertTrue("Overflow is not allowed " + refQueue.toXml("") + "total number of bytes " + totNumOfBytes + " max number of bytes: " + maxNumOfBytes[is], totNumOfBytes <= maxNumOfBytes[is]);
//                        assertTrue(ME + " Overflow is not allowed " + refQueue.toXml("") , checkIfPossible(transientNumOfBytes, persistentNumOfBytes, maxTransientNumOfBytes, maxPersistentNumOfBytes));
                        assertEquals(ME + " number of returned values differe from input values " + refQueue.toXml(""), mustEntries, total.size());
                        log.info("SUCCESS: cacheSize=" + maxNumOfBytesCache[ic] + " maxBytes=" + maxNumOfBytes[is] + " .... looks OK");

                        int count = 0;
                        for (int j=0; j < 3; j++) {
                           Hashtable table = inputTable[j];
                           Enumeration keys = table.keys();
                           while (keys.hasMoreElements()) {
                              ((I_QueueEntry)table.get(keys.nextElement())).getUniqueId();
                              ((I_QueueEntry)total.get(count)).getUniqueId();
                              assertEquals("uniqueId differe for count " + count + " " + refQueue.toXml(""), mustEntries, total.size());
                              count++;
                           }
                        }
                     }
                     catch(XmlBlasterException e) {
                        log.finest("Exception (might be ok): " + e.toString());
                        assertTrue("Overflow is not allowed on location '"+ lastSuccessfulLocation + "' " + refQueue.toXml("") + "total number of bytes " + entrySize*(numOfPersistentEntries[id]+numOfTransientEntries[it]) + " max muber of bytes: " + maxNumOfBytes[is], entrySize*(numOfPersistentEntries[id]+numOfTransientEntries[it]) > maxNumOfBytes[is]);
                        log.info("SUCCESS: Exception is OK: " + e.toString());
                     }
                  }
               }
            }
         }
   }


   public void testAvailability() {
      String queueType = "CACHE";
      try {
         availability();
      }
      catch (XmlBlasterException ex) {
         fail("Exception when testing availability probably due to failed initialization of the queue of type " + queueType);
         ex.printStackTrace();
      }
   }


   /**
    * when queue available:
    * -fill queue with 3 persistent and 2 transient messages -> RAM:5 JDBC:3
    * - queue is made unavailable
    * - queue is filled with 2 persistent and 3 transient msg -> RAM:10 JDBC:3 (since no comm)
    * - peek and then remove all available entries: -> RAM:0 JDBC:3 (since no comm)
    */
   public void availability() throws XmlBlasterException {
      // set up the queues ....
      long maxNumOfBytesCache = 10000L;
      long maxNumOfBytes = 50000L;
      long entrySize = 100L;

      QueuePropertyBase prop = new CbQueueProperty(glob, Constants.RELATING_CALLBACK, "/node/test");
      prop.setMaxEntries(2000L);
      prop.setMaxEntriesCache(1000L);
      prop.setMaxBytes(maxNumOfBytes);
      prop.setMaxBytesCache(maxNumOfBytesCache);
      StorageId queueId = new StorageId(Constants.RELATING_CALLBACK, "CacheQueueTest/jdbc" + maxNumOfBytes + "/ram" + maxNumOfBytesCache);
      this.queue.clear();
      this.queue.shutdown();
      this.queue.initialize(queueId, prop);

      if (!this.queue.isShutdown()) this.queue.shutdown();
      this.queue.initialize(queueId, prop);
      this.queue.clear();

      int numOfEntries = 20;
      int entries1 = 5;
      int entries2 = 10;

      this.queue.clear();
      DummyEntry[] entries = new DummyEntry[numOfEntries];
      PriorityEnum prio = PriorityEnum.toPriorityEnum(4);

      boolean persistent = false;
      for (int i=0; i < numOfEntries; i++) {
         persistent = (i % 2) == 0; // even are persistent uneven are transient
         entries[i] = new DummyEntry(glob, prio, this.queue.getStorageId(), entrySize, persistent);
      }

      // do the test here ....
      for (int i=0; i < entries1; i++) {
         this.queue.put(entries[i], false);
//         assertEquals(ME + " number of entries after putting transients is wrong ", transients.length, queue.getNumOfEntries());
      }

      CacheQueueInterceptorPlugin cacheQueue = (CacheQueueInterceptorPlugin)this.queue;
      cacheQueue.storageUnavailable(I_StorageProblemListener.AVAILABLE);

      for (int i=entries1; i < entries2; i++) {
         this.queue.put(entries[i], false);
      }

      ArrayList list = this.queue.peek(-1, -1L);
      assertEquals(ME + " number of entries when retrieving is wrong ", entries2, list.size());
      for (int i=0; i < list.size(); i++) {
         long uniqueId = ((I_QueueEntry)list.get(i)).getUniqueId();
         assertEquals(ME + " entry sequence is wrong ", entries[i].getUniqueId(), uniqueId);
      }
      long ret = 0L;
      boolean[] tmpArr = this.queue.removeRandom( (I_QueueEntry[])list.toArray(new I_QueueEntry[list.size()]) );
      for (int i=0; i < tmpArr.length; i++) if (tmpArr[i]) ret++;
      assertEquals(ME + " number of entries removed is wrong ", (long)entries2, ret);

      list = this.queue.peek(-1, -1L);
      assertEquals(ME + " number of entries peeked after removal is wrong ", 0, list.size());

      long num = this.queue.getNumOfEntries();
      assertEquals(ME + " number of entries after removal is wrong ", 0L, num);

      cacheQueue.storageAvailable(I_StorageProblemListener.UNAVAILABLE);
      list = this.queue.peek(-1, -1L);
      assertEquals(ME + " number of entries peeked after reconnecting is wrong ", 0, list.size());

      num = this.queue.getNumOfEntries();
      assertEquals(ME + " number of entries after reconnecting is wrong ", 0L, num);

/*
      for (int i=entries2; i < numOfEntries; i++) {
         this.queue.put(entries[i], false);
      }
*/

   }


   /**
    * <pre>
    *  java org.xmlBlaster.test.classtest.queue.CacheQueueTest
    * </pre>
    */
   public static void main(String args[]) {

      Global glob = new Global(args);
      CacheQueueTest testSub = new CacheQueueTest(glob, "CacheQueueTest");

      long startTime = System.currentTimeMillis();

      testSub.setUp();
      testSub.testAvailability();
      testSub.tearDown();

/*
      testSub.setUp();
      testSub.tearDown();

      testSub.setUp();
      testSub.testConfig();
      testSub.tearDown();
*/
      testSub.setUp();
      testSub.testPutPeekRemove();
      testSub.tearDown();

      testSub.setUp();
      testSub.testClearWithSwappedEntries();
      testSub.tearDown();

      long usedTime = System.currentTimeMillis() - startTime;
      log.info("time used for tests: " + usedTime/1000 + " seconds");
   }
}