@Override
    public void run() {
        interrupted = false;
        while (!interrupted && !Thread.interrupted()) {
            Object input = take();
            if (input != null) {
                submit(input);
            }
        }
        logger.info("interrupted");
    }