@Override
    public void run() {
        try {
            while (true) {
                logger.trace("wait next message");
                Object input = source.next();
                if (input != null) {
                    submit(input);
                }
            }
        } catch (InterruptedException e) {
            logger.info("interrupted");
        } catch (Throwable e) {
            logger.error("interrupted", e);
        }
    }