TableMapStore implements MapStore<String, Table> { @Override public Set<String> loadAllKeys() { logger.info("Load all keys called"); SearchResponse response = elasticsearchConnection.getClient() .prepareSearch(TABLE_META_INDEX) .setTypes(TABLE_META_TYPE) .setQuery(QueryBuilders.matchAllQuery()) .setSearchType(SearchType.SCAN) .setScroll(new TimeValue(30, TimeUnit.SECONDS)) .setNoFields() .execute() .actionGet(); Set<String> ids = Sets.newHashSet(); while (true) { response = elasticsearchConnection.getClient().prepareSearchScroll(response.getScrollId()) .setScroll(new TimeValue(60000)) .execute() .actionGet(); SearchHits hits = response.getHits(); for (SearchHit hit : hits) { ids.add(hit.getId()); } if (0 == response.getHits().hits().length) { break; } } logger.info("Loaded value count: " + ids.size()); return ids; } TableMapStore(ElasticsearchConnection elasticsearchConnection); static Factory factory(ElasticsearchConnection elasticsearchConnection); @Override void store(String key, Table value); @Override void storeAll(Map<String, Table> map); @Override void delete(String key); @Override void deleteAll(Collection<String> keys); @Override Table load(String key); @Override Map<String, Table> loadAll(Collection<String> keys); @Override Set<String> loadAllKeys(); static final String TABLE_META_INDEX; static final String TABLE_META_TYPE; }
@Test public void testLoadAllKeys() throws Exception { Map<String, Table> tables = new HashMap<String, Table>(); for (int i = 0; i < 10; i++) { Table table = new Table(); table.setName(UUID.randomUUID().toString()); table.setTtl(20); tables.put(table.getName(), table); elasticsearchServer.getClient().prepareIndex() .setIndex(TABLE_META_INDEX) .setType(TABLE_META_TYPE) .setConsistencyLevel(WriteConsistencyLevel.ALL) .setSource(mapper.writeValueAsString(table)) .setId(table.getName()) .setRefresh(true) .execute() .actionGet(); } Set<String> responseKeys = tableMapStore.loadAllKeys(); for (String name : tables.keySet()) { assertTrue(responseKeys.contains(name)); } }