CombineTermStatsTask extends CommonTask { @Override public void run() throws IOException { Map<String, Long> allTermStats = new HashMap<String, Long>(); for(String instanceId: this.processors) { String statsFile = instanceId + Constants.DOT_JSON; String localStatsFile = Utils.TEMP_FOLDER + statsFile; try { this.getCloudService().downloadObjectFromCloudStorage(statsFile, localStatsFile, bucket); Map<String, Long> termStats = FileUtils.jsonFileToMap(localStatsFile); for(Entry<String, Long> term: termStats.entrySet()) { String key = term.getKey(); Long value = term.getValue(); if(allTermStats.containsKey(key)) { value = allTermStats.get(key) + value; } allTermStats.put(key, value); } log.info("Evms analysed: " + allTermStats.size() + ", terms"); } catch(IOException e) { log.error("failed to download file: " + localStatsFile, e); if(!(e.getMessage().indexOf(GoogleMetaData.NOT_FOUND) >= 0)) { throw e; } } } if(StringUtils.isNotBlank(this.termStatsFile)) { log.info("Serializing term stats to file: " + this.termStatsFile); String localFile = Utils.TEMP_FOLDER + this.termStatsFile; FileUtils.objectToJsonFile(localFile, allTermStats); this.getCloudService().uploadFileToCloudStorage(localFile, bucket); } if(!allTermStats.isEmpty()) { List<Item> items = new ArrayList<>(); for(Entry<String, Long> item: allTermStats.entrySet()) { Item anItem = (new Item()).setKey(item.getKey()).setWeight(item.getValue()); items.add(anItem); } this.addOutput("termItems", items); log.info("Successfully created term stats: " + items); } }  @Override void run(); String getBucket(); void setBucket(String bucket); Collection<String> getProcessors(); void setProcessors(Collection<String> processors); String getTermStatsFile(); void setTermStatsFile(String termStatsFile);  }
@SuppressWarnings("unchecked") @Test public void testRun() throws IOException { CombineTermStatsTask task = new CombineTermStatsTask(); task.setCloudService(service); task.setBucket("swetodblp"); task.setTermStatsFile("term_stats.json"); Set<String> processors = Sets.newHashSet("ecarf-evm-1422704950901", "ecarf-evm-1422704950902", "ecarf-evm-1422704950903", "ecarf-evm-1422704950904"); task.setProcessors(processors); task.run(); Map<String, Object> output = task.getOutput(); assertNotNull(output); assertEquals(1, output.size()); List<Item> items = (List<Item>) output.get("termItems"); assertNotNull(items); assertEquals(28, items.size()); }