container/src/main/java/com/streamsets/datacollector/event/handler/remote/RemoteDataCollector.java
@Override
  public List<PipelineAndValidationStatus> getRemotePipelinesWithChanges() throws PipelineException {
    List<PipelineAndValidationStatus> pipelineAndValidationStatuses = new ArrayList<>();
    for (Pair<PipelineState, Map<String, String>> pipelineStateAndOffset: stateEventListener.getPipelineStateEvents()) {
      PipelineState pipelineState = pipelineStateAndOffset.getLeft();
      Map<String, String> offset = pipelineStateAndOffset.getRight();
      String name = pipelineState.getPipelineId();
      String rev = pipelineState.getRev();
      boolean isClusterMode = (pipelineState.getExecutionMode() != ExecutionMode.STANDALONE) ? true : false;
      List<WorkerInfo> workerInfos = new ArrayList<>();
      String title;
      int runnerCount = 0;
      if (pipelineStore.hasPipeline(name)) {
        title = pipelineStore.getInfo(name).getTitle();
        Runner runner = manager.getRunner(name, rev);
        if (isClusterMode) {
          workerInfos = getWorkers(runner.getSlaveCallbackList(CallbackObjectType.METRICS));
        }
        runnerCount = runner.getRunnerCount();
      } else {
        title = null;
      }
      pipelineAndValidationStatuses.add(new PipelineAndValidationStatus(
          getSchGeneratedPipelineName(name, rev),
          title,
          rev,
          pipelineState.getTimeStamp(),
          true,
          pipelineState.getStatus(),
          pipelineState.getMessage(),
          workerInfos,
          isClusterMode,
          getSourceOffset(name, offset),
          null,
          runnerCount
      ));
    }
    return pipelineAndValidationStatuses;
  }
@Override
  public List<PipelineAndValidationStatus> getRemotePipelinesWithChanges() throws PipelineException {
    List<PipelineAndValidationStatus> pipelineAndValidationStatuses = new ArrayList<>();
    for (Pair<PipelineState, Map<String, String>> pipelineStateAndOffset: stateEventListener.getPipelineStateEvents()) {
      PipelineState pipelineState = pipelineStateAndOffset.getLeft();
      Map<String, String> offset = pipelineStateAndOffset.getRight();
      String name = pipelineState.getPipelineId();
      String rev = pipelineState.getRev();
      boolean isClusterMode = (pipelineState.getExecutionMode() != ExecutionMode.STANDALONE) ? true : false;
      List<WorkerInfo> workerInfos = new ArrayList<>();
      String title;
      int runnerCount = 0;
      if (pipelineStore.hasPipeline(name)) {
        title = pipelineStore.getInfo(name).getTitle();
        Runner runner = manager.getRunner(name, rev);
        if (isClusterMode) {
          workerInfos = getWorkers(runner.getSlaveCallbackList(CallbackObjectType.METRICS));
        }
        runnerCount = runner.getRunnerCount();
      } else {
        title = null;
      }
      pipelineAndValidationStatuses.add(new PipelineAndValidationStatus(
          getSchGeneratedPipelineName(name, rev),
          title,
          rev,
          pipelineState.getTimeStamp(),
          true,
          pipelineState.getStatus(),
          pipelineState.getMessage(),
          workerInfos,
          isClusterMode,
          getSourceOffset(name, offset),
          null,
          runnerCount
      ));
    }
    return pipelineAndValidationStatuses;
  }
@ Override public List < PipelineAndValidationStatus > getRemotePipelinesWithChanges ( ) throws PipelineException { List < PipelineAndValidationStatus > pipelineAndValidationStatuses = new ArrayList <> ( ) ; for ( Pair < PipelineState , Map < String , String > > pipelineStateAndOffset : stateEventListener . getPipelineStateEvents ( ) ) { PipelineState pipelineState = pipelineStateAndOffset . getLeft ( ) ; Map < String , String > offset = pipelineStateAndOffset . getRight ( ) ; String name = pipelineState . getPipelineId ( ) ; String rev = pipelineState . getRev ( ) ; boolean isClusterMode = ( pipelineState . getExecutionMode ( ) != ExecutionMode . STANDALONE ) ? true : false ; List < WorkerInfo > workerInfos = new ArrayList <> ( ) ; String title ; int runnerCount = 0 ; if ( pipelineStore . hasPipeline ( name ) ) { title = pipelineStore . getInfo ( name ) . getTitle ( ) ; Runner runner = manager . getRunner ( name , rev ) ; if ( isClusterMode ) { workerInfos = getWorkers ( runner . getSlaveCallbackList ( CallbackObjectType . METRICS ) ) ; } runnerCount = runner . getRunnerCount ( ) ; } else { title = null ; } pipelineAndValidationStatuses . add ( new PipelineAndValidationStatus ( getSchGeneratedPipelineName ( name , rev ) , title , rev , pipelineState . getTimeStamp ( ) , true , pipelineState . getStatus ( ) , pipelineState . getMessage ( ) , workerInfos , isClusterMode , getSourceOffset ( name , offset ) , null , runnerCount ) ) ; } return pipelineAndValidationStatuses ; }
Returns info about remote pipelines that have changed since the last sending of events
Returns info about remote pipelines that have changed since the last sending of events