i_queuetest.java

java开源的企业总线.xmlBlaster

         queue.remove(2, -1L);
         numOfEntries = 2;
         ret = wrapper.blockingPeek(numOfEntries, 1000L);
         assertEquals("Wrong number of entries found", 1, ret.size());
         queue.clear();
         ret = wrapper.blockingPeek(numOfEntries, 1000L);
         assertEquals("Wrong number of entries found", 0, ret.size());
         
         // and now making asynchronous putting with events
         numOfEntries = 3;
         long delay = 500L;
         boolean inhibitEvents = false;
         QueuePutter putter = new QueuePutter(this.queue, delay, numOfEntries, inhibitEvents);
         putter.start();
         long t0 = System.currentTimeMillis();
         ret = wrapper.blockingPeek(numOfEntries, 10000L);
         assertEquals("Wrong number of entries when blocking with events", numOfEntries, ret.size());
         long delta = System.currentTimeMillis() - t0;
         log.info("The blocking request with events took '" + delta + "' milliseconds");
         assertTrue("The method was blocking too long (did probably not wake up correctly", delta < 7000L);
         queue.clear();
         // and now making asynchronous putting without events (polling should detect it)
         numOfEntries = 3;
         delay = 500L;
         inhibitEvents = true;
         putter = new QueuePutter(this.queue, delay, numOfEntries, inhibitEvents);
         putter.start();
         t0 = System.currentTimeMillis();
         ret = wrapper.blockingPeek(numOfEntries, 10000L);
         assertEquals("Wrong number of entries when blocking with events", numOfEntries, ret.size());
         delta = System.currentTimeMillis() - t0;
         log.info("The blocking request without events took '" + delta + "' milliseconds");
         assertTrue("The method was blocking too long (did probably not wake up correctly", delta < 7000L);
         queue.clear();
      }
      catch (XmlBlasterException ex) {
         log.severe("Exception when testing peekMsg probably due to failed initialization of the queue " + queueType);
      }

   }


   /**
    * Tests peek() and peek(int num) and remove()
    * For a discussion of the sorting order see Javadoc of this class
    */
   private void peekMsg(I_Queue queue) {
      ME = "I_QueueTest.peekMsg(" + queue.getStorageId() + ")[" + queue.getClass().getName() + "]";
      System.out.println("***" + ME);
      try {
         //========== Test 1: peek()
         {
            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);
            for (int ii=0; ii<10; ii++) {
               I_QueueEntry result = queue.peek();
               assertTrue("Missing entry", result != null);
               assertEquals(ME+": Wrong result", queueEntries[0].getUniqueId(), result.getUniqueId());
            }
            queue.remove(); // Remove one
            for (int ii=0; ii<10; ii++) {
               I_QueueEntry result = queue.peek();
               assertTrue("Missing entry", result != null);
               assertEquals(ME+": Wrong result", queueEntries[1].getUniqueId(), result.getUniqueId());
            }
            queue.remove(); // Remove one
            for (int ii=0; ii<10; ii++) {
               I_QueueEntry result = queue.peek();
               assertTrue("Missing entry", result != null);
               assertEquals(ME+": Wrong result", queueEntries[2].getUniqueId(), result.getUniqueId());
            }
            queue.remove(); // Remove one
            for (int ii=0; ii<10; ii++) {
               I_QueueEntry result = queue.peek();
               assertTrue("Unexpected entry", result == null);
            }
            log.info("#1 Success, peek()");
         }


         //========== Test 2: peek(num)
         {
            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)
                                        };
            queue.put(queueEntries, false);

            for (int ii=-1; ii<100; ii++) {
               ArrayList results = queue.peek(ii, -1L); // does no remove
               assertTrue("Missing entry", results != null);
               int expected = ii;
               if (ii == -1 || ii >= queueEntries.length)
                  expected = queueEntries.length;
               assertEquals(ME+": Wrong number of entries returned ii=" + ii, expected, results.size());
            }

            queue.clear();
            log.info("#2 Success, peek(int)");
         }

         //========== Test 3: peekSamePriority(-1)
         {
            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)
                                        };
            queue.put(queueEntries, false);

            int[] prios = { 9, 7, 5 };
            for (int j=0; j<prios.length; j++) {
               for (int ii=0; ii<10; ii++) {
                  ArrayList results = queue.peekSamePriority(-1, -1L); // does no remove
                  assertTrue("Expected results", results != null);
                  assertEquals(ME+": Wrong number of 9 priorities", 4, results.size());
                  for (int k=0; k<results.size(); ++k)
                     assertEquals(ME+": Wrong priority returned", prios[j], ((I_QueueEntry)results.get(k)).getPriority());
               }
               for (int ii=0; ii<4; ii++) {
                  int num = queue.remove();
                  assertEquals(ME+": Expected remove", 1, num);
               }
            }

            assertEquals(ME+": Expected empty queue", 0, queue.getNumOfEntries());

            log.info("#3 Success, peekSamePriority()");
         }

         //========== Test 4: peekWithPriority(-1,7,9)
         {
            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.LOW_PRIORITY, queue.getStorageId(), true),
                         new DummyEntry(glob, PriorityEnum.MIN_PRIORITY, queue.getStorageId(), true),
                         new DummyEntry(glob, PriorityEnum.HIGH_PRIORITY, queue.getStorageId(), true),
                         new DummyEntry(glob, PriorityEnum.MAX_PRIORITY, queue.getStorageId(), true)
                                        };
            queue.put(queueEntries, false);

            for (int ii=0; ii<10; ii++) {
               ArrayList results = queue.peekWithPriority(-1, -1L, 7, 9); // does no remove
               assertTrue("Expected results", results != null);
               assertEquals(ME+": Wrong number of 9 priorities", 8, results.size());
               for (int k=0; k<results.size(); ++k) {
                  assertEquals(ME+": Wrong priority returned", (k<4)?9L:7L, ((I_QueueEntry)results.get(k)).getPriority());
               }
            }
            queue.clear();
            assertEquals(ME+": Expected empty queue", 0, queue.getNumOfEntries());

            log.info("#4 Success, peekWithPriority()");
         }


         //========== Test 5: peek(100, 60)
         {
            DummyEntry[] queueEntries = {
                         new DummyEntry(glob, PriorityEnum.NORM_PRIORITY, queue.getStorageId(), 80, true),
                         new DummyEntry(glob, PriorityEnum.HIGH_PRIORITY, queue.getStorageId(), 80, true),
                         new DummyEntry(glob, PriorityEnum.HIGH_PRIORITY, queue.getStorageId(), 80, true),
                         new DummyEntry(glob, PriorityEnum.HIGH_PRIORITY, queue.getStorageId(), 80, true),
                         new DummyEntry(glob, PriorityEnum.HIGH_PRIORITY, queue.getStorageId(), 80, true),
                                        };
            queue.put(queueEntries, false);

            try {
               ArrayList results = queue.peek(100, 60); // does no remove
               assertNotNull(ME+": the result should not be null");
               assertEquals(ME+": Expected one entry on peek(100,60)", 1, results.size());
            }
            catch (XmlBlasterException e) {
               e.printStackTrace();
               assertTrue("An exception should not occur here " + e.getMessage(), false);
            }
            
            queue.clear();
            assertEquals(ME+": Expected empty queue", 0, queue.getNumOfEntries());

            log.info("#5 Success, peek(100, 60)");
         }

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


//-----------------------------------------
   public void testRemoveWithPriority() {
      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/removeWithPriority");
         this.queue.initialize(queueId, prop);
         queue.clear();
         assertEquals(ME + "wrong size before starting ", 0, queue.getNumOfEntries());
         removeWithPriority(this.queue);
      }
      catch (XmlBlasterException ex) {
         log.severe("Exception when testing removeWithpriority probably due to failed initialization of the queue " + queueType);
      }
   }


   /**
    * Test removeWithPriority(long[])
    */
   private void removeWithPriority(I_Queue queue) {
      ME = "I_QueueTest.removeWithPriority(" + queue.getStorageId() + ")[" + queue.getClass().getName() + "]";
      System.out.println("***" + ME);
      try {
         //========== Test 1: remove prio 7 and 9
         {
           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(-1, -1L, 7, 9);

            assertEquals("number of invocations", 2, this.queueSizeListener.getCount());