/*******************************************************************************
 * Copyright (c) 2009, 2014 IBM Corp.
 *
 * All rights reserved. This program and the accompanying materials
 * are made available under the terms of the Eclipse Public License v1.0
 * and Eclipse Distribution License v1.0 which accompany this distribution. 
 *
 * The Eclipse Public License is available at 
 *    http://www.eclipse.org/legal/epl-v10.html
 * and the Eclipse Distribution License is available at 
 *   http://www.eclipse.org/org/documents/edl-v10.php.
 *
 * Contributors:
 *    Dave Locke - initial API and implementation and/or initial documentation
 */

package org.eclipse.paho.sample.mqttv3app;

import java.io.IOException;
import java.sql.Timestamp;

import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.persist.MqttDefaultFilePersistence;

/**
 * A sample application that demonstrates how to use the Paho MQTT v3.1 Client blocking API.
 *
 * It can be run from the command line in one of two modes:
 *  - as a publisher, sending a single message to a topic on the server
 *  - as a subscriber, listening for messages from the server
 *
 *  There are three versions of the sample that implement the same features
 *  but do so using using different programming styles:
 *  <ol>
 *  <li>Sample (this one) which uses the API which blocks until the operation completes</li>
 *  <li>SampleAsyncWait shows how to use the asynchronous API with waiters that block until
 *  an action completes</li>
 *  <li>SampleAsyncCallBack shows how to use the asynchronous API where events are
 *  used to notify the application when an action completes<li>
 *  </ol>
 *
 *  If the application is run with the -h parameter then info is displayed that
 *  describes all of the options / parameters.
 */
public class Sample implements MqttCallback {

	/**
	 * The main entry point of the sample.
	 *
	 * This method handles parsing of the arguments specified on the
	 * command-line before performing the specified action.
	 */
	public static void main(String[] args) {

		// Default settings:
		boolean quietMode 	= false;
		String action 		= "publish";
		String topic 		= "";
		String message 		= "Message from blocking Paho MQTTv3 Java client sample";
		int qos 			= 2;
		String broker 		= "m2m.eclipse.org";
		int port 			= 1883;
		String clientId 	= null;
		String subTopic		= "Sample/#";
		String pubTopic 	= "Sample/Java/v3";
		boolean cleanSession = true;			// Non durable subscriptions
		boolean ssl = false;
		String password = null;
		String userName = null;
		// Parse the arguments -
		for (int i=0; i<args.length; i++) {
			// Check this is a valid argument
			if (args[i].length() == 2 && args[i].startsWith("-")) {
				char arg = args[i].charAt(1);
				// Handle arguments that take no-value
				switch(arg) {
					case ‘h‘: case ‘?‘:	printHelp(); return;
					case ‘q‘: quietMode = true;	continue;
				}

				// Now handle the arguments that take a value and
				// ensure one is specified
				if (i == args.length -1 || args[i+1].charAt(0) == ‘-‘) {
					System.out.println("Missing value for argument: "+args[i]);
					printHelp();
					return;
				}
				switch(arg) {
					case ‘a‘: action = args[++i];                 break;
					case ‘t‘: topic = args[++i];                  break;
					case ‘m‘: message = args[++i];                break;
					case ‘s‘: qos = Integer.parseInt(args[++i]);  break;
					case ‘b‘: broker = args[++i];                 break;
					case ‘p‘: port = Integer.parseInt(args[++i]); break;
					case ‘i‘: clientId = args[++i];				  break;
					case ‘c‘: cleanSession = Boolean.valueOf(args[++i]).booleanValue();  break;
					case ‘k‘: System.getProperties().put("javax.net.ssl.keyStore", args[++i]); break;
					case ‘w‘: System.getProperties().put("javax.net.ssl.keyStorePassword", args[++i]); break;
					case ‘r‘: System.getProperties().put("javax.net.ssl.trustStore", args[++i]); break;
					case ‘v‘: ssl = Boolean.valueOf(args[++i]).booleanValue(); break;
					case ‘u‘: userName = args[++i];               break;
					case ‘z‘: password = args[++i];               break;
					default:
						System.out.println("Unrecognised argument: "+args[i]);
						printHelp();
						return;
				}
			} else {
				System.out.println("Unrecognised argument: "+args[i]);
				printHelp();
				return;
			}
		}

		// Validate the provided arguments
		if (!action.equals("publish") && !action.equals("subscribe")) {
			System.out.println("Invalid action: "+action);
			printHelp();
			return;
		}
		if (qos < 0 || qos > 2) {
			System.out.println("Invalid QoS: "+qos);
			printHelp();
			return;
		}
		if (topic.equals("")) {
			// Set the default topic according to the specified action
			if (action.equals("publish")) {
				topic = pubTopic;
			} else {
				topic = subTopic;
			}
		}

		String protocol = "tcp://";

    if (ssl) {
      protocol = "ssl://";
    }

    String url = protocol + broker + ":" + port;

		if (clientId == null || clientId.equals("")) {
			clientId = "SampleJavaV3_"+action;
		}

		// With a valid set of arguments, the real work of
		// driving the client API can begin
		try {
			// Create an instance of this class
			Sample sampleClient = new Sample(url, clientId, cleanSession, quietMode,userName,password);

			// Perform the requested action
			if (action.equals("publish")) {
				sampleClient.publish(topic,qos,message.getBytes());
			} else if (action.equals("subscribe")) {
				sampleClient.subscribe(topic,qos);
			}
		} catch(MqttException me) {
			// Display full details of any exception that occurs
			System.out.println("reason "+me.getReasonCode());
			System.out.println("msg "+me.getMessage());
			System.out.println("loc "+me.getLocalizedMessage());
			System.out.println("cause "+me.getCause());
			System.out.println("excep "+me);
			me.printStackTrace();
		}
	}

	// Private instance variables
	private MqttClient 			client;
	private String 				brokerUrl;
	private boolean 			quietMode;
	private MqttConnectOptions 	conOpt;
	private boolean 			clean;
	private String password;
	private String userName;

	/**
	 * Constructs an instance of the sample client wrapper
	 * @param brokerUrl the url of the server to connect to
	 * @param clientId the client id to connect with
	 * @param cleanSession clear state at end of connection or not (durable or non-durable subscriptions)
	 * @param quietMode whether debug should be printed to standard out
   * @param userName the username to connect with
   * @param password the password for the user
	 * @throws MqttException
	 */
    public Sample(String brokerUrl, String clientId, boolean cleanSession, boolean quietMode, String userName, String password) throws MqttException {
    	this.brokerUrl = brokerUrl;
    	this.quietMode = quietMode;
    	this.clean 	   = cleanSession;
    	this.password = password;
    	this.userName = userName;
    	//This sample stores in a temporary directory... where messages temporarily
    	// stored until the message has been delivered to the server.
    	//..a real application ought to store them somewhere
    	// where they are not likely to get deleted or tampered with
    	String tmpDir = System.getProperty("java.io.tmpdir");
    	MqttDefaultFilePersistence dataStore = new MqttDefaultFilePersistence(tmpDir);

    	try {
    		// Construct the connection options object that contains connection parameters
    		// such as cleanSession and LWT
	    	conOpt = new MqttConnectOptions();
	    	conOpt.setCleanSession(clean);
	    	if(password != null ) {
	    	  conOpt.setPassword(this.password.toCharArray());
	    	}
	    	if(userName != null) {
	    	  conOpt.setUserName(this.userName);
	    	}

    		// Construct an MQTT blocking mode client
			client = new MqttClient(this.brokerUrl,clientId, dataStore);

			// Set this wrapper as the callback handler
	    	client.setCallback(this);

		} catch (MqttException e) {
			e.printStackTrace();
			log("Unable to set up client: "+e.toString());
			System.exit(1);
		}
    }

    /**
     * Publish / send a message to an MQTT server
     * @param topicName the name of the topic to publish to
     * @param qos the quality of service to delivery the message at (0,1,2)
     * @param payload the set of bytes to send to the MQTT server
     * @throws MqttException
     */
    public void publish(String topicName, int qos, byte[] payload) throws MqttException {

    	// Connect to the MQTT server
    	log("Connecting to "+brokerUrl + " with client ID "+client.getClientId());
    	client.connect(conOpt);
    	log("Connected");

    	String time = new Timestamp(System.currentTimeMillis()).toString();
    	log("Publishing at: "+time+ " to topic \""+topicName+"\" qos "+qos);

    	// Create and configure a message
   		MqttMessage message = new MqttMessage(payload);
    	message.setQos(qos);

    	// Send the message to the server, control is not returned until
    	// it has been delivered to the server meeting the specified
    	// quality of service.
    	client.publish(topicName, message);

    	// Disconnect the client
    	client.disconnect();
    	log("Disconnected");
    }

    /**
     * Subscribe to a topic on an MQTT server
     * Once subscribed this method waits for the messages to arrive from the server
     * that match the subscription. It continues listening for messages until the enter key is
     * pressed.
     * @param topicName to subscribe to (can be wild carded)
     * @param qos the maximum quality of service to receive messages at for this subscription
     * @throws MqttException
     */
    public void subscribe(String topicName, int qos) throws MqttException {

    	// Connect to the MQTT server
    	client.connect(conOpt);
    	log("Connected to "+brokerUrl+" with client ID "+client.getClientId());

    	// Subscribe to the requested topic
    	// The QoS specified is the maximum level that messages will be sent to the client at.
    	// For instance if QoS 1 is specified, any messages originally published at QoS 2 will
    	// be downgraded to 1 when delivering to the client but messages published at 1 and 0
    	// will be received at the same level they were published at.
    	log("Subscribing to topic \""+topicName+"\" qos "+qos);
    	client.subscribe(topicName, qos);

    	// Continue waiting for messages until the Enter is pressed
    	log("Press <Enter> to exit");
		try {
			System.in.read();
		} catch (IOException e) {
			//If we can‘t read we‘ll just exit
		}

		// Disconnect the client from the server
		client.disconnect();
		log("Disconnected");
    }

    /**
     * Utility method to handle logging. If ‘quietMode‘ is set, this method does nothing
     * @param message the message to log
     */
    private void log(String message) {
    	if (!quietMode) {
    		System.out.println(message);
    	}
    }

	/****************************************************************/
	/* Methods to implement the MqttCallback interface              */
	/****************************************************************/

    /**
     * @see MqttCallback#connectionLost(Throwable)
     */
	public void connectionLost(Throwable cause) {
		// Called when the connection to the server has been lost.
		// An application may choose to implement reconnection
		// logic at this point. This sample simply exits.
		log("Connection to " + brokerUrl + " lost!" + cause);
		System.exit(1);
	}

    /**
     * @see MqttCallback#deliveryComplete(IMqttDeliveryToken)
     */
	public void deliveryComplete(IMqttDeliveryToken token) {
		// Called when a message has been delivered to the
		// server. The token passed in here is the same one
		// that was passed to or returned from the original call to publish.
		// This allows applications to perform asynchronous
		// delivery without blocking until delivery completes.
		//
		// This sample demonstrates asynchronous deliver and
		// uses the token.waitForCompletion() call in the main thread which
		// blocks until the delivery has completed.
		// Additionally the deliveryComplete method will be called if
		// the callback is set on the client
		//
		// If the connection to the server breaks before delivery has completed
		// delivery of a message will complete after the client has re-connected.
		// The getPendingTokens method will provide tokens for any messages
		// that are still to be delivered.
	}

    /**
     * @see MqttCallback#messageArrived(String, MqttMessage)
     */
	public void messageArrived(String topic, MqttMessage message) throws MqttException {
		// Called when a message arrives from the server that matches any
		// subscription made by the client
		String time = new Timestamp(System.currentTimeMillis()).toString();
		System.out.println("Time:\t" +time +
                           "  Topic:\t" + topic +
                           "  Message:\t" + new String(message.getPayload()) +
                           "  QoS:\t" + message.getQos());
	}

	/****************************************************************/
	/* End of MqttCallback methods                                  */
	/****************************************************************/

	   static void printHelp() {
	      System.out.println(
	          "Syntax:\n\n" +
	              "    Sample [-h] [-a publish|subscribe] [-t <topic>] [-m <message text>]\n" +
	              "            [-s 0|1|2] -b <hostname|IP address>] [-p <brokerport>] [-i <clientID>]\n\n" +
	              "    -h  Print this help text and quit\n" +
	              "    -q  Quiet mode (default is false)\n" +
	              "    -a  Perform the relevant action (default is publish)\n" +
	              "    -t  Publish/subscribe to <topic> instead of the default\n" +
	              "            (publish: \"Sample/Java/v3\", subscribe: \"Sample/#\")\n" +
	              "    -m  Use <message text> instead of the default\n" +
	              "            (\"Message from MQTTv3 Java client\")\n" +
	              "    -s  Use this QoS instead of the default (2)\n" +
	              "    -b  Use this name/IP address instead of the default (m2m.eclipse.org)\n" +
	              "    -p  Use this port instead of the default (1883)\n\n" +
	              "    -i  Use this client ID instead of SampleJavaV3_<action>\n" +
	              "    -c  Connect to the server with a clean session (default is false)\n" +
	              "     \n\n Security Options \n" +
	              "     -u Username \n" +
	              "     -z Password \n" +
	              "     \n\n SSL Options \n" +
	              "    -v  SSL enabled; true - (default is false) " +
	              "    -k  Use this JKS format key store to verify the client\n" +
	              "    -w  Passpharse to verify certificates in the keys store\n" +
	              "    -r  Use this JKS format keystore to verify the server\n" +
	              " If javax.net.ssl properties have been set only the -v flag needs to be set\n" +
	              "Delimit strings containing spaces with \"\"\n\n" +
	              "Publishers transmit a single message then disconnect from the server.\n" +
	              "Subscribers remain connected to the server and receive appropriate\n" +
	              "messages until <enter> is pressed.\n\n"
	          );
    }

}

client = new MqttClient(this.brokerUrl,clientId, dataStore);

client.setCallback(this); //注意这里的this实现了MqttCallback接口

        client.connect(conOpt);        
        MqttMessage message = new MqttMessage(payload);
        message.setQos(qos);
        client.publish(topicName, message);

    	client.connect(conOpt);
    	client.subscribe(topicName, qos);

	<dependencies>
		<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-starter-integration</artifactId>
		</dependency>
		<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-integration-stream</artifactId>
		</dependency>
		<dependency>
			<groupId>org.springframework.integration</groupId>
			<artifactId>spring-integration-mqtt</artifactId>
		</dependency>
	</dependencies>

@SpringBootApplication
public class MqttJavaApplication {

    public static void main(String[] args) {
        new SpringApplicationBuilder(MqttJavaApplication.class)
                .web(false)
                .run(args);
    }

    @Bean
    public MessageChannel mqttInputChannel() {
        return new DirectChannel();
    }

    @Bean
    public MessageProducer inbound() {
        MqttPahoMessageDrivenChannelAdapter adapter =
                new MqttPahoMessageDrivenChannelAdapter("tcp://localhost:1883", "testClient",
                                                 "topic1", "topic2");
        adapter.setCompletionTimeout(5000);
        adapter.setConverter(new DefaultPahoMessageConverter());
        adapter.setQos(1);
        adapter.setOutputChannel(mqttInputChannel());
        return adapter;
    }

    @Bean
    @ServiceActivator(inputChannel = "mqttInputChannel")
    public MessageHandler handler() {
        return new MessageHandler() {

            @Override
            public void handleMessage(Message<?> message) throws MessagingException {
                System.out.println(message.getPayload());
            }

        };
    }

}

@Configuration
@ConditionalOnClass(EnableIntegration.class)
@EnableConfigurationProperties(IntegrationProperties.class)
@AutoConfigureAfter(JmxAutoConfiguration.class)
public class IntegrationAutoConfiguration {

	/**
	 * Basic Spring Integration configuration.
	 */
	@Configuration
	@EnableIntegration
	protected static class IntegrationConfiguration {

	}

@Target(ElementType.TYPE)
@Retention(RetentionPolicy.RUNTIME)
@Documented
@Import(IntegrationRegistrar.class)
public @interface EnableIntegration {

}

	public void registerBeanDefinitions(@Nullable AnnotationMetadata importingClassMetadata,
			BeanDefinitionRegistry registry) {

		registerImplicitChannelCreator(registry);
		registerIntegrationConfigurationBeanFactoryPostProcessor(registry);
		registerIntegrationEvaluationContext(registry);
		registerIntegrationProperties(registry);
		registerHeaderChannelRegistry(registry);
		registerGlobalChannelInterceptorProcessor(registry);
		registerBuiltInBeans(registry);
		registerDefaultConfiguringBeanFactoryPostProcessor(registry);
		registerDefaultDatatypeChannelMessageConverter(registry);
		registerArgumentResolverMessageConverter(registry);
		registerArgumentResolvers(registry);
		registerListCapableArgumentResolvers(registry);
		if (importingClassMetadata != null) {
			registerMessagingAnnotationPostProcessors(importingClassMetadata, registry);
		}
		registerMessageBuilderFactory(registry);
		IntegrationConfigUtils.registerRoleControllerDefinitionIfNecessary(registry);
	}

	private void registerMessagingAnnotationPostProcessors(AnnotationMetadata meta, BeanDefinitionRegistry registry) {
		if (!registry.containsBeanDefinition(IntegrationContextUtils.MESSAGING_ANNOTATION_POSTPROCESSOR_NAME)) {
			BeanDefinitionBuilder builder =
					BeanDefinitionBuilder.genericBeanDefinition(MessagingAnnotationPostProcessor.class)
							.setRole(BeanDefinition.ROLE_INFRASTRUCTURE);

			registry.registerBeanDefinition(IntegrationContextUtils.MESSAGING_ANNOTATION_POSTPROCESSOR_NAME,
					builder.getBeanDefinition());
		}

		new PublisherRegistrar().registerBeanDefinitions(meta, registry);
	}

public class MessagingAnnotationPostProcessor implements BeanPostProcessor, BeanFactoryAware, InitializingBean {
        
        。。。。。。

	@Override
	public void afterPropertiesSet() {
		Assert.notNull(this.beanFactory, "BeanFactory must not be null");
		this.postProcessors.put(Filter.class, new FilterAnnotationPostProcessor(this.beanFactory));
		this.postProcessors.put(Router.class, new RouterAnnotationPostProcessor(this.beanFactory));
		this.postProcessors.put(Transformer.class, new TransformerAnnotationPostProcessor(this.beanFactory));
		this.postProcessors.put(ServiceActivator.class, new ServiceActivatorAnnotationPostProcessor(this.beanFactory));
		this.postProcessors.put(Splitter.class, new SplitterAnnotationPostProcessor(this.beanFactory));
		this.postProcessors.put(Aggregator.class, new AggregatorAnnotationPostProcessor(this.beanFactory));
		this.postProcessors.put(InboundChannelAdapter.class,
				new InboundChannelAdapterAnnotationPostProcessor(this.beanFactory));
		this.postProcessors.put(BridgeFrom.class, new BridgeFromAnnotationPostProcessor(this.beanFactory));
		this.postProcessors.put(BridgeTo.class, new BridgeToAnnotationPostProcessor(this.beanFactory));
		Map<Class<? extends Annotation>, MethodAnnotationPostProcessor<?>> customPostProcessors =
				setupCustomPostProcessors();
		if (!CollectionUtils.isEmpty(customPostProcessors)) {
			this.postProcessors.putAll(customPostProcessors);
		}
	}

	@Override
	public Object postProcessAfterInitialization(final Object bean, final String beanName) throws BeansException {
		Assert.notNull(this.beanFactory, "BeanFactory must not be null");

		Class<?> beanClass = AopUtils.getTargetClass(bean);

		// the set will hold records of prior class scans and indicate if no messaging annotations were found
		if (this.noAnnotationsCache.contains(beanClass)) {
			return bean;
		}

		ReflectionUtils.doWithMethods(beanClass, method -> {
			Map<Class<? extends Annotation>, List<Annotation>> annotationChains = new HashMap<>();
			for (Class<? extends Annotation> annotationType :
					this.postProcessors.keySet()) {
				if (AnnotatedElementUtils.isAnnotated(method, annotationType.getName())) {
					List<Annotation> annotationChain = getAnnotationChain(method, annotationType);
					if (annotationChain.size() > 0) {
						annotationChains.put(annotationType, annotationChain);
					}
				}
			}
			if (StringUtils.hasText(MessagingAnnotationUtils.endpointIdValue(method))
					&& annotationChains.keySet().size() > 1) {
				throw new IllegalStateException("@EndpointId on " + method.toGenericString()
						+ " can only have one EIP annotation, found: " + annotationChains.keySet().size());
			}
			for (Entry<Class<? extends Annotation>, List<Annotation>> entry : annotationChains.entrySet()) {
				Class<? extends Annotation> annotationType = entry.getKey();
				List<Annotation> annotations = entry.getValue();
				processAnnotationTypeOnMethod(bean, beanName, method, annotationType, annotations);
			}

			if (annotationChains.size() == 0) {
				this.noAnnotationsCache.add(beanClass);
			}
		}, ReflectionUtils.USER_DECLARED_METHODS);

		return bean;
	}

	protected void processAnnotationTypeOnMethod(Object bean, String beanName, Method method,
			Class<? extends Annotation> annotationType, List<Annotation> annotations) {
		MethodAnnotationPostProcessor<?> postProcessor =
				MessagingAnnotationPostProcessor.this.postProcessors.get(annotationType);
		if (postProcessor != null && postProcessor.shouldCreateEndpoint(method, annotations)) {
			Method targetMethod = method;
			if (AopUtils.isJdkDynamicProxy(bean)) {
				try {
					targetMethod = bean.getClass().getMethod(method.getName(), method.getParameterTypes());
				}
				catch (NoSuchMethodException e) {
					throw new IllegalArgumentException("Service methods must be extracted to the service "
							+ "interface for JdkDynamicProxy. The affected bean is: ‘" + beanName + "‘ "
							+ "and its method: ‘" + method + "‘", e);
				}
			}
			Object result = postProcessor.postProcess(bean, beanName, targetMethod, annotations);
			if (result != null && result instanceof AbstractEndpoint) {
				AbstractEndpoint endpoint = (AbstractEndpoint) result;
				String autoStartup = MessagingAnnotationUtils.resolveAttribute(annotations, "autoStartup",
						String.class);
				if (StringUtils.hasText(autoStartup)) {
					autoStartup = getBeanFactory().resolveEmbeddedValue(autoStartup);
					if (StringUtils.hasText(autoStartup)) {
						endpoint.setAutoStartup(Boolean.parseBoolean(autoStartup));
					}
				}

				String phase = MessagingAnnotationUtils.resolveAttribute(annotations, "phase", String.class);
				if (StringUtils.hasText(phase)) {
					phase = getBeanFactory().resolveEmbeddedValue(phase);
					if (StringUtils.hasText(phase)) {
						endpoint.setPhase(Integer.parseInt(phase));
					}
				}

				Role role = AnnotationUtils.findAnnotation(method, Role.class);
				if (role != null) {
					endpoint.setRole(role.value());
				}

				String endpointBeanName = generateBeanName(beanName, method, annotationType);
				endpoint.setBeanName(endpointBeanName);
				getBeanFactory().registerSingleton(endpointBeanName, endpoint);
				getBeanFactory().initializeBean(endpoint, endpointBeanName);
			}
		}
	}

	public Object postProcess(Object bean, String beanName, Method method, List<Annotation> annotations) {
        。。。。。。
        MessageHandler handler = createHandler(bean, method, annotations);
        。。。。。。
        AbstractEndpoint endpoint = createEndpoint(handler, method, annotations);
		if (endpoint != null) {
			return endpoint;
		}
		return handler;
	}

	protected MessageHandler createHandler(Object bean, Method method, List<Annotation> annotations) {
		AbstractReplyProducingMessageHandler serviceActivator;
		if (AnnotatedElementUtils.isAnnotated(method, Bean.class.getName())) {
			final Object target = this.resolveTargetBeanFromMethodWithBeanAnnotation(method);
			serviceActivator = this.extractTypeIfPossible(target, AbstractReplyProducingMessageHandler.class);
			if (serviceActivator == null) {
				if (target instanceof MessageHandler) {
					/*
					 * Return a reply-producing message handler so that we still get ‘produced no reply‘ messages
					 * and the super class will inject the advice chain to advise the handler method if needed.
					 */
					return new ReplyProducingMessageHandlerWrapper((MessageHandler) target);
				}

	protected AbstractEndpoint createEndpoint(MessageHandler handler, Method method, List<Annotation> annotations) {
		AbstractEndpoint endpoint = null;
		String inputChannelName = MessagingAnnotationUtils.resolveAttribute(annotations, getInputChannelAttribute(),
				String.class);
		if (StringUtils.hasText(inputChannelName)) {
			MessageChannel inputChannel;
			try {
				inputChannel = this.channelResolver.resolveDestination(inputChannelName);
			}
			catch (DestinationResolutionException e) {
				if (e.getCause() instanceof NoSuchBeanDefinitionException) {
					inputChannel = new DirectChannel();
					this.beanFactory.registerSingleton(inputChannelName, inputChannel);
					inputChannel = (MessageChannel) this.beanFactory.initializeBean(inputChannel, inputChannelName);
				}
				else {
					throw e;
				}
			}
			Assert.notNull(inputChannel, "failed to resolve inputChannel ‘" + inputChannelName + "‘");

			endpoint = doCreateEndpoint(handler, inputChannel, annotations);
		}
		return endpoint;
	}

	protected AbstractEndpoint doCreateEndpoint(MessageHandler handler, MessageChannel inputChannel,
			List<Annotation> annotations) {
		AbstractEndpoint endpoint;
		if (inputChannel instanceof PollableChannel) {
			PollingConsumer pollingConsumer = new PollingConsumer((PollableChannel) inputChannel, handler);
			configurePollingEndpoint(pollingConsumer, annotations);
			endpoint = pollingConsumer;
		}
		else {
			Poller[] pollers = MessagingAnnotationUtils.resolveAttribute(annotations, "poller", Poller[].class);
			Assert.state(ObjectUtils.isEmpty(pollers), "A ‘@Poller‘ should not be specified for Annotation-based " +
					"endpoint, since ‘" + inputChannel + "‘ is a SubscribableChannel (not pollable).");
			if (inputChannel instanceof Publisher) {
				endpoint = new ReactiveStreamsConsumer(inputChannel, handler);
			}
			else {
				endpoint = new EventDrivenConsumer((SubscribableChannel) inputChannel, handler);
			}
		}
		return endpoint;
	}

public class EventDrivenConsumer extends AbstractEndpoint implements IntegrationConsumer {

	private final SubscribableChannel inputChannel;

	private final MessageHandler handler;


	public EventDrivenConsumer(SubscribableChannel inputChannel, MessageHandler handler) {
		Assert.notNull(inputChannel, "inputChannel must not be null");
		Assert.notNull(handler, "handler must not be null");
		this.inputChannel = inputChannel;
		this.handler = handler;
		this.setPhase(Integer.MIN_VALUE);
	}

	@Override
	public MessageChannel getInputChannel() {
		return this.inputChannel;
	}

	@Override
	public MessageChannel getOutputChannel() {
		if (this.handler instanceof MessageProducer) {
			return ((MessageProducer) this.handler).getOutputChannel();
		}
		else if (this.handler instanceof MessageRouter) {
			return ((MessageRouter) this.handler).getDefaultOutputChannel();
		}
		else {
			return null;
		}
	}

	@Override
	public MessageHandler getHandler() {
		return this.handler;
	}

	@Override
	protected void doStart() {
		this.logComponentSubscriptionEvent(true);
		this.inputChannel.subscribe(this.handler);
		if (this.handler instanceof Lifecycle) {
			((Lifecycle) this.handler).start();
		}
	}

	@Override
	protected void doStop() {
		this.logComponentSubscriptionEvent(false);
		this.inputChannel.unsubscribe(this.handler);
		if (this.handler instanceof Lifecycle) {
			((Lifecycle) this.handler).stop();
		}
	}

	private void logComponentSubscriptionEvent(boolean add) {
		if (this.handler instanceof NamedComponent && this.inputChannel instanceof NamedComponent) {
			String channelName = ((NamedComponent) this.inputChannel).getComponentName();
			String componentType = ((NamedComponent) this.handler).getComponentType();
			componentType = StringUtils.hasText(componentType) ? componentType : "";
			String componentName = getComponentName();
			componentName = (StringUtils.hasText(componentName) && componentName.contains("#")) ? "" : ":" + componentName;
			StringBuffer buffer = new StringBuffer();
			buffer.append("{" + componentType + componentName + "} as a subscriber to the ‘" + channelName + "‘ channel");
			if (add) {
				buffer.insert(0, "Adding ");
			}
			else {
				buffer.insert(0, "Removing ");
			}
			logger.info(buffer.toString());
		}
	}
}