ExcelSink implements ISink { @Override public void open() throws TexeraException{ if (cursor != CLOSED) { return; } inputOperator.open(); inputSchema = inputOperator.getOutputSchema(); outputSchema = new Schema(inputSchema.getAttributes().stream() .filter(attr -> ! attr.getName().equalsIgnoreCase(SchemaConstants._ID)) .filter(attr -> ! attr.getName().equalsIgnoreCase(SchemaConstants.PAYLOAD)) .filter(attr -> ! attr.getType().equals(AttributeType.LIST)) .toArray(Attribute[]::new)); wb = new XSSFWorkbook(); DateFormat df = new SimpleDateFormat("yyyyMMdd-HHmmss"); fileName = df.format(new Date()) + ".xlsx"; try { if (Files.notExists(excelIndexDirectory)) { Files.createDirectories(excelIndexDirectory); } fileOut = new FileOutputStream(excelIndexDirectory.resolve(fileName).toString()); } catch (IOException e) { throw new DataflowException(e); } sheet = wb.createSheet("new sheet"); Row row = sheet.createRow(0); List<String> attributeNames = outputSchema.getAttributeNames(); for(int i = 0; i < attributeNames.size(); i++) { String attributeName = attributeNames.get(i); row.createCell(i).setCellValue(attributeName); } cursor = OPENED; } ExcelSink(ExcelSinkPredicate predicate); void setInputOperator(IOperator inputOperator); IOperator getInputOperator(); @Override Schema getOutputSchema(); @Override void open(); @Override void processTuples(); @Override Tuple getNextTuple(); @Override void close(); List<Tuple> collectAllTuples(); Path getFilePath(); Schema transformToOutputSchema(Schema... inputSchema);  }
@Test public void testOpen() throws Exception { excelSink.open(); Mockito.verify(inputOperator).open(); Assert.assertEquals(new Schema(new Attribute("content", AttributeType.TEXT)), excelSink.getOutputSchema()); excelSink.close(); Files.deleteIfExists(excelSink.getFilePath()); }