testunsub.java

java开源的企业总线.xmlBlaster

      PublishReturnQos publishReturnQos = null;
      try {
         MsgUnit msgUnit = new MsgUnit(xmlKey, senderContent.getBytes(), "<qos></qos>");
         publishReturnQos = senderConnection.publish(msgUnit);
         publishOid = publishReturnQos.getKeyOid();
         log.info("Success: Publishing done, returned oid=" + publishOid);
      } catch(XmlBlasterException e) {
         log.warning("XmlBlasterException: " + e.getMessage());
         assertTrue("publish - XmlBlasterException: " + e.getMessage(), false);
      }

      assertTrue("returned publishOid == null", publishOid != null);
      assertTrue("returned publishOid: " + publishReturnQos.toXml(), 0 != publishOid.length());
   }

   /**
    * TEST: subscribe and unSubscribe on an empty topic (without a publish)
    * The unSubscribe is done with an oid instead of a subId
    */
   public void testSubscribeUnSubscribeOid()
   {
      log.info("Starting testSubscribeUnSubscribeOid()");
      publishOid = "";

      String oid = "SomeDummySubscribe";
      SubscribeReturnQos subRet = null;
      try {
         SubscribeKey sk = new SubscribeKey(glob, oid);
         SubscribeQos sq = new SubscribeQos(glob);
         subRet = senderConnection.subscribe(sk.toXml(), sq.toXml());
         log.info("testSubscribeUnSubscribeOid() subscribed to " + subRet.getSubscriptionId());
      }
      catch (XmlBlasterException e) {
         log.severe("testSubscribeUnSubscribeOid() subscribe failed: " + e.getMessage());
         fail("testSubscribeUnSubscribeOid() subscribe failed: " + e.getMessage());
      }

      try {
         // !! Here we unsubscribe with the oid instead of using the subId !!
         UnSubscribeKey uk = new UnSubscribeKey(glob, oid);
         UnSubscribeQos uq = new UnSubscribeQos(glob);
         UnSubscribeReturnQos[] urq = senderConnection.unSubscribe(uk.toXml(), uq.toXml());
         log.info("testSubscribeUnSubscribeOid() unSubscribed");
         assertEquals("Return wrong", 1, urq.length);
         assertEquals("SubId wrong", subRet.getSubscriptionId(), urq[0].getSubscriptionId());
      }
      catch (XmlBlasterException e) {
         log.severe("testSubscribeUnSubscribeOid() unSubscribe failed: " + e.getMessage());
         fail("testSubscribeUnSubscribeOid() unSubscribe failed: " + e.getMessage());
      }
   }

   /**
    * TEST: subscribe and unSubscribe on an empty topic (without a publish)
    */
   public void testSubscribeUnSubscribeEmpty()
   {
      log.info("Starting testSubscribeUnSubscribeEmpty()");
      publishOid = "";

      SubscribeReturnQos subRet = null;
      try {
         SubscribeKey sk = new SubscribeKey(glob, "SomeDummySubscribe");
         SubscribeQos sq = new SubscribeQos(glob);
         subRet = senderConnection.subscribe(sk.toXml(), sq.toXml());
         log.info("testSubscribeUnSubscribeEmpty() subscribed to " + subRet.getSubscriptionId());
      }
      catch (XmlBlasterException e) {
         log.severe("testSubscribeUnSubscribeEmpty() subscribe failed: " + e.getMessage());
         fail("testSubscribeUnSubscribeEmpty() subscribe failed: " + e.getMessage());
      }

      try {
         UnSubscribeKey uk = new UnSubscribeKey(glob, subRet.getSubscriptionId());
         UnSubscribeQos uq = new UnSubscribeQos(glob);
         UnSubscribeReturnQos[] urq = senderConnection.unSubscribe(uk.toXml(), uq.toXml());
         log.info("testSubscribeUnSubscribeEmpty() unSubscribed");
         assertEquals("Return wrong", 1, urq.length);
         assertEquals("SubId wrong", subRet.getSubscriptionId(), urq[0].getSubscriptionId());
      }
      catch (XmlBlasterException e) {
         log.severe("testSubscribeUnSubscribeEmpty() unSubscribe failed: " + e.getMessage());
         fail("testSubscribeUnSubscribeEmpty() unSubscribe failed: " + e.getMessage());
      }
   }


   /**
    * TEST: Publish a message, subscribe on it with XPATH and
    *       unSubscribe again with the returned oid.
    */
   public void testSubscribeUnSubscribeExact()
   {
      log.info("Starting testSubscribeUnSubscribeExact()");
      numReceived = 0;
      doPublish();           // Feed some data
      subscribeXPath();    // Subscribe to it
      waitOnUpdate(2000L);
      assertEquals("numReceived after publishing", 1, numReceived); // message arrived?
      unSubscribeExact();  // cancel XPATH subscription with XPATH-subscription-oid
   }


   /**
    * TEST: Publish a message, subscribe on it with XPATH and
    *       unSubscribe again with the same XPATH query.
    */
   public void testSubscribeUnSubscribeXPath()
   {
      log.info("Starting testSubscribeUnSubscribeXPath()");
      numReceived = 0;
      doPublish();           // Feed some data
      subscribeXPath();    // Subscribe to it
      waitOnUpdate(2000L);
      assertEquals("numReceived after publishing", 1, numReceived); // message arrived?
      unSubscribeXPath();  // cancel with XPATH syntax
   }


   /**
    * This is the callback method invoked from xmlBlaster
    * delivering us a new asynchronous message. 
    * @see org.xmlBlaster.client.I_Callback#update(String, UpdateKey, byte[], UpdateQos)
    */
   public String update(String cbSessionId, UpdateKey updateKey, byte[] content, UpdateQos updateQos)
   {
      if (log.isLoggable(Level.FINER)) log.finer("Receiving update of a message ...");

      numReceived += 1;

      //assertEquals("Wrong sender, used="+senderName+" updated="+updateQos.getSender().getRelativeName(), senderName, updateQos.getSender().getRelativeName());
      assertEquals("Wrong oid of message returned", publishOid, updateKey.getOid());
      assertEquals("Message content is corrupted", new String(senderContent), new String(content));
      assertEquals("Message contentMime is corrupted", contentMime, updateKey.getContentMime());
      assertEquals("Message contentMimeExtended is corrupted", contentMimeExtended, updateKey.getContentMimeExtended());

      messageArrived = true;
      return "";
   }


   /**
    * Little helper, waits until the variable 'messageArrive' is set
    * to true, or returns when the given timeout occurs.
    * @param timeout in milliseconds
    */
   private void waitOnUpdate(final long timeout)
   {
      long pollingInterval = 50L;  // check every 0.05 seconds
      if (timeout < 50)  pollingInterval = timeout / 10L;
      long sum = 0L;
      while (!messageArrived) {
         try {
            Thread.sleep(pollingInterval);
         }
         catch( InterruptedException i)
         {}
         sum += pollingInterval;
         if (sum > timeout) {
            log.warning("Timeout of " + timeout + " occurred");
            break;
         }
      }
      messageArrived = false;
   }


   /**
    * Method is used by TestRunner to load these tests
    */
   public static Test suite()
   {
       TestSuite suite= new TestSuite();
       String loginName = "TestUnSub/5";
       Global glob = new Global();
       suite.addTest(new TestUnSub(glob, "testSubscribeUnSubscribeEmpty", loginName));
       suite.addTest(new TestUnSub(glob, "testSubscribeUnSubscribeOid", loginName));
       suite.addTest(new TestUnSub(glob, "testSubscribeUnSubscribeExact", loginName));
       suite.addTest(new TestUnSub(glob, "testSubscribeUnSubscribeXPath", loginName));
       return suite;
   }


   /**
    * Invoke: java org.xmlBlaster.test.qos.TestUnSub
    * @deprecated Use the TestRunner from the testsuite to run it:<p />
    * <pre>   java -Djava.compiler= junit.textui.TestRunner org.xmlBlaster.test.qos.TestUnSub</pre>
    */
   public static void main(String args[])
   {
      TestUnSub testSub = new TestUnSub(new Global(args), "TestUnSub", "TestUnSub/5");

      testSub.setUp();
      testSub.testSubscribeUnSubscribeOid();
      testSub.tearDown();

      testSub.setUp();
      testSub.testSubscribeUnSubscribeEmpty();
      testSub.tearDown();

      testSub.setUp();
      testSub.testSubscribeUnSubscribeXPath();
      testSub.tearDown();

      testSub.setUp();
      testSub.testSubscribeUnSubscribeExact();
      testSub.tearDown();
   }
}