import org.apache.activemq.broker.BrokerFilter; //导入依赖的package包/类
@Override
public Broker installPlugin(final Broker next) throws Exception {

	LOG.info("Installing plugin...");

	return new BrokerFilter(next) {

		@Override
		public void send(final ProducerBrokerExchange producerExchange, final Message messageSend) throws Exception {

			try {
				final String destination = messageSend.getDestination().getPhysicalName();
				final IMessagePublisher<Message, K, V> publisher = messagePublisherProvider.getPublisher(destination);
				if (publisher == null) {
					LOG.debug("No kakfa-publisher found for destination={}. Will forward message.", destination);
					super.send(producerExchange, messageSend);
				} else {
					publisher.publish(messageSend, kafkaTemplate);
				}
			} catch (final Throwable t) {
				LOG.error("Exception occured", t);
				throw t;
			}
		}
	};

}