i_queuetest.java

java开源的企业总线.xmlBlaster

            assertEquals("number of entries incremented on last invocation", -8, this.queueSizeListener.getLastIncrementEntries());
            assertEquals("number of bytes incremented on last invocation", -8*queueEntries[0].getSizeInBytes(), this.queueSizeListener.getLastIncrementBytes());
                                        
            assertEquals(ME+": Wrong number removed", 8, numRemoved);
            assertEquals(ME+": Wrong size", queueEntries.length-8, queue.getNumOfEntries());

            numRemoved = queue.removeWithPriority(-1, -1L, 27, 99);
            long sizeInBytes = (queueEntries.length - 8) * queueEntries[0].getSizeInBytes();
            assertEquals(ME+": Wrong size in bytes ", sizeInBytes, queue.getNumOfBytes());
            assertEquals(ME+": Wrong number removed", 0, numRemoved);
            assertEquals(ME+": Wrong number of entries ", queueEntries.length-8, queue.getNumOfEntries());

            queue.clear();

            log.info("#1 Success, fill and remove");
         }

         //========== Test 2: remove prio 7 and 9 with num limit
         {
            DummyEntry[] queueEntries = {
                         new DummyEntry(glob, PriorityEnum.NORM_PRIORITY, queue.getStorageId(), true),
                         new DummyEntry(glob, PriorityEnum.HIGH_PRIORITY, queue.getStorageId(), true),
                         new DummyEntry(glob, PriorityEnum.MAX_PRIORITY, queue.getStorageId(), true),
                         new DummyEntry(glob, PriorityEnum.NORM_PRIORITY, queue.getStorageId(), true),
                         new DummyEntry(glob, PriorityEnum.HIGH_PRIORITY, queue.getStorageId(), true),
                         new DummyEntry(glob, PriorityEnum.MAX_PRIORITY, queue.getStorageId(), true),
                         new DummyEntry(glob, PriorityEnum.MAX_PRIORITY, queue.getStorageId(), true),
                         new DummyEntry(glob, PriorityEnum.HIGH_PRIORITY, queue.getStorageId(), true),
                         new DummyEntry(glob, PriorityEnum.NORM_PRIORITY, queue.getStorageId(), true),
                         new DummyEntry(glob, PriorityEnum.NORM_PRIORITY, queue.getStorageId(), true),
                         new DummyEntry(glob, PriorityEnum.HIGH_PRIORITY, queue.getStorageId(), true),
                         new DummyEntry(glob, PriorityEnum.MAX_PRIORITY, queue.getStorageId(), true),
                         new DummyEntry(glob, PriorityEnum.MIN_PRIORITY, queue.getStorageId(), true)
                                        };
            this.queue.removeStorageSizeListener(null);
            this.queue.addStorageSizeListener(this.queueSizeListener);
            this.queueSizeListener.clear();
                                        
            queue.put(queueEntries, false);

            long numRemoved = queue.removeWithPriority(2, -1L, 7, 9);

            assertEquals(ME+": Wrong number removed", 2, numRemoved);
            assertEquals(ME+": Wrong size", queueEntries.length-2, queue.getNumOfEntries());
            assertEquals("number of invocations", 2, this.queueSizeListener.getCount());
            assertEquals("number of entries incremented on last invocation", -2, this.queueSizeListener.getLastIncrementEntries());
            assertEquals("number of bytes incremented on last invocation", -2*queueEntries[0].getSizeInBytes(), this.queueSizeListener.getLastIncrementBytes());
                                        
            log.info("#2 Success, fill and remove");
         }
         queue.shutdown();
      }
      catch(XmlBlasterException e) {
         fail(ME + ": Exception thrown: " + e.getMessage());
      }
   }

//------------------------------------
   public void testRemoveRandom() {

      String queueType = "unknown";
      try {
         QueuePropertyBase prop = new CbQueueProperty(glob, Constants.RELATING_CALLBACK, "/node/test");
         queueType = this.queue.toString();
         StorageId queueId = new StorageId(Constants.RELATING_CALLBACK, "QueuePlugin/removeRandom");
         this.queue.initialize(queueId, prop);
         queue.clear();
         assertEquals(ME + "wrong size before starting ", 0, queue.getNumOfEntries());
         removeRandom(this.queue);
      }
      catch (XmlBlasterException ex) {
         log.severe("Exception when testing removeRandom probably due to failed initialization of the queue " + queueType);
      }

   }


   /**
    * Test removeRandom(long[])
    */
   private void removeRandom(I_Queue queue) {
      ME = "I_QueueTest.removeRandom(" + queue.getStorageId() + ")[" + queue.getClass().getName() + "][" + queue.getClass().getName() + "]";
      System.out.println("***" + ME);
      try {
         //========== Test 1: remove 1 from 1
         {
            this.queue.removeStorageSizeListener(null);
            this.queue.addStorageSizeListener(this.queueSizeListener);
            this.queueSizeListener.clear();

            //MsgUnit msgUnit = new MsgUnit("<key/>", "bla".getBytes(), "<qos/>");
            DummyEntry[] queueEntries = { new DummyEntry(glob, PriorityEnum.NORM_PRIORITY, queue.getStorageId(), true) };
            queue.put(queueEntries, false);

            I_QueueEntry[] testEntryArr = { queueEntries[0] };
            long numRemoved = 0L;
            boolean[] tmpArr = queue.removeRandom(testEntryArr);
            for (int i=0; i < tmpArr.length; i++) if(tmpArr[i]) numRemoved++;

            assertEquals("number of entries incremented on last invocation", -1, this.queueSizeListener.getLastIncrementEntries());
            assertEquals("number of bytes incremented on last invocation", -queueEntries[0].getSizeInBytes(), this.queueSizeListener.getLastIncrementBytes());

            assertEquals(ME+": Wrong number removed", 1, numRemoved);
            assertEquals(ME+": Wrong size", 0, queue.getNumOfEntries());
            log.info("#1 Success, fill and random remove");
         }

         //========== Test 2: removeRandom 2 from 3
         {
            DummyEntry[] queueEntries = {
                         new DummyEntry(glob, PriorityEnum.NORM_PRIORITY, queue.getStorageId(), true),
                         new DummyEntry(glob, PriorityEnum.NORM_PRIORITY, queue.getStorageId(), true),
                         new DummyEntry(glob, PriorityEnum.NORM_PRIORITY, queue.getStorageId(), true)
                                        };
            queue.put(queueEntries, false);

            I_QueueEntry[] testEntryArr = { queueEntries[0], 
                                            queueEntries[2]
                                          };
            long numRemoved = 0L;
            boolean[] tmpArr = queue.removeRandom(testEntryArr);
            for (int i=0; i < tmpArr.length; i++) if(tmpArr[i]) numRemoved++;

            assertEquals("number of entries incremented on last invocation", -2, this.queueSizeListener.getLastIncrementEntries());
            assertEquals("number of bytes incremented on last invocation", -2*queueEntries[0].getSizeInBytes(), this.queueSizeListener.getLastIncrementBytes());

            assertEquals(ME+": Wrong number removed", 2, numRemoved);
            assertEquals(ME+": Wrong size", 1, queue.getNumOfEntries());
            I_QueueEntry result = queue.peek();
            assertEquals(ME+": Wrong timestamp", queueEntries[1].getUniqueId(), result.getUniqueId());
            queue.clear();
            log.info("#2 Success, fill and random remove");
         }

         //========== Test 3: removeRandom 5 from 3
         {
            DummyEntry[] queueEntries = {
                         new DummyEntry(glob, PriorityEnum.NORM_PRIORITY, queue.getStorageId(), true),
                         new DummyEntry(glob, PriorityEnum.NORM_PRIORITY, queue.getStorageId(), true),
                         new DummyEntry(glob, PriorityEnum.NORM_PRIORITY, queue.getStorageId(), true)
                                        };
            queue.put(queueEntries, false);

            I_QueueEntry[] dataIdArr = {
                         new DummyEntry(glob, PriorityEnum.NORM_PRIORITY, queue.getStorageId(), true),
                         queueEntries[0],
                         new DummyEntry(glob, PriorityEnum.NORM_PRIORITY, queue.getStorageId(), true),
                         new DummyEntry(glob, PriorityEnum.NORM_PRIORITY, queue.getStorageId(), true),
                         queueEntries[2],
                                        };
            long numRemoved = 0L;
            boolean[] tmpArr = queue.removeRandom(dataIdArr);
            for (int i=0; i < tmpArr.length; i++) if(tmpArr[i]) numRemoved++;

            assertEquals("number of entries incremented on last invocation", -2, this.queueSizeListener.getLastIncrementEntries());
            assertEquals("number of bytes incremented on last invocation", -2*queueEntries[0].getSizeInBytes(), this.queueSizeListener.getLastIncrementBytes());

            assertEquals(ME+": Wrong number removed", 2, numRemoved);
            assertEquals(ME+": Wrong size", 1, queue.getNumOfEntries());

            I_QueueEntry entry = queue.peek();
            assertTrue("Missing entry", (I_QueueEntry)null != entry);
            assertEquals(ME+": Wrong entry removed", queueEntries[1].getUniqueId(), entry.getUniqueId());

            queue.clear();
            log.info("#3 Success, fill and random remove");
         }

         //========== Test 4: removeRandom 0 from 0
         {
            DummyEntry[] queueEntries = new DummyEntry[0];
            queue.put(queueEntries, false);

            I_QueueEntry[] dataIdArr = new I_QueueEntry[0];

            long numRemoved = 0L;
            boolean[] tmpArr = queue.removeRandom(dataIdArr);
            for (int i=0; i < tmpArr.length; i++) if(tmpArr[i]) numRemoved++;

            assertEquals(ME+": Wrong number removed", 0, numRemoved);
            assertEquals(ME+": Wrong size", 0, queue.getNumOfEntries());
            queue.clear();
            log.info("#4 Success, fill and random remove");
         }

         //========== Test 5: removeRandom null from null
         {
            queue.put((DummyEntry[])null, false);

//            long numRemoved = queue.removeRandom((I_QueueEntry[])null);
            long numRemoved = 0L;
            boolean[] tmpArr = queue.removeRandom((I_QueueEntry[])null);
            for (int i=0; i < tmpArr.length; i++) if(tmpArr[i]) numRemoved++;

            assertEquals(ME+": Wrong number removed", 0, numRemoved);
            assertEquals(ME+": Wrong size", 0, queue.getNumOfEntries());
            queue.clear();
            log.info("#5 Success, fill and random remove");
         }

         queue.shutdown();
         System.out.println("***" + ME + " [SUCCESS]");
      }
      catch(XmlBlasterException e) {
         fail(ME + ": Exception thrown: " + e.getMessage());
      }
   }



//------------------------------------
   public void testTakeLowest() {
      String queueType = "unknown";
      try {
         QueuePropertyBase prop = new CbQueueProperty(glob, Constants.RELATING_CALLBACK, "/node/test");
         queueType = this.queue.toString();
         StorageId queueId = new StorageId(Constants.RELATING_CALLBACK, "QueuePlugin/takeLowest");
         this.queue.initialize(queueId, prop);
         queue.clear();
         assertEquals(ME + "wrong size before starting ", 0, queue.getNumOfEntries());
         takeLowest(this.queue);
      }
      catch (XmlBlasterException ex) {
         log.severe("Exception when testing removeRandomLong probably due to failed initialization of the queue " + queueType);
      }

   }



   /**
    * returns the number of entries left in the queue after this processing operation 
    * @param queue the queue to use for this test
    * @param numEntries the number of entries to pass to the takeLowest operation
    * @param numBytes the num of bytes to pass to the takeLowest operation
    * @param leaveOne the flag to pass to the takeLowest operation
    * @param origEntries the array of the original entries put into the queue
    * @param entriesLeft number of entries left in the queue before this operation
    * @param currentEntries the number of entries which should have been processed by this operation
    */
   private final int assertCheckForTakeLowest(I_Queue queue, int numEntries, long numBytes, 
      I_QueueEntry refEntry, boolean leaveOne, I_QueueEntry[] origEntries, int entriesLeft, 
      int currentEntries, long size) throws XmlBlasterException {
      String me = ME + "/" + numEntries + "/" + numBytes + "/" + leaveOne + "/" + entriesLeft + "/" + currentEntries;
      if (log.isLoggable(Level.FINE)) log.fine("");
      assertEquals(me+": Wrong size of entry ", size, origEntries[0].getSizeInBytes());
      assertEquals(me+": Wrong amount of entries in queue before takeLowest invocation ", entriesLeft, queue.getNumOfEntries());
      assertEquals(me+": Wrong size of entries in queue before takeLowest invocation ", size*entriesLeft, queue.getNumOfBytes());
      assertEquals(me+": Wrong amount of persistent entries in queue before takeLowest invocation ", entriesLeft, queue.getNumOfPersistentEntries());
      assertEquals(me+": Wrong size of persistent entries in queue before takeLowest invocation ", size*entriesLeft, queue.getNumOfPersistentBytes());

      ArrayList list = null;
      try {
         list = queue.peekLowest(numEntries, numBytes, refEntry, leaveOne);  // gives back all minus one
         assertEquals(me+": Wrong number of entries in peekLowest return ", currentEntries, list.size());
         assertEquals(me+": Wrong number of entries in queue after peekLowest invocation ", entriesLeft, queue.getNumOfEntries());
         assertEquals(me+": Wrong number of bytes in queue after peekLowest invocation ", size*(entriesLeft), queue.getNumOfBytes());
         assertEquals(me+": Wrong number of persistent bytes in queue after takeLowest invocation ", size*(entriesLeft), queue.getNumOfPersistentBytes());
      }
      catch (XmlBlasterException e) {
         if (e.getErrorCode()!=ErrorCode.INTERNAL_NOTIMPLEMENTED) throw e;
      }

      list = queue.takeLowest(numEntries, numBytes, refEntry, leaveOne);  // gives back all minus one

      assertEquals("number of entries incremented on last invocation", -currentEntries, this.queueSizeListener.getLastIncrementEntries());
      assertEquals("number of bytes incremented on last invocation", -currentEntries*origEntries[0].getSizeInBytes(), this.queueSizeListener.getLastIncrementBytes());

      assertEquals(me+": Wrong number of entries in takeLowest return ", currentEntries, list.size());
      assertEquals(me+": Wrong number of entries in queue after takeLowest invocation ", entriesLeft-currentEntries, queue.getNumOfEntries());
      assertEquals(me+": Wrong number of bytes in queue after takeLowest invocation ", size*(entriesLeft-currentEntries), queue.getNumOfBytes());
      assertEquals(me+": Wrong number of persistent bytes in queue after takeLowest invocation ", size*(entriesLeft-currentEntries), queue.getNumOfPersistentBytes());

      for (int i=entriesLeft-currentEntries; i < entriesLeft; i++) {
         int j = entriesLeft - 1 - i;
         long ref = ((I_QueueEntry)list.get(j)).getUniqueId();