import org.apache.camel.Consume; //导入依赖的package包/类
@Consume(uri = "activemq:queue:inbox?concurrentConsumers=10")
@RecipientList
public String listen(Exchange exchange) {
    topic.send(exchange);

    String type = exchange.getIn().getHeader("type", String.class);
    return "direct:" + type;
}
 

import org.apache.camel.Consume; //导入依赖的package包/类
private boolean shouldDeployDefaultCamelContext(Set<Bean<?>> beans) {
    return beans.stream()
        // Is there a Camel bean with the @Default qualifier?
        // Excluding internal components...
        .filter(bean -> !bean.getBeanClass().getPackage().equals(getClass().getPackage()))
        .filter(hasType(CamelContextAware.class).or(hasType(Component.class))
            .or(hasType(RouteContainer.class).or(hasType(RoutesBuilder.class))))
        .map(Bean::getQualifiers)
        .flatMap(Set::stream)
        .filter(isEqual(DEFAULT))
        .findAny()
        .isPresent()
        // Or a bean with Camel annotations?
        || concat(camelBeans.stream().map(AnnotatedType::getFields),
                  camelBeans.stream().map(AnnotatedType::getMethods))
        .flatMap(Set::stream)
        .map(Annotated::getAnnotations)
        .flatMap(Set::stream)
        .filter(isAnnotationType(Consume.class).and(a -> ((Consume) a).context().isEmpty())
            .or(isAnnotationType(BeanInject.class).and(a -> ((BeanInject) a).context().isEmpty()))
            .or(isAnnotationType(EndpointInject.class).and(a -> ((EndpointInject) a).context().isEmpty()))
            .or(isAnnotationType(Produce.class).and(a -> ((Produce) a).context().isEmpty()))
            .or(isAnnotationType(PropertyInject.class).and(a -> ((PropertyInject) a).context().isEmpty())))
        .findAny()
        .isPresent()
        // Or an injection point for Camel primitives?
        || beans.stream()
        // Excluding internal components...
        .filter(bean -> !bean.getBeanClass().getPackage().equals(getClass().getPackage()))
        .map(Bean::getInjectionPoints)
        .flatMap(Set::stream)
        .filter(ip -> getRawType(ip.getType()).getName().startsWith("org.apache.camel"))
        .map(InjectionPoint::getQualifiers)
        .flatMap(Set::stream)
        .filter(isAnnotationType(Uri.class).or(isAnnotationType(Mock.class)).or(isEqual(DEFAULT)))
        .findAny()
        .isPresent();
}
 

import org.apache.camel.Consume; //导入依赖的package包/类
@Consume(uri = "activemq:Test.BindingQueue")
public void myMethod(@Headers Map<?, ?> headers, String body) {
    this.headers = headers;
    this.body = body;

    // now lets notify we've completed
    producer.sendBody("Completed");
}
 

import org.apache.camel.Consume; //导入依赖的package包/类
private void processAnnotatedType(@Observes ProcessAnnotatedType<?> pat) {
    if (pat.getAnnotatedType().isAnnotationPresent(Vetoed.class)) {
        pat.veto();
    }
    if (hasAnnotation(pat.getAnnotatedType(), Converter.class)) {
        converters.add(pat.getAnnotatedType().getJavaClass());
    }
    if (hasAnnotation(pat.getAnnotatedType(), BeanInject.class, Consume.class, EndpointInject.class, Produce.class, PropertyInject.class)) {
        camelBeans.add(pat.getAnnotatedType());
    }
    if (hasAnnotation(pat.getAnnotatedType(), Consume.class)) {
        eagerBeans.add(pat.getAnnotatedType());
    }
    if (hasAnnotation(pat.getAnnotatedType(), ImportResource.class)) {
        resources.add(pat.getAnnotatedType().getAnnotation(ImportResource.class));
    }
}
 

import org.apache.camel.Consume; //导入依赖的package包/类
@Consume(uri = "seda:book")
public void handleTitle(String title) {
    Transactional tx = this.getClass().getAnnotation(Transactional.class);
    if (tx == null) {
        throw new IllegalStateException("Spring annotation-driven should have instrumented this class as @Transactional");
    }
    if (!"NEVER".equals(tx.propagation().name())) {
        throw new IllegalStateException("Should be NEVER propagation");
    }
    if (!tx.readOnly()) {
        throw new IllegalStateException("Should be read only");
    }

    if (!title.contains("in Action")) {
        throw new IllegalArgumentException("Not a book title we like");
    }

    producer.sendBody(title);
}
 

import org.apache.camel.Consume; //导入依赖的package包/类
@Consume(uri = "activemq:personnel.records")
@RecipientList
public String[] route(@XPath("/person/city/text()") String city) {
    if (city.equals("London")) {
        LOG.info("Person is from EMEA region");
        return new String[] {"file:target/messages/emea/hr_pickup", 
                             "file:target/messages/emea/finance_pickup"};
    } else {
        LOG.info("Person is from AMER region");
        return new String[] {"file:target/messages/amer/hr_pickup",
                             "file:target/messages/amer/finance_pickup"};
    }
}
 

import org.apache.camel.Consume; //导入依赖的package包/类
public void consumerInjection(Method method, Object bean, String beanName) {
    Consume consume = method.getAnnotation(Consume.class);
    if (consume != null && matchContext(consume.context())) {
        LOG.debug("Creating a consumer for: " + consume);
        subscribeMethod(method, bean, beanName, consume.uri(), consume.ref(), consume.property());
    }
}
 

import org.apache.camel.Consume; //导入依赖的package包/类
@Consume(uri = "activemq:queue:foo")
public void doSomething(@Header("JMSReplyTo") Destination jmsReplyTo, @Body String body) throws Exception {
    assertEquals("Hello World", body);

    String endpointName = "activemq:" + jmsReplyTo.toString();
    endpointName = endpointName.replaceAll("//", ":");

    tempName = endpointName;
    latch.countDown();

    template.sendBody(tempName, "Bye World");
}
 

import org.apache.camel.Consume; //导入依赖的package包/类
@Consume(uri = "activemq:queue:foo")
public void doSomething(@Header("JMSReplyTo") Destination jmsReplyTo, @Body String body) throws Exception {
    assertEquals("Hello World", body);

    String endpointName = "activemq:" + jmsReplyTo.toString();
    template.sendBody(endpointName, "Bye World");
}
 

import org.apache.camel.Consume; //导入依赖的package包/类
@Consume(uri = "direct:foo", context = "camel-2")
@RecipientList(context = "camel-2")
public String[] doSomething(String body) {
    LOG.info("Received body: " + body);

    return new String[]{"mock:foo", "mock:result"};
}
 

import org.apache.camel.Consume; //导入依赖的package包/类
@Consume(uri = "direct:start")
public void doSomething(String body) {
    ObjectHelper.notNull(destination, "destination");

    LOG.info("Received body: " + body);
    destination.sendBody(body);
}
 

import org.apache.camel.Consume; //导入依赖的package包/类
@Consume(uri = "direct:start", context = "camel-1")
public void doSomething(String body) {
    ObjectHelper.notNull(destination, "destination");

    LOG.info("Received body: " + body);
    destination.sendBody(body);
}
 

import org.apache.camel.Consume; //导入依赖的package包/类
@Consume(uri = "direct:start", context = "camel-2")
public void doSomething(String body) {
    ObjectHelper.notNull(destination, "destination");

    LOG.info("Received body: " + body);
    destination.sendBody(body);
}
 

import org.apache.camel.Consume; //导入依赖的package包/类
private boolean shouldDeployDefaultCamelContext(BeanManager manager, Set<SyntheticBean<?>> beans) {
    // TODO: find a way to 'pre-filter' by refining the bean types passed to the bean manager
    return concat(manager.getBeans(Object.class, ANY).stream(), beans.stream())
        // Is there a Camel bean with the @Default qualifier?
        // Excluding internal components...
        .filter(bean -> !bean.getBeanClass().getPackage().equals(getClass().getPackage()))
        .filter(hasType(CamelContextAware.class).or(hasType(Component.class))
            .or(hasType(RouteContainer.class).or(hasType(RoutesBuilder.class))))
        .map(Bean::getQualifiers)
        .flatMap(Set::stream)
        .anyMatch(isEqual(DEFAULT))
        // Or a bean with Camel annotations?
        || concat(camelBeans.stream().map(AnnotatedType::getFields),
                  camelBeans.stream().map(AnnotatedType::getMethods))
        .flatMap(Set::stream)
        .map(Annotated::getAnnotations)
        .flatMap(Set::stream)
        .anyMatch(isAnnotationType(Consume.class).and(a -> ((Consume) a).context().isEmpty())
            .or(isAnnotationType(BeanInject.class).and(a -> ((BeanInject) a).context().isEmpty()))
            .or(isAnnotationType(EndpointInject.class).and(a -> ((EndpointInject) a).context().isEmpty()))
            .or(isAnnotationType(Produce.class).and(a -> ((Produce) a).context().isEmpty()))
            .or(isAnnotationType(PropertyInject.class).and(a -> ((PropertyInject) a).context().isEmpty())))
        // Or an injection point for Camel primitives?
        || concat(manager.getBeans(Object.class, ANY).stream(), beans.stream())
        // Excluding internal components...
        .filter(bean -> !bean.getBeanClass().getPackage().equals(getClass().getPackage()))
        .map(Bean::getInjectionPoints)
        .flatMap(Set::stream)
        .filter(ip -> getRawType(ip.getType()).getName().startsWith("org.apache.camel"))
        .map(InjectionPoint::getQualifiers)
        .flatMap(Set::stream)
        .anyMatch(isAnnotationType(Uri.class).or(isEqual(DEFAULT)));
}
 

import org.apache.camel.Consume; //导入依赖的package包/类
private void addConsumeAnnotation(CtMethod ctMethod, String uri) {
	MethodInfo methodInfo = ctMethod.getMethodInfo();
	ConstPool constPool = methodInfo.getConstPool();

	Annotation consume = new Annotation(Consume.class.getName(), constPool);
	StringMemberValue valueVal = new StringMemberValue(constPool);
	valueVal.setValue(uri);
	consume.addMemberValue("uri", valueVal);

	AnnotationsAttribute attr = new AnnotationsAttribute(constPool, AnnotationsAttribute.visibleTag);
	attr.addAnnotation(consume);
	methodInfo.addAttribute(attr);
}
 

import org.apache.camel.Consume; //导入依赖的package包/类
@Consume(uri = "direct:start")
@RoutingSlip(delimiter = ",")
public List<String> routeMe(String body, @Headers Map<String, Object> headers) {
    ArrayList<String> results = new ArrayList<String>();

    Object slip = headers.get("myRoutingSlipHeader");
    if (slip != null) {
        String[] uris = slip.toString().split(",");
        Collections.addAll(results, uris);
    }

    results.add("mock:oneMore");

    return results;
}
 

import org.apache.camel.Consume; //导入依赖的package包/类
/**
 * Returns the next endpoint to route a message to or null to finish routing.
 * This implementation leverages Camel's
 * <a href="http://camel.apache.org/bean-integration.html">Bean injection</a>
 * to pass parts of the Camel Exchange to the method for processing. This can
 * help the code be easier to maintain as it does not need the extra boilerplate
 * code for extracting the relative data from the Exchange.
 * <p></p>
 * This implementation stores an int property with the message exchange that is
 * used to drive the routing behavior. This method will be called from multiple
 * threads, one per message, so storing message specific state as a property is
 * a good strategy.
 *
 * @param body       the IN message converted to a String using Camel Bean injection
 * @param properties the properties map associated with the Camel Exchange
 * @return next endpoint uri(s) to route to or <tt>null</tt> to finish routing
 */
@Consume(uri = "direct:start")
@DynamicRouter(delimiter = ",")
public String routeMe(String body, @ExchangeProperties Map<String, Object> properties) {
    LOG.info("Exchange.SLIP_ENDPOINT = {}, invoked = {}",
        properties.get(Exchange.SLIP_ENDPOINT), properties.get(PROPERTY_NAME_INVOKED));

    // Store a property with the message exchange that will drive the routing
    // decisions of this Dynamic Router implementation.
    int invoked = 0;
    Object current = properties.get(PROPERTY_NAME_INVOKED); // property will be null on first call
    if (current != null) {
        invoked = Integer.valueOf(current.toString());
    }
    invoked++;
    properties.put(PROPERTY_NAME_INVOKED, invoked);

    if (invoked == 1) {
        return "mock:a";
    } else if (invoked == 2) {
        return "mock:b,mock:c";
    } else if (invoked == 3) {
        return "direct:other";
    } else if (invoked == 4) {
        return "mock:result";
    }

    // no more, so return null
    return null;
}
 

import org.apache.camel.Consume; //导入依赖的package包/类
@Consume(uri = "jms:queue:send")
public void send(@Body byte[] vo, @Header("concentratorId") String id) {
	
	Channel session = concentratorOnLine.getIoSession(id);
	if(session == null){
		System.out.println("集中器会话中断,无法发送");
		return;
	}
	session.write(vo);
	
}
 

import org.apache.camel.Consume; //导入依赖的package包/类
@Consume(uri = "jms:queue:com.ymsino.esb.domain?concurrentConsumers=1&maxConcurrentConsumers=30")
public Serializable doManagerExecute(@Body Object vo, @Header("method") String method, @Header("beanName") String beanName) throws IllegalArgumentException, IllegalAccessException, InvocationTargetException {
	if(!beanName.endsWith("Manager"))
		return null;
	
	Object invokeBean = applicationContext.getBean(beanName);
   	Method[] allMethods = invokeBean.getClass().getMethods();
   	for(Method itemMethod : allMethods){
   		if(itemMethod.getName().equals(method)){
   			Serializable resultObj = (Serializable) itemMethod.invoke(invokeBean, vo);
   	    	return resultObj;
   		}
   	}
   	return null;
}
 

import org.apache.camel.Consume; //导入依赖的package包/类
@Consume(uri = "file:src/data?noop=true")
public void onFileSendToQueue(String body, @Header("CamelFileName") String name) {
    LOG.info("Incoming file: {}", name);
    producer.sendBody(body);
}