import com.rethinkdb.net.Connection; //导入依赖的package包/类
/**
 * @param timeout Timeout in seconds.
 * @return Returns a free connection within the specified timeout.
 * @throws ReqlDriverError Throws error when no free connection is available within specified timeout.
 */
@SuppressWarnings("unused")
public Connection getConnection(int timeout) {
    if (pool.isClosed()) {
        throw new ReqlDriverError("Pool is not started.");
    }

    try {
        final long startRetrieve = System.currentTimeMillis();
        final Connection connection = pool.borrowObject(timeout * 1000);
        stats.add(System.currentTimeMillis() - startRetrieve);

        return new PersistentConnection(connection, () -> pool.returnObject(connection));
    } catch (Exception e) {
        LOGGER.error("Failed to retrieve connection from pool.", e);
        throw new ReqlDriverError("Failed to retrieve connection", e);
    }
}
 

import com.rethinkdb.net.Connection; //导入依赖的package包/类
@SuppressWarnings("unused")
public RethinkDBPool build() {
    ConnectionFactory factory = new ConnectionFactory(hostname, port, username, password, database);


    GenericObjectPoolConfig config = this.config;
    if (config == null) {
        config = new GenericObjectPoolConfig();
        config.setMaxTotal(maxConnections);
        config.setMinIdle(minFreeConnections);
        config.setMaxIdle(maxFreeConnections);
    }

    GenericObjectPool<Connection> pool = new GenericObjectPool<>(factory, config);
    return new RethinkDBPool(pool, timeout);
}
 

import com.rethinkdb.net.Connection; //导入依赖的package包/类
/**
 * Initialize the table, automatically create the table and indices if they do not yet exist.
 */
public void initTable() {
    try (Connection connection = connectionProvider.get()) {
        if (!hasTable(connection, tableName)) {
            R.tableCreate(tableName).optArg("primary_key", primaryKey).run(connection);
        }

        for (IndexModel index : indices) {
            String indexName = Joiner.on("_").join(index.getFields());
            if (!hasIndex(connection, indexName)) {
                IndexCreate indexCreate = R.table(tableName)
                        .indexCreate(indexName, row -> indexFieldsToReQL(row, index.getFields()));
                if (index.isGeo()) {
                    indexCreate = indexCreate.optArg("geo", true);
                }

                indexCreate.run(connection);
            }
        }
    }
}
 

import com.rethinkdb.net.Connection; //导入依赖的package包/类
/**
 * Retrieves a iterator returning all models matching the given filter.
 * <br>
 * Be sure to call {@link DAOIterator#close()} after finishing the iterator.
 *
 * @param filter The filter function which should be applied when executing the query.
 * @return An iterator for models matching the given filter.
 */
public DAOIterator<T> read(Function<Table, ReqlExpr> filter) {
    try (Connection connection = connectionProvider.get()) {
        final Table table = R.table(tableName);
        Object result = filter.apply(table).run(connection);
        if (result instanceof List) {
            List<T> list = ((List<?>) result).stream()
                    .map(map -> MAPPER.map((Map) map, clazz))
                    .collect(Collectors.toList());
            return new DAOIterator<>(list.iterator(), clazz, MAPPER);
        } else if (result instanceof Map) {
            return new DAOIterator<>(Lists.newArrayList(result).iterator(), clazz, MAPPER);
        } else if (result instanceof Cursor) {
            Cursor<?> cursor = (Cursor<?>) result;
            return new DAOIterator<>(cursor, clazz, MAPPER);
        } else {
            throw new ReqlInternalError("Unknown return type for query: " + result.getClass());
        }
    }
}
 

import com.rethinkdb.net.Connection; //导入依赖的package包/类
public RdbConnection(String host, int port, String database, String authenticationKey) {

		key = host + ":" + port;

		builder = Connection.build()
			.hostname(host)
			.port(port)
			.timeout(CONNECTION_TIME_OUT)
			.db(database);

		// add Auth key if provided
		if (!StringUtils.isNullOrEmptyTrimmed(authenticationKey)) {
			builder.authKey(authenticationKey);
		}

		// should be true
		Assert.isTrue(CONNECTION_TIME_OUT < WAIT_UNTIL_RETRY, "Time out must be smaller than retry sleep time!");

		active = false;
		lastConnect = 0;
	}
 

import com.rethinkdb.net.Connection; //导入依赖的package包/类
public Connection connect() {

		lastConnect = System.currentTimeMillis();

		try {
//			log.debug("Connecting to: " + getKey());
			Connection connection = builder.connect();
//			active = (connection != null && connection.isOpen());
			return connection;
		}
		catch (ReqlDriverError rde) {
			log.warn("ReqlDriverError: " + rde.getMessage() + (rde.getBacktrace().isPresent() ? " backtrace:" + rde.getBacktrace().get().toString() + " " : " " + getKey()));
			active = false;
			return null;
		}
	}
 

import com.rethinkdb.net.Connection; //导入依赖的package包/类
void createIndex(Class<?> clazz, Connection conn) {

		// look up @Indexed annotations in clazz and prepare data for indexing
		Field[] fields = clazz.getDeclaredFields();

		if (fields != null && fields.length > 0) {
			for (Field field : fields) {

				// ignored fields and keys are skipped
				if (field.isAnnotationPresent(Ignore.class) ||
					field.isAnnotationPresent(Id.class)) {
					continue;
				}

				// field is annotated with index
				Indexed annotation = field.getAnnotation(Indexed.class);
				if (annotation != null) {

					// only create index if not already created
					registerIndex(clazz, field, true, conn);
				}
			}
		}
	}
 

import com.rethinkdb.net.Connection; //导入依赖的package包/类
IndexInfo getIndexInfo(Class clazz, String index, Connection conn) {

		List<String> output = rdb.table(clazz).indexList().run(conn);

		if (output.size() == 0) {
			return null;
		}

		if (output.contains(index)) {

			// get info
			List<HashMap<String, Object>> status = rdb.table(clazz).indexStatus(index).run(conn);
			if (status.size() == 1) {
				return new IndexInfo(clazz, status.get(0));
			}
		}

		return null;
	}
 

import com.rethinkdb.net.Connection; //导入依赖的package包/类
/**
 * Makes sure created index is ready to use
 *
 * @param clazz entity to check indexes
 * @param conn database connection
 */
void checkIndexesAreReady(Class<?> clazz, Connection conn) {

	List<IndexInfo> info = getIndexInfo(clazz, conn);
	for (IndexInfo index : info) {

		if (!index.isReady()) {
			try {
				Thread.sleep(100); // wait some time ..

				// repeat procedure
				log.info("Index not jet ready: " + index + " ... waiting");
				checkIndexesAreReady(clazz, conn);
				break;
			}
			catch (InterruptedException e) {
				log.warn("Index check interrupted!");
				break;
			}
		}

		// all good .. go to next
	}
}
 

import com.rethinkdb.net.Connection; //导入依赖的package包/类
void registerIndex(Class clazz, String indexName, ReqlFunction1 indexDefinition, Map<String, Object> optArgs, boolean waitIndexFinished, Connection conn) {

		List<String> indexList = rdb.table(clazz).indexList().run(conn);

		if (!indexList.contains(indexName)) {
			IndexCreate idxc = rdb.table(clazz).indexCreate(indexName, indexDefinition);
			if (null != optArgs) {
				for (Map.Entry<String, Object> entry : optArgs.entrySet()) {
					idxc = idxc.optArg(entry.getKey(), entry.getValue());
				}
			}
			idxc.run(conn);
			if (waitIndexFinished) {
				rdb.table(clazz).indexWait(indexName).run(conn);
			}
		}
	}
 

import com.rethinkdb.net.Connection; //导入依赖的package包/类
/**
 * Deletes record mapped to given Object. Object is used to determine table name and id of the record to be deleted.
 *
 * @param object Mapped object to be deleted
 * @return true object/entity was deleted
 */
public Boolean delete(Object object) {

	Class<?> clazz = object.getClass();
	ClassMapper mapper = ClassMapper.getMapper(clazz);
	Object id = mapper.getId(object);
	if (id == null) {
		throw new IllegalArgumentException("Error: can not delete entity. Entity is missing field with @Id or id field is null.");
	}

	// return true if one entity was deleted
	try (Connection conn = pool.getConnection()) {
		Response res = Response.parse(table(clazz).get(id).delete().run(conn));
		return res.deleted == 1;
	}
}
 

import com.rethinkdb.net.Connection; //导入依赖的package包/类
public <T, K> List<T> getAllList(Class<T> clazz, K... id) {

		Object res;
		try (Connection conn = pool.getConnection()) {

			if (id == null || id.length == 0) {
				res = table(clazz).run(conn);
			} else {
				ClassMapper<T> classMapper = ClassMapper.getMapper(clazz);
				Object[] idProps = Arrays.stream(id).map(item -> classMapper.toIdValue(item)).toArray();
				res = table(clazz).getAll(idProps).run(conn);
			}

			return getResultList(res, clazz);
		}
	}
 

import com.rethinkdb.net.Connection; //导入依赖的package包/类
public <T, K> Collection<T> getAll(Class<T> clazz, K... id) {

		Object res;
		try (Connection conn = pool.getConnection()) {

			if (id == null || id.length == 0) {
				res = table(clazz).run(conn);
			} else {
				ClassMapper<T> classMapper = ClassMapper.getMapper(clazz);
				Object[] idProps = Arrays.stream(id).map(classMapper::toIdValue).toArray();
				res = table(clazz).getAll(idProps).run(conn);
			}

			return getResultSet(res, clazz);
		}
	}
 

import com.rethinkdb.net.Connection; //导入依赖的package包/类
/**
 * Value filtering by index
 *
 * @param clazz
 * @param <T>
 * @return
 */
/*public <T> List<T> filter(Class<T> clazz, String index, Object... values) {
	return filter(clazz, index, null, values);
}*/
public <T> List<T> filter(Class<T> clazz, ReqlFunction1 filterFunction) {

	ReqlExpr reql = table(clazz);
	if (null != filterFunction) {
		reql = reql.filter(filterFunction);
	}

	// execute queryList
	try (Connection conn = pool.getConnection()) {
		Object res = reql.run(conn);
		return getResultList(res, clazz);
	}
}
 

import com.rethinkdb.net.Connection; //导入依赖的package包/类
/**
 * Value filtering by index
 *
 * @param clazz
 * @param index
 * @param fromValue start
 * @param toValue   end
 * @param <T>
 * @return
 */
public <T> List<T> between(Class<T> clazz, String index, Object fromValue, Object toValue) {


	// get indexed field annotations
	Annotation[] annotations = indexing.getAnnotations(clazz, index);

	if (!hasAnnotation(Indexed.class, annotations)) {
		throw new RdbException(RdbException.Error.IndexNotDefined, "Missing index: '" + index + "' on: " + clazz.getName());
	}

	if (hasAnnotation(Timestamp.class, annotations)) {

		fromValue = toEpochTime(fromValue);
		toValue = toEpochTime(toValue);
	}

	try (Connection conn = pool.getConnection()) {
		Object res = table(clazz).between(fromValue, toValue).optArg("right_bound", "closed").optArg("index", index).run(conn);
		return getResultList(res, clazz);
	}
}
 

import com.rethinkdb.net.Connection; //导入依赖的package包/类
/**
 * Value filtering by index, returns a list
 *
 * @param clazz  to be filtered
 * @param index  index name
 * @param values to filter for
 * @param <T>    result type
 * @return list of found results or empty if none found
 */
public <T> List<T> queryList(Class<T> clazz, String index, Object... values) {

	// get indexed field annotations
	Annotation[] annotations = indexing.getAnnotations(clazz, index);

	if (!hasAnnotation(Indexed.class, annotations)) {
		throw new RdbException(RdbException.Error.IndexNotDefined, "Missing index: '" + index + "' on: " + clazz.getName());
	}

	ClassMapper<T> classMapper = ClassMapper.getMapper(clazz);
	values = Arrays.stream(values).map(value -> classMapper.toPropertyComponentValue(index, value)).toArray();

	try (Connection conn = pool.getConnection()) {
		ReqlExpr reql = table(clazz).getAll(values).optArg("index", index);

		// execute queryList
		Object res = reql.run(conn);
		return getResultList(res, clazz);
	}
}
 

import com.rethinkdb.net.Connection; //导入依赖的package包/类
/**
 * Value filtering by index
 *
 * @param clazz  to be filtered
 * @param index  index name
 * @param values to filter for
 * @param <T>    result type
 * @return a de-duplicated set of found results or empty if none found
 */
public <T> Collection<T> query(Class<T> clazz, String index, Object... values) {

	// get indexed field annotations
	Annotation[] annotations = indexing.getAnnotations(clazz, index);

	if (!hasAnnotation(Indexed.class, annotations)) {
		throw new RdbException(RdbException.Error.IndexNotDefined, "Missing index: '" + index + "' on: " + clazz.getName());
	}

	ClassMapper<T> classMapper = ClassMapper.getMapper(clazz);
	values = Arrays.stream(values).map(value -> classMapper.toPropertyComponentValue(index, value)).toArray();

	try (Connection conn = pool.getConnection()) {
		ReqlExpr reql = table(clazz).getAll(values).optArg("index", index);

		// execute queryList
		Object res = reql.run(conn);
		return getResultSet(res, clazz);
	}
}
 

import com.rethinkdb.net.Connection; //导入依赖的package包/类
@Test
public void connectionPoolRefreshTest() throws InterruptedException {


	RDB rdb = new RDB();

	rdb.addConnection("localhost", 28015, "test");
	rdb.addConnection("localhost", 28016, "test");
	rdb.addConnection("localhost", 28017, "test"); // this one should not exist
	rdb.addConnection("localhost", 28018, "test"); // this one should not exist

	rdb.initialize();

	for (int i = 0; i < 1000; i ++) {
		try (Connection conn = rdb.getConnection()) {
			assertTrue(conn.isOpen());
			assertEquals("localhost", conn.hostname);

			log.info("Got connection: " + conn.hostname + ":" + conn.port);

			Thread.sleep(1000);
		}
	}
}
 

import com.rethinkdb.net.Connection; //导入依赖的package包/类
@Test
public void ignoreNullField() {

	EntityOne entity = TestUtils.randomEntityOne();

	// this will make RDB skip this field on write
	entity.eight = null;

	Object returnedId = rdb.create(entity);
	Assert.assertEquals(entity.userId, returnedId);

	try (Connection conn = rdb.getConnection()) {
		Boolean hasFieldTwo = rdb.table(EntityOne.class).get(returnedId).keys().contains("two").run(conn);
		Assert.assertTrue(hasFieldTwo);

		Boolean hasFieldEight = rdb.table(EntityOne.class).get(returnedId).keys().contains("eight").run(conn);
		Assert.assertFalse(hasFieldEight);
	}
}
 

import com.rethinkdb.net.Connection; //导入依赖的package包/类
@Test
public void getArrayIntegers() {

	try (Connection conn = rdb.getConnection()) {

		// integer array
		int[] ids = new int[]{1, 2, 3};

		// we provide an array of floats as ID
		rdb.table(EntityArray.class).insert(r.hashMap("id", r.array(1.0f, 2.0f, 3.0f))).run(conn);

		EntityArray res = rdb.get(EntityArray.class, ids);

		// id arrays have equal values
		Assert.assertEquals(ids.length, res.id.length);
		for (int i = 0; i < ids.length; i++) {
			Assert.assertEquals(ids[i], res.id[i]);
		}
	}
}