class CSVWrapper extends AbstractWrapper {

    @Override
    public boolean initialize() {
        setName("CSVWrapper-Thread" + (++threadCounter));
        AddressBean addressBean = getActiveAddressBean();
        dataFileName = addressBean.getPredicateValueWithException("file");
        String csvFields = addressBean.getPredicateValueWithException("fields");
        String csvFormats = addressBean.getPredicateValueWithException("formats");
        //String csvSeparator = addressBean.getPredicateValueWithDefault("separator",",");
        String value = addressBean.getPredicateValue("separator");
        String csvSeparator = (value == null || value.length() == 0) ? "," : value;
        checkPointDir = addressBean.getPredicateValueWithDefault("check-point-directory", "./csv-check-points");
        String csvStringQuote = addressBean.getPredicateValueWithDefault("quote", "\"");
        int skipFirstXLine = addressBean.getPredicateValueAsInt("skip-first-lines", 0);
        String timezone = addressBean.getPredicateValueWithDefault("timezone", CSVHandler.LOCAL_TIMEZONE_ID);
        String nullValues = addressBean.getPredicateValueWithDefault("bad-values", "");
        String strUseCounterForCheckPoint = addressBean.getPredicateValueWithDefault("use-counter-for-check-point", "false");
        samplingPeriodInMsc = addressBean.getPredicateValueAsInt("sampling", 10000);
        samplingCountPerPeriod = addressBean.getPredicateValueAsInt("sampling-count", 250);

        /*
         DEBUG_INFO(dataFile);
         */
        if (csvSeparator != null && csvSeparator.length() != 1) {
            logger.warn("The provided CSV separator:>" + csvSeparator + "< should only have  1 character, thus ignored and instead \",\" is used.");
            csvSeparator = ",";
        }

        if (csvStringQuote.length() != 1) {
            logger.warn("The provided CSV quote:>" + csvSeparator + "< should only have 1 character, thus ignored and instead '\"' is used.");
            csvStringQuote = "\"";
        }

        try {
            if (strUseCounterForCheckPoint.equalsIgnoreCase("true")) {
                useCounterForCheckPoint = true;
                logger.warn("Using counter-based check points");
            }
            //String checkPointFile = new File(checkPointDir).getAbsolutePath()+"/"+(new File(dataFile).getName())+"-"+addressBean.hashCode();
            StringBuilder checkPointFile = new StringBuilder()
                    .append(new File(checkPointDir).getAbsolutePath())
                    .append("/")
                    .append(addressBean.getVirtualSensorName())
                    .append("_")
                    .append(addressBean.getInputStreamName())
                    .append("_")
                    .append(addressBean.getWrapper())
                    .append("_")
                    .append(new File(dataFileName).getName());
            if (!handler.initialize(dataFileName.trim(), csvFields, csvFormats, csvSeparator.toCharArray()[0], csvStringQuote.toCharArray()[0], skipFirstXLine, nullValues, timezone, checkPointFile.toString())) {
                return false;
            }

            String val = FileUtils.readFileToString(new File(checkPointFile.toString()), "UTF-8");
            long lastItem = 0;
            if (val != null && val.trim().length() > 0) {
                lastItem = Long.parseLong(val.trim());
            }
            logger.warn("Latest item: " + lastItem);

            if (useCounterForCheckPoint) {
                processedLineCounter = lastItem;
            }

        } catch (IOException | NumberFormatException e) {
            logger.error("Loading the csv-wrapper failed:" + e.getMessage(), e);
            return false;
        }

        dataField = handler.getDataFields();

        logger.warn("Reading from: " + dataFileName);

        return true;
    }

}