if (exchange instanceof InOnly) { exchange.setStatus(ExchangeStatus.DONE);
if (t instanceof Exception) { exchange.setError((Exception) t);
    protected void pollDeliveryChannel() {        synchronized (polling) {            polling.set(true);            polling.notify();        }        while (running.get()) {            try {                final MessageExchange exchange = channel.accept(1000L);                if (exchange != null) {                    final Transaction tx = (Transaction) exchange.getProperty(MessageExchange.JTA_TRANSACTION_PROPERTY_NAME);                    if (tx != null) {                        if (transactionManager == null) {                            throw new IllegalStateException("Exchange is enlisted in a transaction, but no transaction manager is available");                        }                        transactionManager.suspend();                    }                    workManager.scheduleWork(new Work() {                        public void release() {                        }                        public void run() {                            try {                                if (tx != null) {                                    transactionManager.resume(tx);                                }                                processExchange(exchange);                            } catch (Throwable t) {                                logger.error("Error processing exchange " + exchange, t);                                // Set an error on message                                try {                                    if (exchange instanceof InOnly) {                                        exchange.setStatus(ExchangeStatus.DONE);                                    } else {                                        if (t instanceof Exception) {                                            exchange.setError((Exception) t);                                        } else {                                            exchange.setError(new Exception("Throwable", t));                                        }                                    }                                    channel.send(exchange);                                } catch (Exception inner) {                                    logger.error("Error setting exchange status to ERROR", inner);                                }                            } finally {                                // Check transaction status                                if (tx != null) {                                    int status = Status.STATUS_NO_TRANSACTION;                                    try {                                        status = transactionManager.getStatus();                                    } catch (Throwable t) {                                        logger.error("Error checking transaction status.", t);                                    }                                    if (status != Status.STATUS_NO_TRANSACTION) {                                        logger.error("Transaction is still active after exchange processing. Trying to rollback transaction.");                                        try {                                            transactionManager.rollback();                                        } catch (Throwable t) {                                            logger.error("Error trying to rollback transaction.", t);                                        }                                    }                                }                            }                        }                    });                }            } catch (Throwable t) {                if (running.get() == false) {                    // Should have been interrupted, discard the throwable                    if (logger.isDebugEnabled()) {                        logger.debug("Polling thread will stop");                    }                } else {                    logger.error("Error polling delivery channel", t);                }            }        }        synchronized (polling) {            polling.set(false);            polling.notify();        }    }