import org.pentaho.di.job.JobConfiguration; //导入依赖的package包/类
public JobMap()
{
    jobMap = new Hashtable<String, Job>();
    configurationMap  = new Hashtable<String, JobConfiguration>();
    loggingMap        = new Hashtable<String, Appender>();
    
}
 

import org.pentaho.di.job.JobConfiguration; //导入依赖的package包/类
public synchronized JobConfiguration getConfiguration(String jobName) {
  for (CarteObjectEntry entry : configurationMap.keySet()) {
    if (entry.getName().equals(jobName)) {
      return configurationMap.get(entry);
    }
  }
  return null;
}
 

import org.pentaho.di.job.JobConfiguration; //导入依赖的package包/类
public synchronized JobConfiguration getConfiguration( String jobName ) {
  for ( CarteObjectEntry entry : configurationMap.keySet() ) {
    if ( entry.getName().equals( jobName ) ) {
      return getConfiguration( entry );
    }
  }
  return null;
}
 

import org.pentaho.di.job.JobConfiguration; //导入依赖的package包/类
@Override
public synchronized void jobDelegationStarted( Job delegatedJob,
                                               JobExecutionConfiguration jobExecutionConfiguration ) {

  JobConfiguration jc = new JobConfiguration( delegatedJob.getJobMeta(), jobExecutionConfiguration );
  jobMap.registerJob( delegatedJob, jc );

  delegatedJob.addDelegationListener( this );
}
 

import org.pentaho.di.job.JobConfiguration; //导入依赖的package包/类
@BeforeClass
public static void init() {
  jobMap = new JobMap();
  for ( int i = 0; i < INITIAL_JOB_MAP_SIZE; i++ ) {
    jobMap.addJob( JOB_NAME_STRING + i, JOB_ID_STRING + i, mockJob( i ), mock( JobConfiguration.class ) );
  }
}
 

import org.pentaho.di.job.JobConfiguration; //导入依赖的package包/类
@Override
public Object doCall() throws Exception {
  while ( condition.get() ) {

    int i = random.nextInt( INITIAL_JOB_MAP_SIZE );
    CarteObjectEntry entry = jobMap.getJobObjects().get( i );

    if ( entry == null ) {
      throw new IllegalStateException(
        String.format( "Returned CarteObjectEntry must not be null. EntryId = %d", i ) );
    }
    final String jobName = JOB_NAME_STRING + i;

    Job job = jobMap.getJob( entry.getName() );
    if ( job == null ) {
      throw new IllegalStateException( String.format( "Returned job must not be null. Job name = %s", jobName ) );
    }

    JobConfiguration jobConfiguration = jobMap.getConfiguration( entry.getName() );
    if ( jobConfiguration == null ) {
      throw new IllegalStateException(
        String.format( "Returned jobConfiguration must not be null. Job name = %s", jobName ) );
    }
  }

  return null;
}
 

import org.pentaho.di.job.JobConfiguration; //导入依赖的package包/类
@Override
public Exception call() throws Exception {
  Exception exception = null;
  try {
    for ( int i = 0; i < cycles; i++ ) {
      int id = generator.get();
      jobMap.addJob( JOB_NAME_STRING + id, JOB_ID_STRING + id, mockJob( id ), mock( JobConfiguration.class ) );
    }
  } catch ( Exception e ) {
    exception = e;
  }
  return exception;
}
 

import org.pentaho.di.job.JobConfiguration; //导入依赖的package包/类
public synchronized void addJob(String jobName, Job job, JobConfiguration jobConfiguration)
{
    jobMap.put(jobName, job);
    configurationMap.put(jobName, jobConfiguration);
}
 

import org.pentaho.di.job.JobConfiguration; //导入依赖的package包/类
public synchronized JobConfiguration getConfiguration(String jobName)
{
    return configurationMap.get(jobName);
}
 

import org.pentaho.di.job.JobConfiguration; //导入依赖的package包/类
/**
 * @return the configurationMap
 */
public Map<String, JobConfiguration> getConfigurationMap()
{
    return configurationMap;
}
 

import org.pentaho.di.job.JobConfiguration; //导入依赖的package包/类
/**
 * @param configurationMap the configurationMap to set
 */
public void setConfigurationMap(Map<String, JobConfiguration> configurationMap)
{
    this.configurationMap = configurationMap;
}
 

import org.pentaho.di.job.JobConfiguration; //导入依赖的package包/类
public JobMap() {
  jobMap = new Hashtable<CarteObjectEntry, Job>();
  configurationMap = new Hashtable<CarteObjectEntry, JobConfiguration>();
}
 

import org.pentaho.di.job.JobConfiguration; //导入依赖的package包/类
public synchronized void addJob(String jobName, String carteObjectId, Job job, JobConfiguration jobConfiguration) {
  CarteObjectEntry entry = new CarteObjectEntry(jobName, carteObjectId);
  jobMap.put(entry, job);
  configurationMap.put(entry, jobConfiguration);
}
 

import org.pentaho.di.job.JobConfiguration; //导入依赖的package包/类
public void replaceJob(CarteObjectEntry entry, Job job, JobConfiguration jobConfiguration) {
  jobMap.put(entry, job);
  configurationMap.put(entry, jobConfiguration);
}
 

import org.pentaho.di.job.JobConfiguration; //导入依赖的package包/类
/**
 * @return the configurationMap
 */
public Map<CarteObjectEntry, JobConfiguration> getConfigurationMap() {
  return configurationMap;
}
 

import org.pentaho.di.job.JobConfiguration; //导入依赖的package包/类
@GET
@Path("/start/{id : .+}")
@Produces({ MediaType.APPLICATION_JSON })
public JobStatus startJob(@PathParam("id") String id) {
  Job job = CarteResource.getJob(id);
  CarteObjectEntry entry = CarteResource.getCarteObjectEntry(id);
  try {
    if (job.isInitialized() && !job.isActive()) {
      // Re-create the job from the jobMeta
      //
      // We might need to re-connect to the repository
      //
      if (job.getRep() != null && !job.getRep().isConnected()) {
        job.getRep().connect(job.getRep().getUserInfo().getLogin(), job.getRep().getUserInfo().getPassword());
      }

      // Create a new job object to start from a sane state. Then replace
      // the new job in the job map
      //
      synchronized (CarteSingleton.getInstance().getJobMap()) {
        JobConfiguration jobConfiguration = CarteSingleton.getInstance().getJobMap().getConfiguration(entry);

        String carteObjectId = UUID.randomUUID().toString();
        SimpleLoggingObject servletLoggingObject = new SimpleLoggingObject(getClass().getName(), LoggingObjectType.CARTE, null);
        servletLoggingObject.setContainerObjectId(carteObjectId);

        Job newJob = new Job(job.getRep(), job.getJobMeta(), servletLoggingObject);
        newJob.setLogLevel(job.getLogLevel());

        // Discard old log lines from the old job
        //
        CentralLogStore.discardLines(job.getLogChannelId(), true);

        CarteSingleton.getInstance().getJobMap().replaceJob(entry, newJob, jobConfiguration);
        job = newJob;
      }
    }
    job.start();
  } catch (KettleException e) {
    e.printStackTrace();
  }
  return getJobStatus(id);
}
 

import org.pentaho.di.job.JobConfiguration; //导入依赖的package包/类
public synchronized void addJob( String jobName, String carteObjectId, Job job, JobConfiguration jobConfiguration ) {
  CarteObjectEntry entry = new CarteObjectEntry( jobName, carteObjectId );
  jobMap.put( entry, job );
  configurationMap.put( entry, jobConfiguration );
}
 

import org.pentaho.di.job.JobConfiguration; //导入依赖的package包/类
public synchronized void registerJob( Job job, JobConfiguration jobConfiguration ) {
  job.setContainerObjectId( UUID.randomUUID().toString() );
  CarteObjectEntry entry = new CarteObjectEntry( job.getJobMeta().getName(), job.getContainerObjectId() );
  jobMap.put( entry, job );
  configurationMap.put( entry, jobConfiguration );
}
 

import org.pentaho.di.job.JobConfiguration; //导入依赖的package包/类
public synchronized void replaceJob( CarteObjectEntry entry, Job job, JobConfiguration jobConfiguration ) {
  jobMap.put( entry, job );
  configurationMap.put( entry, jobConfiguration );
}
 

import org.pentaho.di.job.JobConfiguration; //导入依赖的package包/类
@GET
@Path( "/start/{id : .+}" )
@Produces( { MediaType.APPLICATION_JSON } )
public JobStatus startJob( @PathParam( "id" ) String id ) {
  Job job = CarteResource.getJob( id );
  CarteObjectEntry entry = CarteResource.getCarteObjectEntry( id );
  try {
    if ( job.isInitialized() && !job.isActive() ) {
      // Re-create the job from the jobMeta
      //
      // We might need to re-connect to the repository
      //
      if ( job.getRep() != null && !job.getRep().isConnected() ) {
        if ( job.getRep().getUserInfo() != null ) {
          job.getRep().connect( job.getRep().getUserInfo().getLogin(), job.getRep().getUserInfo().getPassword() );
        } else {
          job.getRep().connect( null, null );
        }
      }

      // Create a new job object to start from a sane state. Then replace
      // the new job in the job map
      //
      synchronized ( this ) {
        JobConfiguration jobConfiguration = CarteSingleton.getInstance().getJobMap().getConfiguration( entry );

        String carteObjectId = UUID.randomUUID().toString();
        SimpleLoggingObject servletLoggingObject =
          new SimpleLoggingObject( getClass().getName(), LoggingObjectType.CARTE, null );
        servletLoggingObject.setContainerObjectId( carteObjectId );

        Job newJob = new Job( job.getRep(), job.getJobMeta(), servletLoggingObject );
        newJob.setLogLevel( job.getLogLevel() );

        // Discard old log lines from the old job
        //
        KettleLogStore.discardLines( job.getLogChannelId(), true );

        CarteSingleton.getInstance().getJobMap().replaceJob( entry, newJob, jobConfiguration );
        job = newJob;
      }
    }
    job.start();
  } catch ( KettleException e ) {
    e.printStackTrace();
  }
  return getJobStatus( id );
}