TableMapStore implements MapStore<String, Table> { @Override public Map<String, Table> loadAll(Collection<String> keys) { logger.info("Load all called for multiple keys"); MultiGetResponse response = elasticsearchConnection.getClient() .prepareMultiGet() .add(TABLE_META_INDEX, TABLE_META_TYPE, keys) .execute() .actionGet(); Map<String, Table> tables = Maps.newHashMap(); for (MultiGetItemResponse multiGetItemResponse : response) { try { Table table = objectMapper.readValue(multiGetItemResponse.getResponse().getSourceAsString(), Table.class); tables.put(table.getName(), table); } catch (Exception e) { throw new RuntimeException("Error getting data for table: " + multiGetItemResponse.getId()); } } logger.info("Loaded value count: " + tables.size()); return tables; } 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 testLoadAll() 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> names = ImmutableSet.copyOf(Iterables.limit(tables.keySet(), 5)); Map<String, Table> responseTables = tableMapStore.loadAll(names); assertEquals(names.size(), responseTables.size()); for (String name : names) { compareTables(tables.get(name), responseTables.get(name)); } }
@Test(expected = NullPointerException.class) public void testLoadAllNull() throws Exception { tableMapStore.loadAll(null); }
@Test(expected = RuntimeException.class) public void testLoadAllKeyWithWrongJson() throws Exception { elasticsearchServer.getClient().prepareIndex() .setIndex(TABLE_META_INDEX) .setType(TABLE_META_TYPE) .setConsistencyLevel(WriteConsistencyLevel.ALL) .setSource("{ \"test\" : \"test\"}") .setId(TEST_TABLE) .setRefresh(true) .execute() .actionGet(); tableMapStore.loadAll(Arrays.asList(TEST_TABLE)); }