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

 private final MimeTypeDetector mimeTypeDetector = new MimeTypeDetector();

  private DBClient client;
  private String connectionUrl;
  private String connectorName;
  private String login;
  private String password;
  private String sqlQuery;
  private String authZQuery;
  private String googleConnectorWorkDir;
  private String primaryKeys;
  private String xslt;
  private String driverClassName;
  private String documentURLField;
  private String documentIdField;
  private String baseURL;
  private String lobField;
  private String fetchURLField;
  private String lastModifiedDate;
  private String extMetadataType;
  private int numberOfRows = 500;
  private Integer minValue = -1;
  private boolean publicFeed = true;
  private boolean parameterizedQueryFlag = false;
  private Boolean nullsSortLow = null;
  private Collator collator;

  public DBContext() {
  }

  public void init() throws DBException {
    client.setDBContext(this);

    // If the NULL value sort behaviour has not been explicitly overriden
    // in the configuration, fetch it from the DatabaseMetadata.
    if (nullsSortLow == null) {
      nullsSortLow = client.nullsAreSortedLow();
      if (nullsSortLow == null) {
        throw new DBException("nullsSortLowFlag must be set in configuration.");
      }
    }
  }

private boolean hasCustomCollationQuery = false;
  protected DBContext dbContext;
  protected SqlSessionFactory sqlSessionFactory;
  protected DatabaseType databaseType;

  static {
    org.apache.ibatis.logging.LogFactory.useJdkLogging();
  }

  public DBClient() {
  }

  public void setDBContext(DBContext dbContext) throws DBException {
    this.dbContext = dbContext;
    generateSqlMap();
    this.sqlSessionFactory = getSqlSessionFactory(generateMyBatisConfig());
    LOG.info("DBClient for database " + getDatabaseInfo() + " is instantiated");
    this.databaseType = getDatabaseType();
  }

  /**
   * Constructor used for testing purpose. DBCLient initialized with sqlMap
   * having crawl query without CDATA section.
   */
  @VisibleForTesting
  DBClient(DBContext dbContext) throws DBException {
    this.dbContext = dbContext;
    this.sqlSessionFactory = getSqlSessionFactory(generateMyBatisConfig());
    this.databaseType = getDatabaseType();
  }

  private SqlSessionFactory getSqlSessionFactory(String config) {
    try {
      SqlSessionFactoryBuilder builder = new SqlSessionFactoryBuilder();
      return builder.build(new StringReader(config));
    } catch (RuntimeException e) {
      throw new RuntimeException("XML is not well formed", e);
    }
  }

  /**
   * @return a SqlSession
   */
  @VisibleForTesting
  SqlSession getSqlSession()
      throws SnapshotRepositoryRuntimeException {
    try {
      return sqlSessionFactory.openSession();
    } catch (RuntimeException e) {
      Throwable cause = (e.getCause() != null &&
          e.getCause() instanceof SQLException) ? e.getCause() : e;
      LOG.log(Level.WARNING, "Unable to connect to the database.", cause);
      throw new SnapshotRepositoryRuntimeException(
          "Unable to connect to the database.", cause);
    }
  }

/**
   * @param skipRows number of rows to skip in the database.
   * @param maxRows max number of rows to return.
   * @return rows - subset of the result of executing the SQL query. E.g.,
   *         result table with columns id and lastName and two rows will be
   *         returned as
   *
   *         <pre>
   *         [{id=1, lastName=last_01}, {id=2, lastName=last_02}]
   * </pre>
   * @throws DBException
   */
  public List<Map<String, Object>> executePartialQuery(int skipRows, int maxRows)
      throws SnapshotRepositoryRuntimeException {
    // TODO(meghna): Think about a better way to scroll through the result set.
    List<Map<String, Object>> rows;
    LOG.info("Executing partial query with skipRows = " + skipRows + " and "
        + "maxRows = " + maxRows);
    SqlSession session = getSqlSession();
    try {
      rows = session.selectList("IbatisDBClient.getAll", null,
                                new RowBounds(skipRows, maxRows));
      LOG.info("Sucessfully executed partial query with skipRows = "
          + skipRows + " and maxRows = " + maxRows);
    } catch (RuntimeException e) {
      checkDBConnection(session, e);
      rows = new ArrayList<Map<String, Object>>();
    } finally {
      session.close();
    }
    LOG.info("Number of rows returned " + rows.size());
    return rows;
  }

/**
   * 实际调用的默认是这个实现类
   * @author Administrator
   *
   */
  private class PartialQueryStrategy implements QueryStrategy {
    private int skipRows = 0;

    @Override
    public List<Map<String, Object>> executeQuery() {
      return dbClient.executePartialQuery(skipRows,
            dbContext.getNumberOfRows());
    }

    @Override
    public void resetCursor() {
      skipRows = 0;
    }

    @Override
    public void updateCursor(List<Map<String, Object>> rows) {
      skipRows += rows.size();
    }

    @Override
    public void logComplete() {
      LOG.info("Total " + skipRows
          + " records are crawled during this crawl cycle");
    }
  }

/**
   * 重启后都是重新开始获取数据，不记录批次信息
   * Function for fetching database rows and providing a collection of
   * snapshots.
   */
  public List<DocumentSnapshot> executeQueryAndAddDocs()
      throws SnapshotRepositoryRuntimeException {
    List<Map<String, Object>> rows = null;

    try {
      rows = queryStrategy.executeQuery();
    } catch (SnapshotRepositoryRuntimeException e) {
      LOG.info("Repository Unreachable. Resetting DB cursor to "
          + "start traversal from begining after recovery.");
      queryStrategy.resetCursor();
      LOG.warning("Unable to connect to the database\n" + e.toString());
      throw new SnapshotRepositoryRuntimeException(
          "Unable to connect to the database.", e);
    }
    if (rows.size() == 0) {
      queryStrategy.logComplete();
      LOG.info("Crawl cycle of database is complete. Resetting DB cursor to "
          + "start traversal from begining");
      queryStrategy.resetCursor();
    } else {
      queryStrategy.updateCursor(rows);
    }

    if (traversalContext == null) {
      LOG.info("Setting Traversal Context");
      traversalContext = traversalContextManager.getTraversalContext();
      JsonDocument.setTraversalContext(traversalContext);
    }

    return getDocList(rows);
  }

/**
   * 将数据包装为List<DocumentSnapshot>
   * @param rows
   * @return
   */
  private List<DocumentSnapshot> getDocList(List<Map<String, Object>> rows) {
    LOG.log(Level.FINE, "Building document snapshots for {0} rows.",
        rows.size());
    List<DocumentSnapshot> docList = Lists.newArrayList();
    for (Map<String, Object> row : rows) {
      try {
        DocumentSnapshot snapshot = docBuilder.getDocumentSnapshot(row);
        if (snapshot != null) {
          if (LOG.isLoggable(Level.FINER)) {
            LOG.finer("DBSnapshotRepository returns document with docID "
                + snapshot.getDocumentId());
          }
          docList.add(snapshot);
        }
      } catch (DBException e) {
        // See the similar log message in DBSnapshot.getDocumentHandle.
        LOG.log(Level.WARNING, "Cannot convert database record to snapshot "
            + "for record " + row, e);
      }
    }
    LOG.info(docList.size() + " document(s) to be fed to GSA");
    return docList;
  }

/**
 * Iterates over the collections of {@link DocumentSnapshot} objects
 * produced by a {@code RepositoryHandler}.
 */
public class RepositoryHandlerIterator
    extends AbstractIterator<DocumentSnapshot> {
  private final RepositoryHandler repositoryHandler;
  private Iterator<DocumentSnapshot> current;

  /**
   * @param repositoryHandler RepositoryHandler object for fetching DB rows in
   *        DocumentSnapshot form.
   */
  public RepositoryHandlerIterator(RepositoryHandler repositoryHandler) {
    this.repositoryHandler = repositoryHandler;
    this.current = Iterators.emptyIterator();
  }

  @Override
  protected DocumentSnapshot computeNext() {
    if (current.hasNext()) {
      return current.next();
    } else {
      current = repositoryHandler.executeQueryAndAddDocs().iterator();
      if (current.hasNext()) {
        return current.next();
      } else {
        return endOfData();
      }
    }
  }
}

/**
 * An iterable over the database rows. The main building block for
 * interacting with the diffing package.
 */
public class DBSnapshotRepository
    implements SnapshotRepository<DocumentSnapshot> {
  private final RepositoryHandler repositoryHandler;

  public DBSnapshotRepository(RepositoryHandler repositoryHandler) {
    this.repositoryHandler = repositoryHandler;
  }

  @Override
  public Iterator<DocumentSnapshot> iterator()
      throws SnapshotRepositoryRuntimeException {
    return new RepositoryHandlerIterator(repositoryHandler);
  }

  @Override
  public String getName() {
    return DBSnapshotRepository.class.getName();
  }
}