class CSVSerializer implements Serializer {

    @Override
    public InputStream serialize(InputStream rawContent, DataSetMetadata metadata, long limit) {
        try {
            PipedInputStream pipe = new PipedInputStream();
            PipedOutputStream jsonOutput = new PipedOutputStream(pipe);
            // Serialize asynchronously for better performance (especially if caller doesn't consume all, see sampling).
            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(); // Skip all header lines
                    }
                    generator.writeStartArray();
                    writeLineContent(reader, metadata, generator, separator, limit);
                    generator.writeEndArray();
                    generator.flush();
                } catch (Exception e) {
                    // Consumer may very well interrupt consumption of stream (in case of limit(n) use for sampling).
                    // This is not an issue as consumer is allowed to partially consumes results, it's up to the
                    // consumer to ensure data it consumed is consistent.
                    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);
        }
    }

}