CSVReaderDataPool extends FileReaderDataPool { @Override public void close() { super.close(); try { tempFileWriter.close(); if (isWriteEnabled() && hasChanges.get()) { File initialFile = new File(filePath + ".initial"); Files.move(new File(filePath), initialFile); Files.move(tempFile, new File(filePath)); initialFile.delete(); } tempFile.delete(); } catch (IOException e) { logger.error("Error while closing the CSV dataset", e); } } CSVReaderDataPool(CSVDataPool configuration); @Override void init(); @Override void close(); @Override void writeRow(DataPoolRow row); @Override Object postProcess(String line); Vector<String> getHeaders(String readOneLine); Vector<String> splitCSV(String readOneLine); @Override void doFirst_(); @Override void addRow(Object row); static final Logger logger; }
@Test public void testEmpty() { DataSet<?> pool = getDataPool("empty.csv"); Assert.assertNull(pool.next()); pool.close(); }
@Test public void testCSVReaderDataPoolPut() throws IOException { File tempFile = FileHelper.extractResourceToTempFile(this.getClass(), "testCSVReaderDataPoolPut.csv"); DataSet<?> pool = getDataPool(tempFile, true); ExecutorService threadPool = Executors.newCachedThreadPool(); for(int i=0;i<5;i++) { threadPool.submit(() -> { DataPoolRow row = null; while((row=pool.next())!=null) { try { ((SimpleStringMap)row.getValue()).put("Col4", "test"); } finally { if(row != null) { row.commit(); } } } }); } pool.close(); PlanRunnerResultAssert.assertEquals(getClass(), "testCSVReaderDataPoolPut.expected.csv", tempFile); }
@Test public void testCSVReaderDataPoolToString() { DataSet<?> pool = getDataPool("File.csv"); Assert.assertEquals("Col1=row11 Col2=row12 Col3=row13 Col4= Col5=", ((SimpleStringMap)pool.next().getValue()).toString()); pool.close(); }