企业搜索引擎开发之连接器connector（二十一）

/**
 * A {@link Runnable} that supports cancellation.
 */
public interface Cancelable extends Runnable {
  /**
   * Cancel the operation performed by this {@link Runnable}.
   * While this {@link Runnable#run} method is running in one thread this
   * may be called in another so implementors must provide any needed
   * synchronization.
   */
  public void cancel();
}

/**
 * A {@link Runnable} that supports cancellation and timeout.
 */
public interface TimedCancelable extends Cancelable {
  /**
   * Complete the operations performed by this {@link Runnable} due to the
   * expiration of its time interval. While this {@link Runnable#run} method is
   * running in one thread this may be called in another so implementors must
   * provide any needed synchronization.
   */
  public void timeout(TaskHandle taskHandle);
}

/**
 * A {@link TimedCancelable} for running a {@link Connector} batch using
 * a {@link Traverser}
 */
class CancelableBatch implements TimedCancelable {
  private static final Logger LOGGER =
    Logger.getLogger(CancelableBatch.class.getName());

  final Traverser traverser;
  final String traverserName;
  final BatchResultRecorder batchResultRecorder;
  final BatchTimeout batchTimeout;
  final BatchSize batchSize;

  /**
   * Construct a {@link CancelableBatch}.
   *
   * @param traverser {@link Traverser} for running the batch.
   * @param traverserName traverser name for logging purposes.
   * @param batchResultRecorder {@link BatchResultRecorder} for recording
   *        the result of running the batch.
   * @param batchSize hint and constraints as to the number of documents
   *        to process in the batch.
   */
  public CancelableBatch(Traverser traverser, String traverserName,
      BatchResultRecorder batchResultRecorder, BatchTimeout batchTimeout,
      BatchSize batchSize) {
    this.traverser = traverser;
    this.traverserName = traverserName;
    this.batchResultRecorder = batchResultRecorder;
    this.batchSize = batchSize;
    this.batchTimeout = batchTimeout;
  }
  
  /**
   * 取消执行
   */
  public void cancel() {
   traverser.cancelBatch();
  }
  
  /**
   * 运行超时
   */
  public void timeout(TaskHandle taskHandle) {
     batchTimeout.timeout();
  }

  public void run() {
    NDC.push("Traverse " + traverserName);
    try {
      LOGGER.fine("Begin runBatch; traverserName = " + traverserName
          + "  " + batchSize);
      BatchResult batchResult = traverser.runBatch(batchSize);
      LOGGER.fine("Traverser " + traverserName + " batchDone with result = "
          + batchResult);
      batchResultRecorder.recordResult(batchResult);
    } finally {
      NDC.remove();
    }
  }

  @Override
  public String toString() {
    return "CancelableBatch: traverser = " + traverser
           + ", batchSize = " + batchSize;
  }
}

/**
   * 取消执行
   */
  public void cancel() {
   traverser.cancelBatch();
  }
  
  /**
   * 运行超时
   */
  public void timeout(TaskHandle taskHandle) {
     batchTimeout.timeout();
  }

/**
   * Starts running a batch for this {@link ConnectorCoordinator} if a batch is
   * not already running.
   *
   * @return true if this call started a batch
   * @throws ConnectorNotFoundException if this {@link ConnectorCoordinator}
   *         does not exist.
   */
  public boolean startBatch() throws ConnectorNotFoundException;

/**
   * 开始采集
   * Starts running a batch for this {@link ConnectorCoordinator} if a batch is
   * not already running.
   *
   * @return true if this call started a batch
   */
  /* @Override */
  public synchronized boolean startBatch() {
    if (!shouldRun()) {
      return false;
    }

    BatchSize batchSize = loadManager.determineBatchSize();
    if (batchSize.getHint() == 0) {
      return false;
    }

    try {
      TraversalManager traversalManager = getTraversalManager();
      if (traversalManager == null) {
        return false;
      }
      //当前标识
      currentBatchKey = new Object();
      
      BatchCoordinator batchCoordinator = new BatchCoordinator(this);
      
      //batchCoordinator 作为 TraversalStateStore stateStore角色
      Traverser traverser = new QueryTraverser(pusherFactory,
          traversalManager, batchCoordinator, name,
          Context.getInstance().getTraversalContext(), clock);
      
      //batchCoordinator 作为 BatchResultRecorder batchResultRecorder, BatchTimeout batchTimeout角色
      
      //调用Traverser traverser的取消方法   
      //BatchResultRecorder batchResultRecorder记录运行结果；[不能由外部调用]
      //BatchTimeout batchTimeout的超时方法
      TimedCancelable batch =  new CancelableBatch(traverser, name,
          batchCoordinator, batchCoordinator, batchSize);
      taskHandle = threadPool.submit(batch);
      //threadPool.shutdown(interrupt, waitMillis)
      //taskHandle.cancel();
      
      return true;
    } catch (ConnectorNotFoundException cnfe) {
      LOGGER.log(Level.WARNING, "Connector not found - this is normal if you "
          + " recently reconfigured your connector instance: " + cnfe);
    } catch (InstantiatorException ie) {
      LOGGER.log(Level.WARNING,
          "Failed to perform connector content traversal.", ie);
      delayTraversal(TraversalDelayPolicy.ERROR);
    }
    return false;
  }

class BatchCoordinator implements TraversalStateStore,
BatchResultRecorder, BatchTimeout

class ConnectorCoordinatorImpl implements
     ConnectorCoordinator, ChangeHandler, BatchResultRecorder

BatchCoordinator batchCoordinator = new BatchCoordinator(this);
      
      //batchCoordinator 作为 TraversalStateStore stateStore角色
      Traverser traverser = new QueryTraverser(pusherFactory,
          traversalManager, batchCoordinator, name,
          Context.getInstance().getTraversalContext(), clock);

 public String getTraversalState() {
    synchronized (connectorCoordinator) {
      if (connectorCoordinator.currentBatchKey == requiredBatchKey) {
        return cachedState;
      } else {
        throw new BatchCompletedException();
      }
    }
  }

  public void storeTraversalState(String state) {
    synchronized (connectorCoordinator) {
      // Make sure our batch is still valid and that nobody has modified
      // the checkpoint while we were away.
      try {
        if ((connectorCoordinator.currentBatchKey == requiredBatchKey) &&
            isCheckpointUnmodified()) {
          connectorCoordinator.setConnectorState(state);
          cachedState = state;
        } else {
          throw new BatchCompletedException();
        }
      } catch (ConnectorNotFoundException cnfe) {
        // Connector disappeared while we were away.
        // Don‘t try to store results.
        throw new BatchCompletedException();
      }
    }
  }

/**
   * Returns the Connector‘s traversal state.
   *
   * @return String representation of the stored state, or
   *         null if no state is stored.
   * @throws ConnectorNotFoundException if this {@link ConnectorCoordinator}
   *         does not exist.
   */
  /* @Override */
  public synchronized String getConnectorState()
      throws ConnectorNotFoundException {
    return getInstanceInfo().getConnectorState();
  }

/**
   * Set the Connector‘s traversal state.
   *
   * @param state a String representation of the state to store.
   *        If null, any previous stored state is discarded.
   * @throws ConnectorNotFoundException if this {@link ConnectorCoordinator}
   *         does not exist.
   */
  /* @Override */
  public synchronized void setConnectorState(String state)
      throws ConnectorNotFoundException {
    getInstanceInfo().setConnectorState(state);
    // Must not call ChangeDetector, as this is called from a synchronized
    // block in BatchCoordinator.
  }

 //batchCoordinator 作为 BatchResultRecorder batchResultRecorder, BatchTimeout batchTimeout角色
      
      //调用Traverser traverser的取消方法   
      //BatchResultRecorder batchResultRecorder记录运行结果；[不能由外部调用]
      //BatchTimeout batchTimeout的超时方法
      TimedCancelable batch =  new CancelableBatch(traverser, name,
          batchCoordinator, batchCoordinator, batchSize);

public void recordResult(BatchResult result) {
    synchronized (connectorCoordinator) {
      if (connectorCoordinator.currentBatchKey == requiredBatchKey) {
        connectorCoordinator.recordResult(result);
      } else {
        LOGGER.fine("Ignoring a BatchResult returned from a "
            + "prevously canceled traversal batch.  Connector = "
            + connectorCoordinator.getConnectorName()
            + "  result = " + result + "  batchKey = " + requiredBatchKey);
      }
    }
  }

/**
   * Records the supplied traversal batch results.  Updates the
   * {@link LoadManager} with number of documents traversed,
   * and implements the requested {@link TraversalDelayPolicy}.
   *
   * @param result a BatchResult
   */
  /* @Override */
  public synchronized void recordResult(BatchResult result) {
    loadManager.recordResult(result);
    delayTraversal(result.getDelayPolicy());
  }

public void timeout() {
    synchronized (connectorCoordinator) {
      if (connectorCoordinator.currentBatchKey == requiredBatchKey) {
        connectorCoordinator.resetBatch();
      } else {
        LOGGER.warning("Ignoring Timeout for previously prevously canceled"
            + " or completed traversal batch.  Connector = "
            + connectorCoordinator.getConnectorName()
            + "  batchKey = "+ requiredBatchKey);
      }
    }
  }

/**
   * 取消采集
   * Halts any in-progess traversals for this {@link Connector} instance.
   * Some or all of the information collected during the current traversal
   * may be discarded.
   */
  synchronized void resetBatch() {
    if (taskHandle != null) {
      taskHandle.cancel();
    }
    taskHandle = null;
    currentBatchKey = null;
    interfaces = null;

    // Discard cached interface instances.
    traversalManager = null;
    retriever = null;
    traversalSchedule = null;
  }

/**
 * Handle for the management of a {@link Cancelable} primary task.
 */
public class TaskHandle {
  /**
   * The primary {@link Cancelable} that is run by this task to
   * perform some useful work.
   */
  final Cancelable cancelable;

  /*
   * The {@link future} for the primary task.
   */
  final Future<?> taskFuture;

  /*
   * The time the task starts.
   */
  final long startTime;

  /**
   * Create a TaskHandle.
   *
   * @param cancelable {@link Cancelable} for the primary task.
   * @param taskFuture {@link Future} for the primary task.
   * @param startTime startTime for the primary task.
   */
  TaskHandle(Cancelable cancelable, Future<?> taskFuture, long startTime) {
    this.cancelable = cancelable;
    this.taskFuture = taskFuture;
    this.startTime = startTime;
  }

  /**
   * Cancel the primary task and the time out task.
   */
  public void cancel() {
    cancelable.cancel();
    taskFuture.cancel(true);
  }

  /**
   * Return true if the primary task has completed.
   */
  public boolean isDone() {
    return taskFuture.isDone();
  }
}