CreateFileItemsTask extends CommonTask { @Override public void run() throws IOException { log.info("Processing partition load, with maxSize: " + maxSize); List<StorageObject> objects = this.getCloudService().listCloudStorageObjects(bucket); List<Item> items = new ArrayList<>(); String triplesFilesStats = null; long totalSize = 0; for(StorageObject object: objects) { String filename = object.getName(); boolean addFile = false; if(StringUtils.isBlank(fileMatch)) { addFile = filename.endsWith(Constants.COMPRESSED_N_TRIPLES); } else { addFile = filename.indexOf(fileMatch) > -1; } if(addFile) { long size = object.getSize().longValue(); totalSize = totalSize + size; items.add(new Item(filename, size)); if(maxSize > 0 && totalSize >= maxSize) { log.info("Got files with the maximum size of: " + totalSize); break; } } else if(FilenameUtils.TRIPLES_FILES_STATS_JSON.equals(filename)) { triplesFilesStats = filename; log.info("Found triples files stats: " + filename); } else { log.warn("Skipping file: " + filename); } } if(triplesFilesStats != null) { totalSize = 0; Map<String, Item> fileItems = new HashMap<>(); for(Item item: items) { fileItems.put(item.getKey(), item); } try { String localFile = FilenameUtils.getLocalFilePath(triplesFilesStats); this.cloudService.downloadObjectFromCloudStorage(triplesFilesStats, localFile, bucket); List<TriplesFileStats> stats = TriplesFileStats.fromJsonFile(localFile, false); for(TriplesFileStats stat: stats) { Item item = fileItems.get(stat.getFilename()); if(item != null) { long size = 0; if(stat.getProcessingTime() != null) { size = stat.getProcessingTime().longValue(); item.setWeight(size); } else if(stat.getStatements() != null) { size = stat.getStatements(); item.setWeight(size); } else if(stat.getSize() != null) { size = stat.getSize(); item.setWeight(size); } totalSize = totalSize + size; } } } catch(Exception e) { log.error("Failed to download and process triples files stats", e); } } this.addOutput("fileItems", items); log.info("Successfully processed partition load, total items: " + items.size() + ", total size: " + totalSize); }  @Override void run(); String getBucket(); void setBucket(String bucket); String getFileMatch(); void setFileMatch(String fileMatch); long getMaxSize(); void setMaxSize(long maxSize);  }
@Test @Ignore public void testRun() throws IOException { CreateFileItemsTask task = new CreateFileItemsTask(); task.setCloudService(service); task.setBucket("dbpedia"); task.run(); Map<String, Object> output = task.getOutput(); assertNotNull(output); assertEquals(1, output.size()); List<Item> items = (List<Item>) output.get("fileItems"); assertEquals(items.size(), this.triplesStat.size()); assertEquals(items.size(), this.triplesTimeStat.size()); for(Item item: items) { System.out.println("Filename: " + item.getKey()); System.out.println("Weight: " + item.getWeight() + "\n"); Long statements = this.triplesStat.get(item.getKey()); Double time = this.triplesTimeStat.get(item.getKey()); assertNotNull(statements); assertEquals(time, item.getWeight(), 0.9); } }