replicationagent.java

java开源的企业总线.xmlBlaster

         ClassLoader cl = clazz.getClassLoader();
         StringBuffer buf = new StringBuffer();
         if (cl instanceof URLClassLoader) {
            URL[] urls = ((URLClassLoader)cl).getURLs();
            for (int i=0; i < urls.length; i++) 
               buf.append(urls[i].toString()).append("\n");
         }
         throw new IOException("init: no file found with the name '" + filename + "' : " + (buf.length() > 0 ? " classpath: " + buf.toString() : ""));
      }
   }
   
   
   /**
    * Initializes the necessary stuff (encapsulated DbWatcher and DbWriter) and starts the DbWriter.
    * Note that the DbWatcher is only started if used, i.e. if the readerInfo is not null.
    * 
    * @param readerInfo
    * @param writerInfo
    * @throws Exception
    */
   public void init(I_Info readerInfo, I_Info writerInfo) throws Exception {
      // check if the info objects are really different (they must never be the same instance !!!)
      if (readerInfo != null && writerInfo != null && readerInfo == writerInfo)
         throw new Exception("ReplicationAgent.init: the info objects are the same instance. This will lead to problems. Check your code and make sure they are separate instances");
      if (writerInfo != null) {
         log.info("setUp: Instantiating DbWriter");
         GlobalInfo.setStrippedHostname(writerInfo, GlobalInfo.UPPER_CASE);
         this.dbWriter = new DbWriter();
         this.dbWriter.init(writerInfo);
      }
      this.dbWatcher = initializeDbWatcher(readerInfo, this.dbWriter);
   }
   

   private static DbWatcher initializeDbWatcher(I_Info readerInfo, DbWriter dbWriter) throws Exception {
      DbWatcher dbWatcher = null;
      if (readerInfo != null) {
         try {
            log.info("setUp: Instantiating DbWatcher");
            GlobalInfo.setStrippedHostname(readerInfo, GlobalInfo.UPPER_CASE);
            dbWatcher = new DbWatcher();
            dbWatcher.init(readerInfo);
            
            I_DbSpecific dbSpecific = null;
            if (readerInfo != null) {
               if (readerInfo.getBoolean("replication.doBootstrap", false)) {
                  boolean needsPublisher = readerInfo.getBoolean(I_DbSpecific.NEEDS_PUBLISHER_KEY, true);
                  boolean forceCreationAndInitNo = false;
                  dbSpecific = ReplicationConverter.getDbSpecific(readerInfo, forceCreationAndInitNo); // done only on master !!!
                  readerInfo.put(I_DbSpecific.NEEDS_PUBLISHER_KEY, "" + needsPublisher); // back to original
                  I_DbPool pool = (I_DbPool)readerInfo.getObject("db.pool");
                  if (pool == null)
                     throw new Exception("ReplicationAgent.init: the db pool is null");
                  Connection conn = pool.reserve();
                  try {
                     boolean doWarn = false;
                     boolean force = false;
                     dbSpecific.bootstrap(conn, doWarn, force);
                  }
                  catch (Exception ex) {
                     conn = SpecificDefault.removeFromPool(conn, SpecificDefault.ROLLBACK_YES, pool);
                  }
                  finally {
                     conn = SpecificDefault.releaseIntoPool(conn, SpecificDefault.COMMIT_YES, pool);
                  }
               }
            }
            dbWatcher.startAlertProducers();
         }
         catch (Exception ex) {
            if (dbWriter != null) {
               try {
                  dbWriter.shutdown();
               }
               catch (Exception e) {
                  ex.printStackTrace();
               }
            }
            throw ex;
         }
      }
      return dbWatcher;
   }

   private final void shutdownDbWatcher() throws Exception {
      if (this.dbWatcher != null) {
         try {
            this.dbWatcher.shutdown();
            this.dbWatcher = null;
         }
         catch (Exception ex) {
            ex.printStackTrace();
         }
      }      
   }

   public void shutdown() {
      try {
         shutdownDbWatcher();
      }
      catch (Exception ex) {
         log.severe("An exception occured when shutting down the agent");
         ex.printStackTrace();
      }
      finally {
         if (this.dbWriter != null) {
            this.dbWriter.shutdown();
            this.dbWriter = null;
         }
      }
   }

   private static boolean needsHelp(String[] args) {
      for (int i=0; i < args.length; i++) {
         if ( args[i].equalsIgnoreCase("-h") ||
               args[i].equalsIgnoreCase("-help") ||
               args[i].equalsIgnoreCase("--h") ||
               args[i].equalsIgnoreCase("--help") ||
               args[i].equalsIgnoreCase("?")) {
            return true; 
         }
      }
      return false;
   }
   
   private void fillInfoWithCommandLine(String[] args, GlobalInfo cfgInfo) {
      String masterFilename = null;
      String slaveFilename = null;
      String isInteractiveTxt = "false";
      
      for (int i=0; i < args.length; i++) {
         if (args[i].equalsIgnoreCase("-master")) {
            if (i == (args.length-1) || args[i+1].startsWith("-")) {
               masterFilename = "default";
            }
            else
               masterFilename = args[i+1];
         }
         if (args[i].equalsIgnoreCase("-slave")) {
            if (i == (args.length-1) || args[i+1].startsWith("-")) {
               slaveFilename = "default";
            }
            else
               slaveFilename = args[i+1];
         }
         if (args[i].equalsIgnoreCase("-interactive")) {
            if (i < (args.length-1))
               isInteractiveTxt = args[i+1];
         }
      }
      cfgInfo.put("masterFilename", masterFilename);
      cfgInfo.put("slaveFilename", slaveFilename);
      cfgInfo.put("interactive", isInteractiveTxt);
      
   }

   private void displayHelp(I_Info readerInfo, I_Info writerInfo) throws Exception {
      System.out.println("usage: java org.xmlBlaster.contrib.replication.ReplicationAgent [-master masterPropertiesFilename] [-slave slavePropertiesFilename] [-interactive true/false]");
      System.out.println("for example :");
      System.out.println("   java org.xmlBlaster.contrib.replication.ReplicationAgent -master master.properties");
      System.out.println("   java org.xmlBlaster.contrib.replication.ReplicationAgent -slave slave.properties");
      System.out.println("   java org.xmlBlaster.contrib.replication.ReplicationAgent -master master.properties -slave slave.properties");
      System.out.println("   java org.xmlBlaster.contrib.replication.ReplicationAgent -master default -interactive true");
      System.out.println("   java org.xmlBlaster.contrib.replication.ReplicationAgent -master");
      System.out.println("   java org.xmlBlaster.contrib.replication.ReplicationAgent -master -slave");
      System.out.println("\nwhere in the first case it will act as a master, the second as a slave and the third as both master and slave");
      System.out.println("The fourth will act as a master with the default properties, and so the fifth.");
      if (readerInfo != null) {
         Map defaultMap = getReaderDefaultMap(readerInfo);
         showSubHelp(readerInfo, defaultMap, System.out);
      }
      System.out.println("You could have several instances of slaves running but only one master instance on the same topic.");
      System.out.println("The 'interactive' flag is false per default.");
      System.out.println("\n");
      System.out.println("The content of the property files follows the java properties syntax");
      System.out.println("\nif you want the default configuration parameters but only a few exceptions, you don't need to specify");
      System.out.println("file names for the properties. You can set these with the JVM arguments as for example:");
      System.out.println("java -Dmom.loginName=dummyName org.xmlBlaster.contrib.replication.ReplicationAgent -master -slave");
      System.out.println("\n");
      if (writerInfo != null) {
         Map defaultMap = getWriterDefaultMap(readerInfo);
         showSubHelp(writerInfo, defaultMap, System.out);
      }
   }
   
   
   public final void process(I_Info readerInfo, I_Info writerInfo) throws Exception {
      InputStreamReader isr = new InputStreamReader(System.in);
      BufferedReader br = new BufferedReader(isr);

      I_Info info = readerInfo;
      if (info == null)
         info = writerInfo;
      if (info == null)
         return;
      
      I_DbPool pool = (I_DbPool)info.getObject("db.pool");
      Connection conn = null;
      
      String prompt = "master>";
      if (readerInfo == null)
         prompt = "slave>"; 
      
      System.out.println(prompt + "make your sql statement");
      System.out.print(prompt);
      String line = null;
      while ( (line = br.readLine()) != null) {

         if (line.trim().length() < 1) {
            System.out.print(prompt);
            continue;
         }
         line = line.trim();
         if (line.equalsIgnoreCase("q") || 
               line.equalsIgnoreCase("quit") ||
               line.equalsIgnoreCase("exit") ||
               line.equalsIgnoreCase("stop") ||
               line.equalsIgnoreCase("finish")) 
            break;

         try {
            conn  = pool.reserve();
            Statement st = conn.createStatement();
            boolean ret = st.execute(line);
            if (ret) {
               ResultSet rs = st.getResultSet();
               while (rs.next()) {
                  int nmax = rs.getMetaData().getColumnCount();
                  StringBuffer buf = new StringBuffer(prompt);
                  for (int i=0; i < nmax; i++) {
                     buf.append(rs.getObject(i+1)).append("\t");
                  }
                  System.out.println(buf.toString());
               }
               System.out.println(prompt);
               rs.close();
            }
            st.close();
         } 
         catch (Exception ex) {
            ex.printStackTrace();
            conn = SpecificDefault.removeFromPool(conn, SpecificDefault.ROLLBACK_NO, pool);
         }
         finally {
            conn = SpecificDefault.releaseIntoPool(conn, SpecificDefault.COMMIT_NO, pool);
         }
         System.out.println(prompt + "make your sql statement");
         System.out.print(prompt);
      }
   
   }

   public void registerForUpdates(I_Update registeredForUpdates) {
      if (this.dbWriter != null)
         this.dbWriter.registerForUpdates(registeredForUpdates);
   }
   

}