producer consumer model like scenario is not working

Question

This is the scenario. I have to poll a ftp server in a timed interval and get the csv files. Then these CSV files has to be parsed and send as a input to some business logic. I have done it in this way.

FTPClientPolling (Producer)

public class FTPClientPolling {

    private static FTPClientPolling instance = null;

    private FTPClientPolling() {
    }

    public synchronized static FTPClientPolling getInstance() {
        if (instance == null) {
            logger.info("Object created for Client Polling");
            instance = new FTPClientPolling();
            initializeFTPClient();
        }
        return instance;
    }

    public static void initializeFTPClient() {
        // initialize the values from properties file
    }

    public void startPolling() {

        FTPClient ftpClient = null;
        try {
            //connecting to ftp server

            //iterating the files in it
            FTPFile[] filesList = ftpClient.listFiles();
            for (FTPFile tmpFile : filesList) {
                //..
                File tempFile = File.createTempFile(tmpFile.getName(), null);
                FileOutputStream fileOut = new FileOutputStream(tempFile);
                ftpClient.retrieveFile(tmpFile.getName(), fileOut);

                //adding the file to the Queue of the file processor
                FileProcessor.getInstance().getFilesToBeProcessedQueue().add(tempFile);
            }

            if (ftpClient.isConnected())
                ftpClient.disconnect();
        } catch (Exception e) {
            //logging
        } finally {
            //closing ftpclient
        }

    }
}

FTPClientPollingTasker (Producer Tasker)

public class FTPClientTasker extends TimerTask {
    private static Long timeInterval = 10000l;

    @Override
    public void run() {
        FTPClientPolling.getInstance().startPolling();
    }

    public static void start() {
        TimerTask timerTask = new FTPClientTasker();
        Timer timer = new Timer();
        timer.scheduleAtFixedRate(timerTask, timeInterval, timeInterval);
    }

    public static void main(String[] args) {
        start();
    }
}

FileProcessor (Consumer)

public final class FileProcessor {

    private static FileProcessor instance = null;
    private Queue<File> filesToBeProcessedQueue = new ArrayBlockingQueue<File>(10);

    private FileProcessor() {
    }

    public synchronized static FileProcessor getInstance() {
        if (instance == null) {
            instance = new FileProcessor();
        }
        return instance;
    }

    public void run() {
        while (!filesToBeProcessedQueue.isEmpty()) {
            processSyncFiles(filesToBeProcessedQueue.poll());
        }
    }

    private void processSyncFiles(File inputFile) {
        try {
            HashMap<String, Boolean> outputConsolidation = new HashMap<String, Boolean>();

            FileReader fileReader = new FileReader(inputFile);
            List<InputBean> csvContentsList = CSVParser.readContentsFromCSV(fileReader, new InputBean());
            for (InputBean inputBean : csvContentsList) {
                boolean output = false;
                // some business logic
                outputConsolidation.put(inputBean.toString(), output);
            }
        } catch (Exception e) {
            //logging
        }
    }

    public synchronized Queue<File> getFilesToBeProcessedQueue() {
        return filesToBeProcessedQueue;
    }
}

FileProcessor Tasker (Consumer Scheduler) This class creates a Tasker for FileProcessor and runs it in a scheduled interval.

public final class FileProcessorTasker extends TimerTask {

    private static Long timeInterval = 5000l;

    @Override
    public void run() {
        FileProcessor.getInstance().run();
    }

    public static void start() {
        TimerTask timerTask = new FileProcessorTasker();
        Timer timer = new Timer();
        timer.schedule(timerTask, timeInterval, timeInterval);
    }

    public static void main(String[] args) {
        FileProcessorTasker.start();
    }
}

Both the programs are working well individually. But when linked together through the filesToBeProcessedQueue it doesn't seems to be working. The problem is FTPClientPolling creates an object of FileProcessor and adds the file to the Queue. But FileProcessorTasker creates another object of FileProcessor which has queue size as zero. This two different objects is the problem. How is it creating two objects, when the class is a singleton. Am I missing something in the singleton implementation?

Answer 1

First of all, Don't use Timer and TimerTask. Use a ExecutorService for multi-threading.

And Use Eager Initialization in your Singleton classes. Or double checked locking of null in order to make you Singleton really singleton.

FTPClientPolling.java

public class FTPClientPolling {

    private static FTPClientPolling instance = new FTPClientPolling();

    private FTPClientPolling() {
        logger.info("Object created for Client Polling");
        initializeFTPClient();
    }

    public static FTPClientPolling getInstance() {
        return instance;
    }

    public static void initializeFTPClient() {
        // initialize the values from properties file
    }

    public void startPolling() {

        FTPClient ftpClient = null;
        try {
            //connecting to ftp server

            //iterating the files in it
            FTPFile[] filesList = ftpClient.listFiles();
            for (FTPFile tmpFile : filesList) {
                //..
                File tempFile = File.createTempFile(tmpFile.getName(), null);
                FileOutputStream fileOut = new FileOutputStream(tempFile);
                ftpClient.retrieveFile(tmpFile.getName(), fileOut);

                //adding the file to the Queue of the file processor
                FileProcessor.getInstance().getFilesToBeProcessedQueue().add(tempFile);
            }

            if (ftpClient.isConnected())
                ftpClient.disconnect();
        } catch (Exception e) {
            //logging
        } finally {
            //closing ftpclient
        }

    }
}

FileProcessor.java

public final class FileProcessor {

    private static FileProcessor instance = new FileProcessor();
    private Queue<File> filesToBeProcessedQueue = new ArrayBlockingQueue<File>(10);

    private FileProcessor() {
    }

    public static FileProcessor getInstance() {
        return instance;
    }

    public void run() {
        while (!filesToBeProcessedQueue.isEmpty()) {
            processSyncFiles(filesToBeProcessedQueue.poll());
        }
    }

    private void processSyncFiles(File inputFile) {
        try {
            HashMap<String, Boolean> outputConsolidation = new HashMap<String, Boolean>();

            FileReader fileReader = new FileReader(inputFile);
            List<InputBean> csvContentsList = CSVParser.readContentsFromCSV(fileReader, new InputBean());
            for (InputBean inputBean : csvContentsList) {
                boolean output = false;
                // some business logic
                outputConsolidation.put(inputBean.toString(), output);
            }
        } catch (Exception e) {
            //logging
        }
    }

    public synchronized Queue<File> getFilesToBeProcessedQueue() {
        return filesToBeProcessedQueue;
    }
}

Read this post for more information.