testtopiclifecycle.java

java开源的企业总线.xmlBlaster

         return ""; // never reached
      }
   }

   /**
    * unSubscribe a message.
    */
   public void unSubscribeMsg() {
      log.info("unSubscribing a volatile message ...");
      try {
         // Subscribe for the volatile message
         UnSubscribeKey sk = new UnSubscribeKey(glob, subscribeReturnQos.getSubscriptionId());
         UnSubscribeQos sq = new UnSubscribeQos(glob);
         con.unSubscribe(sk.toXml(), sq.toXml());
         log.info("UnSubscribing of '" + publishOid + "' done");
      } catch(XmlBlasterException e) {
         log.severe("unSubscribe() XmlBlasterException: " + e.getMessage());
         assertTrue("unSubscribe - XmlBlasterException: " + e.getMessage(), false);
      }
   }

   /**
    * Retrieve a dump of xmlBlaster to analyse
    */
   private String getDump() {
      try {
         GetKey gk = new GetKey(glob, "__cmd:?dump");
         GetQos gq = new GetQos(glob);
         MsgUnit[] msgs = con.get(gk.toXml(), gq.toXml());
         assertEquals("Did not expect returned msg for get()", 1, msgs.length);
         return msgs[0].getContentStr();
      }
      catch (XmlBlasterException e) {
         fail("Didn't expect an exception in get(): " + e.getMessage());
      }
      return "";
   }

   /**
    * THIS IS THE TEST
    * <p>
    * We traverse the transitions
    * <pre>
    * Start -[2]->  ALIVE (3 sec)
    *       -[6]->  UNREFERENCED (3 sec)
    *       -[11]-> DEAD
    * <pre>
    * as described in requirement engine.message.lifecycle by sending some expiring messages (see
    * state transition brackets in requirement)
    * </p>
    */
   public void testExpiry() {
      log.info("Entering testExpiry ...");
      this.updateInterceptor.clear();

      {  // topic transition from START -> [2] -> ALIVE (3 sec)
         long topicDestroyDelay = 6000L;
         long msgLifeTime = 3000L;
         sendExpiringMsg(true, topicDestroyDelay, msgLifeTime); 
         assertEquals("numReceived after sending", 0, this.updateInterceptor.waitOnUpdate(1000L, 0)); // no message arrived?
         String dump = getDump();
         log.fine(dump);
         // Expecting something like:
         // <TopicHandler id='http_192_168_1_4_3412/topic/TestTopicLifeCycleMsg' state='ALIVE'>
         //  <uniqueKey>TestTopicLifeCycleMsg</uniqueKey>
         assertTrue("Missing topic", dump.indexOf("<uniqueKey>"+publishOid+"</uniqueKey>") != -1);
         assertTrue("Topic in wrong state:" + dump, dump.indexOf("TestTopicLifeCycleMsg' state='ALIVE'") != -1);
      }


      {  // topic transition from ALIVE -> [6] -> UNREFERENCED (3 sec)
         try { Thread.sleep(3500L); } catch( InterruptedException i) {}
         String dump = getDump();
         // Expecting something like:
         // <TopicHandler id='http_192_168_1_4_3412/topic/TestTopicLifeCycleMsg' state='UNREFERENCED'>
         //  <uniqueKey>TestTopicLifeCycleMsg</uniqueKey>
         assertTrue("Missing topic", dump.indexOf("<uniqueKey>"+publishOid+"</uniqueKey>") != -1);
         assertTrue("Topic in wrong state:" + dump, dump.indexOf("TestTopicLifeCycleMsg' state='UNREFERENCED'") != -1);
      }

      {  // topic transition from UNREFERENCED -> [11] -> DEAD
         log.info("Sleeping for another 5 sec, the topic (with destroyDelay=6sec) should be dead then");
         try { Thread.sleep(6000); } catch( InterruptedException i) {}
         // Topic should be destroyed now

         String dump = getDump();
         log.fine("IS DEAD?"+dump);
         assertTrue("Not expected a dead topic", dump.indexOf("<uniqueKey>"+publishOid+"</uniqueKey>") == -1);
      }
      log.info("SUCCESS testExpiry");
   }

   /**
    * THIS IS THE TEST
    * <p>
    * We traverse the transitions
    * <pre>
    * Start -[2]->  ALIVE (3 sec)
    *       -[6]->  UNREFERENCED (3 sec)
    *       -[5]->  ALIVE (3 sec)
    *       -[11]-> DEAD
    * <pre>
    * as described in requirement engine.message.lifecycle by sending some expiring messages (see
    * state transition brackets in requirement)
    * </p>
    */
   public void testUnreferencedAlive() throws Exception {
      log.info("Entering testUnreferencedAlive ...");
      this.updateInterceptor.clear();

      {  log.info("topic transition from START -> [2] -> ALIVE (3 sec)");
         long topicDestroyDelay = 6000L;
         long msgLifeTime = 3000L;
         sendExpiringMsg(true, topicDestroyDelay, msgLifeTime); 
         assertEquals("numReceived after sending", 0, this.updateInterceptor.waitOnUpdate(1000L, 0)); // no message arrived?
         String dump = getDump();
         log.fine(dump);
         // Expecting something like:
         // <TopicHandler id='http_192_168_1_4_3412/topic/TestTopicLifeCycleMsg' state='ALIVE'>
         //  <uniqueKey>TestTopicLifeCycleMsg</uniqueKey>
         assertTrue("Missing topic", dump.indexOf("<uniqueKey>"+publishOid+"</uniqueKey>") != -1);
         assertTrue("Topic in wrong state:" + dump, dump.indexOf("TestTopicLifeCycleMsg' state='ALIVE'") != -1);
      }

      {  log.info("topic transition from ALIVE -> [6] -> UNREFERENCED (3 sec)");
         try { Thread.sleep(3500L); } catch( InterruptedException i) {}
         String dump = getDump();
         // Expecting something like:
         // <TopicHandler id='http_192_168_1_4_3412/topic/TestTopicLifeCycleMsg' state='UNREFERENCED'>
         //  <uniqueKey>TestTopicLifeCycleMsg</uniqueKey>
         assertTrue("Missing topic", dump.indexOf("<uniqueKey>"+publishOid+"</uniqueKey>") != -1);
         assertTrue("Topic in wrong state:" + dump, dump.indexOf("TestTopicLifeCycleMsg' state='UNREFERENCED'") != -1);
      }

      {  log.info("topic transition from UNREFERENCED -> [5] -> ALIVE (3 sec)");
         long msgLifeTime = 3000L;
         sendExpiringMsg(true, 0L, msgLifeTime); 
         assertEquals("numReceived after sending", 0, this.updateInterceptor.waitOnUpdate(1000L, 0)); // no message arrived?
         String dump = getDump();
         log.fine(dump);
         // Expecting something like:
         // <TopicHandler id='http_192_168_1_4_3412/topic/TestTopicLifeCycleMsg' state='ALIVE'>
         //  <uniqueKey>TestTopicLifeCycleMsg</uniqueKey>
         assertTrue("Missing topic", dump.indexOf("<uniqueKey>"+publishOid+"</uniqueKey>") != -1);
         assertTrue("Topic in wrong state:" + dump, dump.indexOf("TestTopicLifeCycleMsg' state='ALIVE'") != -1);
         //System.out.println(dump);
         // This assert could find "__sys__UserList" instead of the wanted "TestTopicLifeCycleMsg"
         //assertXpathEvaluatesTo(publishOid, "//uniqueKey", dump);
      }

      {  log.info("topic transition from ALIVE -> [10] -> DEAD");
         boolean forceDestroy = true;
         EraseReturnQos[] erq = sendErase(forceDestroy);
         assertEquals("erase failed", 1, erq.length);
         String dump = getDump();
         assertTrue("Not expected a dead topic:" + dump, dump.indexOf("<uniqueKey>"+publishOid+"</uniqueKey>") == -1);
      }

      {  log.info("topic transition from ALIVE -> [10] -> DEAD with XPath subscription");
         subscribeXPathMsg();
         long topicDestroyDelay = 0L;
         long msgLifeTime = 0L;
         String oid = sendExpiringXPathMsg(topicDestroyDelay, msgLifeTime); 
         assertEquals("numReceived after sending", 1, this.updateInterceptor.waitOnUpdate(1000L, oid, Constants.STATE_OK));
         assertEquals("", 1, this.updateInterceptor.getMsgs().length);
         String dump = getDump();
         assertTrue("Not expected a dead topic:" + dump, dump.indexOf("<uniqueKey>"+oid+"</uniqueKey>") == -1);
         // assert does not work because of other internal topics:
         //assertXpathNotExists("//uniqueKey", dump);
         unSubscribeMsg();
      }

      log.info("SUCCESS testUnreferencedAlive");
   }

   /**
    * THIS IS THE TEST
    * <p>
    * We traverse the transitions
    * <pre>
    * Start -[2]->  ALIVE (0 sec)
    *       -[6]->  UNREFERENCED (0 sec)
    *       -[11]-> DEAD
    * <pre>
    * as described in requirement engine.message.lifecycle by sending some expiring messages (see
    * state transition brackets in requirement)<br />
    * Please see individual test for a description
    * </p>
    */
   public void testVolatile() {
      log.info("Entering testVolatile ...");
      this.updateInterceptor.clear();

      {  // topic transition from START -> [2] -> ALIVE -> DEAD
         long topicDestroyDelay = 0L;
         long msgLifeTime = 0L;
         sendExpiringMsg(true, topicDestroyDelay, msgLifeTime); 
         assertTrue("Not expected a dead topic", getDump().indexOf("<uniqueKey>"+publishOid+"</uniqueKey>") == -1);
      }
      log.info("SUCCESS testVolatile");
   }

   /**
    * THIS IS THE TEST
    * Transitions [1] -> [4] -> [6] -> [11]
    */
   public void testSubscribeVolatile() {
      log.info("Entering testSubscribeVolatile ...");
      this.updateInterceptor.clear();

      {  // topic transition from START -> [1] -> UNCONFIGURED
         subscribeMsg();
         if (log.isLoggable(Level.FINE)) log.fine("Retrieving initial dump=" + getDump());
         String dump = getDump();
         log.fine(dump);
         // Expecting something like:
         // <TopicHandler id='http_192_168_1_4_3412/topic/TestTopicLifeCycleMsg' state='UNCONFIGURED'>
         //  <uniqueKey>TestTopicLifeCycleMsg</uniqueKey>
         assertTrue("Missing topic", dump.indexOf("<uniqueKey>"+publishOid+"</uniqueKey>") != -1);
         assertTrue("Topic in wrong state:" + dump, dump.indexOf("TestTopicLifeCycleMsg' state='UNCONFIGURED'") != -1);
      }

      {  // topic transition from UNCONFIGURED -> [4] -> ALIVE
         long topicDestroyDelay = 0L;
         long msgLifeTime = 0L;
         sendExpiringMsg(true, topicDestroyDelay, msgLifeTime); 
         assertEquals("numReceived after sending", 1, this.updateInterceptor.waitOnUpdate(2000L, 1));
         String dump = getDump();
         log.fine(dump);
         // Expecting something like:
         // <TopicHandler id='http_192_168_1_4_3412/msg/TestTopicLifeCycleMsg' state='ALIVE'>
         //  <uniqueKey>TestTopicLifeCycleMsg</uniqueKey>
         assertTrue("Missing topic", dump.indexOf("<uniqueKey>"+publishOid+"</uniqueKey>") != -1);
         assertTrue("Topic in wrong state:" + dump, dump.indexOf("TestTopicLifeCycleMsg' state='ALIVE'") != -1);
      }

      {  // topic transition from ALIVE -> [6] -> UNREFERENCED
         try { Thread.sleep(1000L); } catch( InterruptedException i) {}
         unSubscribeMsg();
         // topic transition from UNREFERENCED -> [11] DEAD (wait 200 millis as this is done by timeout thread (async))
         try { Thread.sleep(200L); } catch( InterruptedException i) {}
         String dump = getDump();
         assertTrue("Not expected a dead topic:" + dump, dump.indexOf("<uniqueKey>"+publishOid+"</uniqueKey>") == -1);
      }
      log.info("SUCCESS testSubscribeVolatile");
   }

   /**
    * THIS IS THE TEST
    * Transitions [1] -> [13] -> [9]
    */
   public void testUnconfiguredSubscribeSubscribe() {
      log.info("Entering testUnconfiguredSubscribeSubscribe ...");
      this.updateInterceptor.clear();

      {  // topic transition from START -> [1] -> UNCONFIGURED
         subscribeMsg();
         if (log.isLoggable(Level.FINE)) log.fine("Retrieving initial dump=" + getDump());
         String dump = getDump();