serverentryfactory.java

java开源的企业总线.xmlBlaster

         try {
            ObjectInputStream objStream = new ObjectInputStream(is);
            Object[] obj = (Object[])objStream.readObject();
            if (obj.length < 3) {
               throw new XmlBlasterException(glob, ErrorCode.INTERNAL_ILLEGALARGUMENT, ME,
                         "Expected 3 entries in serialized object stream but got " + obj.length + " for priority=" + priority + " timestamp=" + timestamp);
            }
            MsgUnit msgUnit = (MsgUnit)obj[0];
            Integer referenceCounter = (Integer)obj[1];
            Integer historyReferenceCounter = (Integer)obj[2];
            msgUnit.setGlobal(glob);
            MsgUnitWrapper msgUnitWrapper = new MsgUnitWrapper(glob, msgUnit, storageId,
                             referenceCounter.intValue(), historyReferenceCounter.intValue(), sizeInBytes);
            msgUnitWrapper.startExpiryTimer();
            return msgUnitWrapper;
         }
         catch (Exception ex) {
            throw new XmlBlasterException(glob, ErrorCode.INTERNAL_UNKNOWN, ME, "createEntry-MsgUnitWrapper", ex);
         }
      }

      else if (ENTRY_TYPE_TOPIC_XML.equalsIgnoreCase(type)) { // still used
         try {
            ObjectInputStream objStream = new ObjectInputStream(is);
            Object[] obj = (Object[])objStream.readObject();
            if (obj.length < 2) {
               throw new XmlBlasterException(glob, ErrorCode.INTERNAL_ILLEGALARGUMENT, ME,
                         "Expected 2 entries in serialized object stream but got " + obj.length + " for priority=" + priority + " timestamp=" + timestamp);
            }
            String qos = (String)obj[0];
            String key = (String)obj[1];
            byte[] content = null;
            PublishQosServer publishQosServer = new PublishQosServer(glob, qos, true); // true marks from persistent store (prevents new timestamp)
            MsgKeyData msgKeyData = glob.getMsgKeyFactory().readObject(key);
            MsgUnit msgUnit = new MsgUnit(msgKeyData, content, publishQosServer.getData());
            TopicEntry topicEntry = new TopicEntry(glob, msgUnit, type, sizeInBytes);
            return topicEntry;
         }
         catch (Exception ex) {
            throw new XmlBlasterException(glob, ErrorCode.INTERNAL_UNKNOWN, ME, "createEntry-TopicEntry", ex);
         }
      }
      else if (ENTRY_TYPE_TOPIC_SERIAL.equalsIgnoreCase(type)) { // probably unused (not found in my tests)
         try {
            ObjectInputStream objStream = new ObjectInputStream(is);
            Object[] obj = (Object[])objStream.readObject();
            if (obj.length < 1) {
               throw new XmlBlasterException(glob, ErrorCode.INTERNAL_ILLEGALARGUMENT, ME,
                         "Expected 1 entry in serialized object stream but got " + obj.length + " for priority=" + priority + " timestamp=" + timestamp);
            }
            MsgUnit msgUnit = (MsgUnit)obj[0];
            msgUnit.setGlobal(glob);
            TopicEntry topicEntry = new TopicEntry(glob, msgUnit, type, sizeInBytes);
            return topicEntry;
         }
         catch (Exception ex) {
            throw new XmlBlasterException(glob, ErrorCode.INTERNAL_UNKNOWN, ME, "createEntry-TopicEntry", ex);
         }
      }

      else if (ENTRY_TYPE_SESSION.equalsIgnoreCase(type)) {  // still used
         try {
            ObjectInputStream objStream = new ObjectInputStream(is);
            Object[] obj = (Object[])objStream.readObject();
            if (obj.length < 1) {
               throw new XmlBlasterException(glob, ErrorCode.INTERNAL_ILLEGALARGUMENT, ME,
                         "Expected 1 entry in serialized object stream but got " + obj.length + " for priority=" + priority + " timestamp=" + timestamp);
            }
            String xmlLiteral = (String)obj[0];
            SessionEntry sessionEntry = new SessionEntry(xmlLiteral, timestamp, sizeInBytes);
            return sessionEntry;
         }
         catch (Exception ex) {
            throw new XmlBlasterException(glob, ErrorCode.INTERNAL_UNKNOWN, ME, "createEntry-TopicEntry", ex);
         }
      }
      else if (ENTRY_TYPE_SUBSCRIBE.equalsIgnoreCase(type)) {  // still used
         try {
            ObjectInputStream objStream = new ObjectInputStream(is);
            Object[] obj = (Object[])objStream.readObject();
            if (obj.length < 3) {
               throw new XmlBlasterException(glob, ErrorCode.INTERNAL_ILLEGALARGUMENT, ME,
                         "Expected 3 entries in serialized object stream but got " + obj.length + " for priority=" + priority + " timestamp=" + timestamp);
            }
            String keyLiteral = (String)obj[0];
            String qosLiteral = (String)obj[1];
            String sessionName = (String)obj[2];
            SubscribeEntry subscribeEntry = new SubscribeEntry(keyLiteral, qosLiteral, sessionName, timestamp, sizeInBytes);
            return subscribeEntry;
         }
         catch (Exception ex) {
            throw new XmlBlasterException(glob, ErrorCode.INTERNAL_UNKNOWN, ME, "createEntry-TopicEntry", ex);
         }
      }

      else if (ENTRY_TYPE_DUMMY.equalsIgnoreCase(type)) { // still used (for testing)
         DummyEntry entry = new DummyEntry(glob, PriorityEnum.toPriorityEnum(priority), new Timestamp(timestamp), storageId, sizeInBytes, persistent);
         //entry.setUniqueId(timestamp);
         return entry;
      }

      throw new XmlBlasterException(glob, ErrorCode.INTERNAL_NOTIMPLEMENTED, ME, "Persistent object '" + type + "' is not implemented");
   }

   /**
    * Is called after the instance is created.
    * @param name A name identifying this plugin.
    */
   public void initialize(org.xmlBlaster.util.Global glob) {
      this.glob = (org.xmlBlaster.engine.ServerScope)glob;

      if (log.isLoggable(Level.FINE)) log.fine("Successfully initialized");
   }

   /**
    * Allows to overwrite properties which where passed on initialize()
    * The properties which support hot configuration are depending on the used implementation
    */
   public void setProperties(Object userData) {
   }

   /**
    * Access the current Parser configuration
    */
   public Object getProperties() {
      return null;
   }

   /**
    * Measure performance for XML-ASCII versus java.io.Serializable persistence.
    * <pre>
    * java org.xmlBlaster.engine.queuemsg.ServerEntryFactory
    * </pre>
    */
   public static void main(String[] args) {
      ServerScope glob = new ServerScope(args);

      try {
         String[] persistType = new String[] { ENTRY_TYPE_MSG_SERIAL, ENTRY_TYPE_MSG_XML };

         int numRuns = 4;
         for(int ii=0; ii<numRuns; ii++) {
            for(int jj=0; jj<persistType.length; jj++) {
               PublishKey publishKey = new PublishKey(glob, "HA");
               PublishQosServer publishQosServer = new PublishQosServer(glob, "<qos><persistent/></qos>");
               publishQosServer.getData().setPriority(PriorityEnum.HIGH_PRIORITY);
               MsgUnit msgUnit = new MsgUnit(publishKey.getData(), "HO".getBytes(), publishQosServer.getData());
               StorageId storageId = new StorageId("mystore", "someid");
               MsgUnitWrapper msgUnitWrapper = new MsgUnitWrapper(glob, msgUnit, null, storageId, 0, 0, persistType[jj], -1);
               msgUnitWrapper.startExpiryTimer();
               I_EntryFactory factory = glob.getEntryFactory();

               int priority = msgUnitWrapper.getPriority();
               long timestamp = msgUnitWrapper.getUniqueId();
               String type = msgUnitWrapper.getEmbeddedType();
               boolean persistent = msgUnitWrapper.isPersistent();
               long sizeInBytes = msgUnitWrapper.getSizeInBytes();

               int numTransform = 1000;
               org.xmlBlaster.util.StopWatch stopWatchToBlob = new org.xmlBlaster.util.StopWatch();
               for(int kk=0; kk<numTransform; kk++) {
                  /*byte[] blob =*/ factory.toBlob(msgUnitWrapper);
               }
               double elapsed = stopWatchToBlob.elapsed();
               log.info("num toBlob=" + numTransform + " elapsed=" + elapsed + stopWatchToBlob.nice());

               byte[] blob = factory.toBlob(msgUnitWrapper);
               MsgUnitWrapper newWrapper = null;
               org.xmlBlaster.util.StopWatch stopWatchToObj = new org.xmlBlaster.util.StopWatch();
               for(int kk=0; kk<numTransform; kk++) {
                  newWrapper = (MsgUnitWrapper)factory.createEntry(priority,
                                              timestamp, type, persistent, sizeInBytes, new ByteArrayInputStream(blob), storageId);
               }
               elapsed = stopWatchToObj.elapsed();
               log.info("num toObj=" + numTransform + " elapsed=" + elapsed + stopWatchToObj.nice());

               log.fine("SUCESS BEFORE: " + msgUnitWrapper.toXml());
               log.fine("SUCESS AFTER: " + newWrapper.toXml());
            }
         }
      }
      catch (XmlBlasterException e) {
         System.out.println("ERROR " + e.getMessage());
      }
   }
}