Page History

public class SPStream
{
     // Package-specific Exception types:   

        

   public static final int   int    invalid_stream_ex                =   ex                =    -20026;

     // ProcessingStatus Enumeration Constants:   

        

   public static final String   String    ps_waiting   waiting    =       "WAITING";
   public  public static final String   String    ps_ready   ready    =       "READY";
     public static final String   String    ps_queued   queued    =       "QUEUED";
     public static final String   String    ps_submitted   submitted    =       "SUBMITTED";
     public static final String   String    ps_running   running  =      =    "RUNNING";
     public static final String   String    ps_success   success    =       "SUCCESS";
     public static final String   String    ps_failed   failed    =       "FAILED";
     public static final String   String    ps_terminated   terminated    =       "TERMINATED";
     public static final String String    ps_canceled    =     ps_canceled    =    "CANCELED";
     public static final String   String    ps_skipped   skipped    =       "SKIPPED";


     // StreamStatus Enumeration Constants:   

        

   public static final String   String    ss_waiting   waiting    =       "WAITING";
     public static final String   String    ss_queued   queued =      =    "QUEUED";
     public static final String   String    ss_running   running    =       "RUNNING";
     public static final String   String    ss_success   success    =       "SUCCESS";
     public static final String   String    ss_failed   failed    =       "FAILED";
     public static final String String      ss_terminated   terminated    =       "TERMINATED";
     public static final String   String    ss_canceled   canceled    =       "CANCELED";
     public static final String   String    ss_terminating   terminating    =       "TERMINATING";
     public static final String   String    ss_canceling   canceling    =       "CANCELING";

   private  private static HashSet processingStates = new HashSet();
     static {
           processingStates.add(ps_waiting);
           processingStates.add(ps_ready);
           processingStates.add(ps_queued);
           processingStates.add(ps_submitted);
           processingStates.add(ps_running);
           processingStates.add(ps_success);
           processingStates.add(ps_failed);
           processingStates.add(ps_terminated);  
           processingStates.add(ps_canceled);
      processingStates      processingStates.add(ps_terminated);  
           processingStates.add(ps_skipped);     
     }
  
     private static HashSet finalProcessingStates = new HashSet();
     static {
           finalProcessingStates.add(ps_terminated);  
           finalProcessingStates.add(ps_canceled);
           finalProcessingStates.add(ps_success);
           finalProcessingStates.add(ps_failed);
           finalProcessingStates.add(ps_skipped);
     }
  
     private static HashSet streamStates = new HashSet();
     static {
           streamStates.add(ss_waiting);
           streamStates.add(ss_queued);
           streamStates.add(ss_running);
           streamStates.add(ss_success);
           streamStates.add(ss_failed);
           streamStates.add(ss_terminating);
           streamStates.add(ss_terminated);
           streamStates.add(ss_canceling);
           streamStates.add(ss_canceled);
     }
  
     private static HashSet finalStreamStates = new HashSet();
     static {
           finalStreamStates.add(ss_terminated);  
           finalStreamStates.add(ss_canceled);
           finalStreamStates.add(ss_success);
           finalStreamStates.add(ss_failed);
     }
  
     private static boolean allComplete(Map procStatMap, Map streamStatMap) {
           for (Iterator i = processingStates.iterator(); i.hasNext();) {
                 String state = (String)i.next();
                 int count = ((Integer)procStatMap.get(state)).intValue();
            if ( !finalProcessingStates    if ( !finalProcessingStates.contains(state) && (count != 0) )
                       return false;
           }

           for (Iterator i = streamStates.iterator(); i.hasNext();) {
                 String state = (String)i.next();
                 int count = ((Integer)streamStatMap.get(state)).intValue();
                 if ( !finalStreamStates.contains(state) && (count != 0) )
             return false;
               return false;
      }

           return true;
     }
  
     private static boolean all(String key, Map map) {
           int total = 0;
           for (Iterator i = map.values().iterator(); i.hasNext();) {
                 total += ((Integer)i.next()).intValue();
           }
           return ((Integer)map.get(key)).intValue() == total;
     }
  
     private static boolean any(String key, Map map) {
           return ((Integer)map.get(key)).intValue() != 0;
     }

     public static String calculateStreamStatus(int stream) throws ClassNotFoundException {
           String retVal = ss_waiting;
           Connection connection = null; // Database connection object
           ResultSet rset = null;
           int parentStream = 0;
     
     
           try {
                 // Get a Default Database Connection using Server Side JDBC Driver.
         // Note        // Note : This class will be loaded on the Database Server and hence use a
                 // Server Side JDBC Driver to get default Connection to Database

                 if (System.getProperty("java.vendor").toLowerCase().indexOf("oracle") != -1) {
                       // we're running as a stored procedure, connect locally:
                       connection = DriverManager.getConnection("jdbc:default:connection");
                 } else {
                        // we're running outside the DB, connect via standard oracle jdbc driver:
                       Class.forName("oracle.jdbc.driver.OracleDriver");
                       connection = DriverManager.getConnection("jdbc:oracle:thin:@glast-oracle02.slac.stanford.edu:1521:GLASTDEV","GLAST_DP_TEST","BT33%Q9]MU");
                 }

                 //
                 //   Get information about current stream
                  //

        
        
         String currentStatus;
                 PreparedStatement stmt = connection.prepareStatement(
                             "select * " +
                             "from Stream " +
                "where Stream = ? "
                        "where Stream = ? "
               );
                 stmt.setInt(1,stream);
                 rset = stmt.executeQuery(); // Execute the query, get Resultset
                 if (rset.next()) {
             currentStatus =          currentStatus = rset.getString("STREAMSTATUS");
                       parentStream = rset.getInt("PARENTSTREAM");
                 } else {
                       throw new SQLException("No such stream", "", invalid_stream_ex);
                 }
                 stmt.close();
 
        
         //
                 //   Get a count of the Processes in each state for this Stream:
                 //

                 stmt = connection.prepareStatement(
                             "select ProcessingStatus, count(ProcessingStatus) AS COUNT " +
               "from ProcessInstance " +
               "where Stream =  ? " +
               "group by ProcessingStatus "
               );
         stmt.setInt(1,stream);
         rset = stmt.executeQuery(); // Execute the query, get Resultset
         Map procStatCounts = new HashMap();
         for (Iterator i = processingStates.iterator(); i.hasNext();)
            procStatCounts.put((String)i.next(), new Integer(0));
         while (rset.next()) {
            procStatCounts.put(rset.getString("PROCESSINGSTATUS"), new Integer(rset.getInt("COUNT")));
         }
         stmt.close();
        
         //
         //  Get a count of the sub-Streams in each state for this Stream:
         //

         stmt = connection.prepareStatement(
               "select StreamStatus, count(StreamStatus) AS COUNT " +
               "from Stream " +
               "where ParentStream = ? " +
               "group by StreamStatus"
               );
         stmt.setInt(1, stream);
         rset = stmt.executeQuery();
         Map subStreamStatCountsfrom ProcessInstance " +
               "where Stream = ? " +
               "group by ProcessingStatus "
               );
         stmt.setInt(1,stream);
         rset = stmt.executeQuery(); // Execute the query, get Resultset
         Map procStatCounts = new HashMap();
                 for (Iterator i = streamStatesprocessingStates.iterator(); i.hasNext();)        
            subStreamStatCounts.put(            procStatCounts.put((String)i.next(), new Integer(0));
                 while (rset.next()) {
            subStreamStatCounts            procStatCounts.put(rset.getString("STREAMSTATUSPROCESSINGSTATUS"), new Integer(rset.getInt("COUNT")));
                 }
                 stmt.close();
         
         if (currentStatus == ss_terminating) {
            if (allComplete(procStatCounts, subStreamStatCounts)) {//
                retVal = ss_terminated;
            } else {
               retVal = ss_terminating;
            }
         } else if (currentStatus == ss_canceling) {
            if (allComplete(procStatCounts, subStreamStatCounts)) {
               retVal = ss_canceled;
            } else {
               retVal = ss_canceling;
            }
         } else if ( any(ps_running, procStatCounts) || any(ss_running, subStreamStatCounts) ) {
            retVal = ss_running;
         } else if ( any(ps_queued, procStatCounts) || any(ps_submitted, procStatCounts) || any(ss_queued, subStreamStatCounts) ) {
            retVal = ss_queued;
         } else if ( any(ps_waiting, procStatCounts) || any(ss_waiting, subStreamStatCounts) ) {
            retVal = ss_waiting;
         } else if ( any(ps_failed, procStatCounts) || any(ss_failed, subStreamStatCounts) ) {
            retVal = ss_failed;
         } else if ( all(ps_success, procStatCounts) && all(ss_success, subStreamStatCounts) ) {
            retVal = ss_success;
         }

         stmt = connection.prepareStatement("update Stream set StreamStatus = ? where Stream = ?");
         stmt.setString(1,retVal);
         stmt.setInt(2,stream);
         stmt.executeUpdate();        

         if (parentStream != 0)
            calculateStreamStatus(parentStream);
               
       } catch (SQLException ex) { // Trap SQL Errors
          ex.printStackTrace();
       } finally {
          try{
            if (connection != null || !connection.isClosed())
                connection.close();   // Close the database connection
            } catch(SQLException ex){
              ex.printStackTrace();
            }
       }

       return retVal;
   }
}

//  Get a count of the sub-Streams in each state for this Stream:
         //

         stmt = connection.prepareStatement(
               "select StreamStatus, count(StreamStatus) AS COUNT " +
               "from Stream " +
               "where ParentStream = ? " +
               "group by StreamStatus"
               );
         stmt.setInt(1, stream);
         rset = stmt.executeQuery();
         Map subStreamStatCounts = new HashMap();
         for (Iterator i = streamStates.iterator(); i.hasNext();)
            subStreamStatCounts.put((String)i.next(), new Integer(0));
         while (rset.next()) {
            subStreamStatCounts.put(rset.getString("STREAMSTATUS"), new Integer(rset.getInt("COUNT")));
         }
         stmt.close();

         if (currentStatus == ss_terminating) {
            if (allComplete(procStatCounts, subStreamStatCounts)) {
               retVal = ss_terminated;
            } else {
               retVal = ss_terminating;
            }
         } else if (currentStatus == ss_canceling) {
            if (allComplete(procStatCounts, subStreamStatCounts)) {
               retVal = ss_canceled;
            } else {
               retVal = ss_canceling;
            }
         } else if ( any(ps_running, procStatCounts) || any(ss_running, subStreamStatCounts) ) {
            retVal = ss_running;
         } else if ( any(ps_queued, procStatCounts) || any(ps_submitted, procStatCounts) || any(ss_queued, subStreamStatCounts) ) {
            retVal = ss_queued;
         } else if ( any(ps_waiting, procStatCounts) || any(ss_waiting, subStreamStatCounts) ) {
            retVal = ss_waiting;
         } else if ( any(ps_failed, procStatCounts) || any(ss_failed, subStreamStatCounts) ) {
            retVal = ss_failed;
         } else if ( all(ps_success, procStatCounts) && all(ss_success, subStreamStatCounts) ) {
            retVal = ss_success;
         }

         stmt = connection.prepareStatement("update Stream set StreamStatus = ? where Stream = ?");
         stmt.setString(1,retVal);
         stmt.setInt(2,stream);
         stmt.executeUpdate();

         if (parentStream != 0)
            calculateStreamStatus(parentStream);

       } catch (SQLException ex) { // Trap SQL Errors
          ex.printStackTrace();
       } finally {
          try{
            if (connection != null || !connection.isClosed())
                connection.close();   // Close the database connection
            } catch(SQLException ex){
              ex.printStackTrace();
            }
       }

       return retVal;
   }
}