samoa-api/src/main/java/org/apache/samoa/tasks/PrequentialCVEvaluation.java [83:108]:
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
  @Override
  public void init() {
    // TODO remove the if statement
    // theoretically, dynamic binding will work here!
    // test later!
    // for now, the if statement is used by Storm

    if (builder == null) {
      builder = new TopologyBuilder();
      logger.debug("Successfully instantiating TopologyBuilder");

      builder.initTopology(evaluationNameOption.getValue());
      logger.debug("Successfully initializing SAMOA topology with name {}", evaluationNameOption.getValue());
    }

    // instantiate PrequentialSourceProcessor and its output stream
    // (sourcePiOutputStream)
    preqSource = new PrequentialSourceProcessor();
    preqSource.setStreamSource((InstanceStream) this.streamTrainOption.getValue());
    preqSource.setMaxNumInstances(instanceLimitOption.getValue());
    preqSource.setSourceDelay(sourceDelayOption.getValue());
    preqSource.setDelayBatchSize(batchDelayOption.getValue());
    builder.addEntranceProcessor(preqSource);
    logger.debug("Successfully instantiating PrequentialSourceProcessor");

    sourcePiOutputStream = builder.createStream(preqSource);
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -



samoa-api/src/main/java/org/apache/samoa/tasks/PrequentialEvaluation.java [137:167]:
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
  @Override
  public void init() {
    // TODO remove the if statement
    // theoretically, dynamic binding will work here!
    // test later!
    // for now, the if statement is used by Storm

    if (builder == null) {
      builder = new TopologyBuilder();
      logger.debug("Successfully instantiating TopologyBuilder");

      builder.initTopology(evaluationNameOption.getValue());
      logger.debug("Successfully initializing SAMOA topology with name {}", evaluationNameOption.getValue());
    }

    // instantiate PrequentialSourceProcessor and its output stream
    // (sourcePiOutputStream)
    preqSource = new PrequentialSourceProcessor();
    preqSource.setStreamSource((InstanceStream) this.streamTrainOption.getValue());
    preqSource.setMaxNumInstances(instanceLimitOption.getValue());
    preqSource.setSourceDelay(sourceDelayOption.getValue());
    preqSource.setDelayBatchSize(batchDelayOption.getValue());
    builder.addEntranceProcessor(preqSource);
    logger.debug("Successfully instantiating PrequentialSourceProcessor");

    // preqStarter = new PrequentialSourceTopologyStarter(preqSource,
    // instanceLimitOption.getValue());
    // sourcePi = builder.createEntrancePi(preqSource, preqStarter);
    // sourcePiOutputStream = builder.createStream(sourcePi);

    sourcePiOutputStream = builder.createStream(preqSource);
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -