@Test public void should_serialize_csv_with_missing_values() throws IOException { InputStream inputStream = this.getClass().getResourceAsStream("missing_values.csv"); DataSetMetadata datasetMetadata = getSimpleDataSetMetadata("character", "actor", "active"); InputStream input = serializer.serialize(inputStream, datasetMetadata, -1); String actual = IOUtils.toString(input, UTF_8); InputStream expected = this.getClass().getResourceAsStream("missing_values.csv_expected.json"); Assert.assertThat(actual, sameJSONAsFile(expected)); }
@Override public InputStream serialize(InputStream rawContent, DataSetMetadata metadata, long limit) { try { PipedInputStream pipe = new PipedInputStream(); PipedOutputStream jsonOutput = new PipedOutputStream(pipe); Runnable r = () -> { final Map<String, String> parameters = metadata.getContent().getParameters(); final String separator = parameters.get(CSVFormatFamily.SEPARATOR_PARAMETER); final char actualSeparator = separator.charAt(0); final char textEnclosureChar = getFromParameters(parameters, TEXT_ENCLOSURE_CHAR, defaultTextEnclosure); final char escapeChar = getFromParameters(parameters, CSVFormatFamily.ESCAPE_CHAR, defaultEscapeChar); try (InputStreamReader input = new InputStreamReader(rawContent, metadata.getEncoding()); CSVReader reader = new CSVReader(input, actualSeparator, textEnclosureChar, escapeChar)) { JsonGenerator generator = new JsonFactory().createGenerator(jsonOutput); int i = 0; while (i++ < metadata.getContent().getNbLinesInHeader()) { reader.readNext(); } generator.writeStartArray(); writeLineContent(reader, metadata, generator, separator, limit); generator.writeEndArray(); generator.flush(); } catch (Exception e) { LOGGER.debug("Unable to continue serialization for {}. Skipping remaining content.", metadata.getId(), e); } finally { try { jsonOutput.close(); } catch (IOException e) { LOGGER.error("Unable to close output", e); } } }; executor.execute(r); return pipe; } catch (IOException e) { throw new TDPException(CommonErrorCodes.UNABLE_TO_SERIALIZE_TO_JSON, e); } }
CSVSerializer implements Serializer { @Override public InputStream serialize(InputStream rawContent, DataSetMetadata metadata, long limit) { try { PipedInputStream pipe = new PipedInputStream(); PipedOutputStream jsonOutput = new PipedOutputStream(pipe); Runnable r = () -> { final Map<String, String> parameters = metadata.getContent().getParameters(); final String separator = parameters.get(CSVFormatFamily.SEPARATOR_PARAMETER); final char actualSeparator = separator.charAt(0); final char textEnclosureChar = getFromParameters(parameters, TEXT_ENCLOSURE_CHAR, defaultTextEnclosure); final char escapeChar = getFromParameters(parameters, CSVFormatFamily.ESCAPE_CHAR, defaultEscapeChar); try (InputStreamReader input = new InputStreamReader(rawContent, metadata.getEncoding()); CSVReader reader = new CSVReader(input, actualSeparator, textEnclosureChar, escapeChar)) { JsonGenerator generator = new JsonFactory().createGenerator(jsonOutput); int i = 0; while (i++ < metadata.getContent().getNbLinesInHeader()) { reader.readNext(); } generator.writeStartArray(); writeLineContent(reader, metadata, generator, separator, limit); generator.writeEndArray(); generator.flush(); } catch (Exception e) { LOGGER.debug("Unable to continue serialization for {}. Skipping remaining content.", metadata.getId(), e); } finally { try { jsonOutput.close(); } catch (IOException e) { LOGGER.error("Unable to close output", e); } } }; executor.execute(r); return pipe; } catch (IOException e) { throw new TDPException(CommonErrorCodes.UNABLE_TO_SERIALIZE_TO_JSON, e); } } }
CSVSerializer implements Serializer { @Override public InputStream serialize(InputStream rawContent, DataSetMetadata metadata, long limit) { try { PipedInputStream pipe = new PipedInputStream(); PipedOutputStream jsonOutput = new PipedOutputStream(pipe); Runnable r = () -> { final Map<String, String> parameters = metadata.getContent().getParameters(); final String separator = parameters.get(CSVFormatFamily.SEPARATOR_PARAMETER); final char actualSeparator = separator.charAt(0); final char textEnclosureChar = getFromParameters(parameters, TEXT_ENCLOSURE_CHAR, defaultTextEnclosure); final char escapeChar = getFromParameters(parameters, CSVFormatFamily.ESCAPE_CHAR, defaultEscapeChar); try (InputStreamReader input = new InputStreamReader(rawContent, metadata.getEncoding()); CSVReader reader = new CSVReader(input, actualSeparator, textEnclosureChar, escapeChar)) { JsonGenerator generator = new JsonFactory().createGenerator(jsonOutput); int i = 0; while (i++ < metadata.getContent().getNbLinesInHeader()) { reader.readNext(); } generator.writeStartArray(); writeLineContent(reader, metadata, generator, separator, limit); generator.writeEndArray(); generator.flush(); } catch (Exception e) { LOGGER.debug("Unable to continue serialization for {}. Skipping remaining content.", metadata.getId(), e); } finally { try { jsonOutput.close(); } catch (IOException e) { LOGGER.error("Unable to close output", e); } } }; executor.execute(r); return pipe; } catch (IOException e) { throw new TDPException(CommonErrorCodes.UNABLE_TO_SERIALIZE_TO_JSON, e); } }  }
CSVSerializer implements Serializer { @Override public InputStream serialize(InputStream rawContent, DataSetMetadata metadata, long limit) { try { PipedInputStream pipe = new PipedInputStream(); PipedOutputStream jsonOutput = new PipedOutputStream(pipe); Runnable r = () -> { final Map<String, String> parameters = metadata.getContent().getParameters(); final String separator = parameters.get(CSVFormatFamily.SEPARATOR_PARAMETER); final char actualSeparator = separator.charAt(0); final char textEnclosureChar = getFromParameters(parameters, TEXT_ENCLOSURE_CHAR, defaultTextEnclosure); final char escapeChar = getFromParameters(parameters, CSVFormatFamily.ESCAPE_CHAR, defaultEscapeChar); try (InputStreamReader input = new InputStreamReader(rawContent, metadata.getEncoding()); CSVReader reader = new CSVReader(input, actualSeparator, textEnclosureChar, escapeChar)) { JsonGenerator generator = new JsonFactory().createGenerator(jsonOutput); int i = 0; while (i++ < metadata.getContent().getNbLinesInHeader()) { reader.readNext(); } generator.writeStartArray(); writeLineContent(reader, metadata, generator, separator, limit); generator.writeEndArray(); generator.flush(); } catch (Exception e) { LOGGER.debug("Unable to continue serialization for {}. Skipping remaining content.", metadata.getId(), e); } finally { try { jsonOutput.close(); } catch (IOException e) { LOGGER.error("Unable to close output", e); } } }; executor.execute(r); return pipe; } catch (IOException e) { throw new TDPException(CommonErrorCodes.UNABLE_TO_SERIALIZE_TO_JSON, e); } }  @Override InputStream serialize(InputStream rawContent, DataSetMetadata metadata, long limit); }
CSVSerializer implements Serializer { @Override public InputStream serialize(InputStream rawContent, DataSetMetadata metadata, long limit) { try { PipedInputStream pipe = new PipedInputStream(); PipedOutputStream jsonOutput = new PipedOutputStream(pipe); Runnable r = () -> { final Map<String, String> parameters = metadata.getContent().getParameters(); final String separator = parameters.get(CSVFormatFamily.SEPARATOR_PARAMETER); final char actualSeparator = separator.charAt(0); final char textEnclosureChar = getFromParameters(parameters, TEXT_ENCLOSURE_CHAR, defaultTextEnclosure); final char escapeChar = getFromParameters(parameters, CSVFormatFamily.ESCAPE_CHAR, defaultEscapeChar); try (InputStreamReader input = new InputStreamReader(rawContent, metadata.getEncoding()); CSVReader reader = new CSVReader(input, actualSeparator, textEnclosureChar, escapeChar)) { JsonGenerator generator = new JsonFactory().createGenerator(jsonOutput); int i = 0; while (i++ < metadata.getContent().getNbLinesInHeader()) { reader.readNext(); } generator.writeStartArray(); writeLineContent(reader, metadata, generator, separator, limit); generator.writeEndArray(); generator.flush(); } catch (Exception e) { LOGGER.debug("Unable to continue serialization for {}. Skipping remaining content.", metadata.getId(), e); } finally { try { jsonOutput.close(); } catch (IOException e) { LOGGER.error("Unable to close output", e); } } }; executor.execute(r); return pipe; } catch (IOException e) { throw new TDPException(CommonErrorCodes.UNABLE_TO_SERIALIZE_TO_JSON, e); } }  @Override InputStream serialize(InputStream rawContent, DataSetMetadata metadata, long limit);  }