class CsvExtractorFactory implements IExtractorFactory<StreamConnector> {

    @Override
    public Extractor createExtractor(StreamConnector connector, ExtractorModel model) {

        String charsetName = model.getPropertyValue(CHARSET_PROPERTY);

        Charset charset = charsetName != null ? Charset.forName(charsetName) : defaultCharset;

        CSVFormat csvFormat = CSVFormat.RFC4180;
        String delimiter = model.getPropertyValue(DELIMITER_PROPERTY);
        if (delimiter != null) {
            if (delimiter.length() != 1) {
                throw new LmRuntimeException("Invalid delimiter (should be exactly one character): " + delimiter);
            }
            csvFormat = csvFormat.withDelimiter(delimiter.charAt(0));
        }

        String readFromString = model.getPropertyValue(READ_FROM_PROPERTY);
        Integer readFrom = readFromString != null ? Integer.valueOf(readFromString) : null;

        return new CsvExtractor(connector, model.getAttributes(), charset, csvFormat, readFrom);
    }

    public  CsvExtractorFactory();

}