testresultsettoxmlconverter.java

java开源的企业总线.xmlBlaster

               assertXpathNotExists("/myRootTag/row[@num='0']", xml);
            }
            this.updateMap.clear();

            writeToFile("db.change.event.DROP", xml);

            changeDetector.checkAgain(null);
            sleep(500);
            assertEquals("Number of message is wrong", 0, this.updateMap.size());
         }

      }
      log.info("SUCCESS");
   }

   /**
    * Test synchronous all possible table changes.
    * We drive two test, one with meat and one as content less event messages.
    * If the table does not exist we expect a null ResultSet
    * @throws Exception Any type is possible
    */
   public final void testGroupedTableStates() throws Exception {
      log.info("Start testGroupedTableStates()");

      this.prefs.put("converter.rootName", "myRootTag");
      this.prefs.put("db.queryMeatStatement", "");
      this.prefs.put("alertScheduler.pollInterval", "0"); // switch off
      this.prefs.put("changeDetector.groupColName", "ICAO_ID");
      this.prefs.put("converter.addMeta", ""+true);
      this.prefs.put("changeDetector.detectStatement", "SELECT colKey, col1, col2, ICAO_ID FROM TEST_POLL ORDER BY ICAO_ID");
      this.prefs.put("mom.topicName", "db.change.event.${groupColValue}");
      
      boolean hasConverter = false;
      for (int run=0; run<2; run++) {
         if (run == 0) {
            this.prefs.put("converter.class", "");
         }
         else {        
            this.processor.shutdown();
            this.prefs.put("converter.class", "org.xmlBlaster.contrib.dbwatcher.convert.ResultSetToXmlConverter");
            hasConverter = true;
         }
         this.processor = createTest(new Info(prefs), this.updateMap);
         I_ChangeDetector changeDetector = processor.getChangeDetector();

         for (int i=0; i<2; i++) {
            log.info("Testing no table ...");
            changeDetector.checkAgain(null);
            sleep(500);
            assertEquals("Number of message is wrong", 0, this.updateMap.size());
         }

         {
         log.info("Now testing an empty table ...");
         this.dbPool.update("CREATE TABLE TEST_POLL (colKey NUMBER(10,3), col1 VARCHAR(20), col2 NUMBER(12), ICAO_ID VARCHAR(10))");
         changeDetector.checkAgain(null);
         sleep(500);
         assertEquals("Number of message is wrong", 1, this.updateMap.size());
         String xml = (String)this.updateMap.get("db.change.event.${groupColValue}");
         assertNotNull("No db.change.event.${groupColValue} message has arrived", xml);
         if (hasConverter) {
            assertXpathNotExists("/myRootTag/row[@num='0']", xml);
            assertXpathEvaluatesTo("CREATE", "/myRootTag/desc/command/text()", xml);
         }
         this.updateMap.clear();

         writeToFile("db.change.event.CREATE", xml);

         changeDetector.checkAgain(null);
         sleep(500);
         assertEquals("Number of message is wrong", 0, this.updateMap.size());
         }

         {
         log.info("Insert one row");
         this.dbPool.update("INSERT INTO TEST_POLL VALUES ('1.1', '<Bla', '9000', 'EDDI')");
         changeDetector.checkAgain(null);
         sleep(500);
         assertEquals("Number of message is wrong", 1, this.updateMap.size());
         String xml = (String)this.updateMap.get("db.change.event.EDDI");
         assertNotNull("No db.change.event.EDDI message has arrived", xml);
         if (hasConverter) {
            assertXpathEvaluatesTo("INSERT", "/myRootTag/desc/command/text()", xml);
            assertXpathEvaluatesTo("<Bla", "/myRootTag/row[@num='0']/col[@name='COL1']/text()", xml);
         }
         this.updateMap.clear();

         writeToFile("db.change.event.INSERT", xml);

         changeDetector.checkAgain(null);
         sleep(500);
         assertEquals("Number of message is wrong", 0, this.updateMap.size());
         }
               
         {
            log.info("Update one row");
            this.dbPool.update("UPDATE TEST_POLL SET col1='BXXX' WHERE ICAO_ID='EDDI'");
            changeDetector.checkAgain(null);
            sleep(500);
            assertEquals("Number of message is wrong", 1, this.updateMap.size());
            String xml = (String)this.updateMap.get("db.change.event.EDDI");
            assertNotNull("No db.change.event.EDDI message has arrived", xml);
            if (hasConverter) {
               assertXpathEvaluatesTo("UPDATE", "/myRootTag/desc/command/text()", xml);
               assertXpathEvaluatesTo("BXXX", "/myRootTag/row[@num='0']/col[@name='COL1']/text()", xml);
            }
            this.updateMap.clear();

            writeToFile("db.change.event.UPDATE", xml);

            changeDetector.checkAgain(null);
            sleep(500);
            assertEquals("Number of message is wrong", 0, this.updateMap.size());
         }

         {
            log.info("Delete one row");
            this.dbPool.update("DELETE FROM TEST_POLL WHERE ICAO_ID='EDDI'");
            changeDetector.checkAgain(null);
            sleep(500);
            assertEquals("Number of message is wrong", 1, this.updateMap.size());
            String xml = (String)this.updateMap.get("db.change.event.EDDI");
            assertNotNull("No db.change.event.EDDI message has arrived", xml);
            if (hasConverter) {
               assertXpathEvaluatesTo("DELETE", "/myRootTag/desc/command/text()", xml);
               assertXpathNotExists("/myRootTag/row[@num='0']", xml);
            }
            this.updateMap.clear();

            writeToFile("db.change.event.DELETE", xml);

            changeDetector.checkAgain(null);
            sleep(500);
            assertEquals("Number of message is wrong", 0, this.updateMap.size());
         }

         {
            log.info("Drop a table");
            this.dbPool.update("DROP TABLE TEST_POLL");
            changeDetector.checkAgain(null);
            sleep(500);
            assertEquals("Number of message is wrong", 1, this.updateMap.size());
            String xml = (String)this.updateMap.get("db.change.event.${groupColValue}");
            assertNotNull("No db.change.event.${groupColValue} message has arrived", xml);
            if (hasConverter) {
               assertXpathEvaluatesTo("DROP", "/myRootTag/desc/command/text()", xml);
               assertXpathNotExists("/myRootTag/row[@num='0']", xml);
            }
            this.updateMap.clear();

            writeToFile("db.change.event.DROP", xml);

            changeDetector.checkAgain(null);
            sleep(500);
            assertEquals("Number of message is wrong", 0, this.updateMap.size());
         }
      }
      log.info("SUCCESS");
   }

   /**
    * Test one round trip, the message content is created on the fly.  
    * You need a running database and a running xmlBlaster server.
    * @throws Exception
    */
   public final void testRoundTripWithImplicitMeat() throws Exception {
      log.info("Start testRoundTripWithImplicitMeat()");

      this.dbPool.update("CREATE TABLE TEST_POLL (colKey NUMBER(10), col1 VARCHAR(20), col2 VARCHAR(20), ICAO_ID VARCHAR(10))");
      this.dbPool.update("INSERT INTO TEST_POLL VALUES ('1', '<Bla', 'Blub', 'EDDI')");
      this.dbPool.update("INSERT INTO TEST_POLL VALUES ('2', 'Lol<', 'Lal', 'EDDF')");
      this.dbPool.update("INSERT INTO TEST_POLL VALUES ('3', 'Cl&&i', 'Clo', 'EDDP')");
      this.dbPool.update("INSERT INTO TEST_POLL VALUES ('4', 'Bl]]>Bl', 'BBBB', 'EDDI')");
      this.dbPool.update("INSERT INTO TEST_POLL VALUES ('5', 'BOO', 'BIII', 'EDDI')");
      
      this.prefs.put("converter.rootName", "myRootTag");
      this.prefs.put("db.queryMeatStatement", "");
      this.prefs.put("alertScheduler.pollInterval", "500");
      this.prefs.put("changeDetector.groupColName", "ICAO_ID");
      this.prefs.put("changeDetector.detectStatement", "SELECT col1, col2, ICAO_ID FROM TEST_POLL ORDER BY ICAO_ID");
      this.prefs.put("mom.topicName", "db.change.event.${groupColValue}");
      
      this.processor = createTest(new Info(prefs), this.updateMap);

      {
      log.info("Testing startup events ...");
      try { Thread.sleep(1500); } catch(Exception e) { /* Ignore */ }
      assertEquals("Number of message is wrong", 3, this.updateMap.size());
      String xml = (String)this.updateMap.get("db.change.event.EDDP");
      assertNotNull("No db.change.event.EDDP message has arrived", xml);
      assertXpathEvaluatesTo("EDDP", "/myRootTag/desc/ident/text()", xml);
      assertXpathEvaluatesTo("Cl&&i", "/myRootTag/row[@num='0']/col[@name='COL1']/text()", xml);
      assertXpathNotExists("/myRootTag/row[@num='1']", xml);
      
      xml = (String)this.updateMap.get("db.change.event.EDDI");
      writeToFile("db.change.event.EDDI", xml);
      assertXpathEvaluatesTo("3", "count(/myRootTag/desc/colname)", xml);
      assertXpathEvaluatesTo("EDDI", "/myRootTag/desc/ident/text()", xml);
      assertXpathEvaluatesTo("1", "count(/myRootTag/desc)", xml);
      assertXpathEvaluatesTo("3", "count(/myRootTag/row)", xml);
      assertXpathEvaluatesTo("<Bla", "/myRootTag/row[@num='0']/col[@name='COL1']/text()", xml);
      
      this.updateMap.clear();
      }
      
      {      
      log.info("Testing change event ...");
      this.dbPool.update("UPDATE TEST_POLL SET col1='BXXX' WHERE ICAO_ID='EDDP'");
      try { Thread.sleep(1500); } catch(Exception e) { /* Ignore */ }
      String xml = (String)this.updateMap.get("db.change.event.EDDP");
      assertNotNull("No db.change.event.EDDP message has arrived", xml);
      assertEquals("Number of message is wrong", 1, this.updateMap.size());
      assertXpathEvaluatesTo("BXXX", "/myRootTag/row[@num='0']/col[@name='COL1']/text()", xml);
      assertXpathNotExists("/myRootTag/row[@num='1']", xml);
      this.updateMap.clear();
      }
      
      {
         log.info("Drop a table with entries");
         this.dbPool.update("DROP TABLE TEST_POLL");
         try { Thread.sleep(1500); } catch(Exception e) { /* Ignore */ }
         assertEquals("Number of message is wrong", 3, this.updateMap.size());
         /*
         String xml = (String)this.updateMap.get("db.change.event.${groupColValue}");
         assertXpathEvaluatesTo("DROP", "/myRootTag/desc/command/text()", xml);
         assertXpathNotExists("/myRootTag/row[@num='0']", xml);
         */
         this.updateMap.clear();
         //writeToFile("db.change.event.DROP", xml);
      }
      try { Thread.sleep(1000); } catch(Exception e) { /* Ignore */ }
      
      log.info("SUCCESS");
   }

   /**
    * Test one round trip, the message content is empty sending an event only.  
    * You need a running database and a running xmlBlaster server.
    * @throws Exception
    */
   public final void testRoundTripWithoutMeat() throws Exception {
      log.info("Start testRoundTripWithoutMeat()");

      this.dbPool.update("CREATE TABLE TEST_POLL (colKey NUMBER(10), col1 VARCHAR(20), col2 VARCHAR(20), ICAO_ID VARCHAR(10))");
      this.dbPool.update("INSERT INTO TEST_POLL VALUES ('1', '<Bla', 'Blub', 'EDDI')");
      this.dbPool.update("INSERT INTO TEST_POLL VALUES ('2', 'Lol<', 'Lal', 'EDDF')");
      this.dbPool.update("INSERT INTO TEST_POLL VALUES ('3', 'Cl&&i', 'Clo', 'EDDP')");
      this.dbPool.update("INSERT INTO TEST_POLL VALUES ('4', 'Bl]]>Bl', 'BBBB', 'EDDI')");
      this.dbPool.update("INSERT INTO TEST_POLL VALUES ('5', 'BOO', 'BIII', 'EDDI')");
      
      this.prefs.put("converter.rootName", "myRootTag");
      this.prefs.put("db.queryMeatStatement", "");
      this.prefs.put("alertScheduler.pollInterval", "500");
      this.prefs.put("changeDetector.groupColName", "ICAO_ID");
      this.prefs.put("changeDetector.detectStatement", "SELECT col1, col2, ICAO_ID FROM TEST_POLL ORDER BY ICAO_ID");
      this.prefs.put("mom.topicName", "db.change.event.${groupColValue}");
      this.prefs.put("converter.class", ""); // No change detector class !
      
      this.processor = createTest(new Info(prefs), this.updateMap);

      {
      log.info("Testing startup events ...");
      try { Thread.sleep(1500); } catch(Exception e) { /* Ignore */ }
      assertEquals("Number of message is wrong", 3, this.updateMap.size());
      String xml = (String)this.updateMap.get("db.change.event.EDDP");
      assertNotNull("No db.change.event.EDDP message has arrived", xml);
      assertEquals("No content expected", "", xml);
      this.updateMap.clear();
      }
      
      {      
      log.info("Testing change event ...");
      this.dbPool.update("UPDATE TEST_POLL SET col1='BXXX' WHERE ICAO_ID='EDDP'");
      try { Thread.sleep(1500); } catch(Exception e) { /* Ignore */ }
      String xml = (String)this.updateMap.get("db.change.event.EDDP");
      assertNotNull("No db.change.event.EDDP message has arrived", xml);
      assertEquals("No content expected", "", xml);
      assertEquals("Number of message is wrong", 1, this.updateMap.size());
      this.updateMap.clear();
      }
      
      {
         log.info("Drop a table with entries");
         this.dbPool.update("DROP TABLE TEST_POLL");
         try { Thread.sleep(1500); } catch(Exception e) { /* Ignore */ }
         assertEquals("Number of message is wrong", 3, this.updateMap.size());
         this.updateMap.clear();
      }
      
      log.info("SUCCESS");
   }

   /**
    * Dump to file. 
    * @param topic The file name body
    * @param xml The file content
    * @throws Exception IOException
    */
   public static void writeToFile(String topic, String xml) throws Exception {
      java.io.File f = new java.io.File(System.getProperty("java.io.tmpdir")+System.getProperty("file.separator")+topic+".xml");
      java.io.FileOutputStream to = new java.io.FileOutputStream(f);
      to.write(xml.getBytes());
      to.close();
   }
}