Java 类org.apache.camel.component.bean.BeanInvocation 实例源码

/**
 * Marshal inputObjects to SOAP xml. If the exchange or message has an
 * EXCEPTION_CAUGTH property or header then instead of the object the
 * exception is marshaled.
 * 
 * To determine the name of the top level xml elements the elementNameStrategy
 * is used.
 * @throws IOException,SAXException 
 */
public void marshal(Exchange exchange, Object inputObject, OutputStream stream) throws IOException, SAXException {
    checkElementNameStrategy(exchange);

    String soapAction = getSoapActionFromExchange(exchange);
    if (soapAction == null && inputObject instanceof BeanInvocation) {
        BeanInvocation beanInvocation = (BeanInvocation) inputObject;
        WebMethod webMethod = beanInvocation.getMethod().getAnnotation(WebMethod.class);
        if (webMethod != null && webMethod.action() != null) {
            soapAction = webMethod.action();
        }
    }

    Object envelope = adapter.doMarshal(exchange, inputObject, stream, soapAction);

    // and continue in super
    super.marshal(exchange, envelope, stream);
}

@Override
public void configure() throws Exception {
    errorHandler(new LoggingErrorHandlerBuilder(log));
    from("direct:" + LumenChannel.PERSISTENCE_FACT.key())
            .process(exchange -> {
                final BeanInvocation invocation = exchange.getIn().getBody(BeanInvocation.class);
                final FactRequest factRequest = new FactRequest();
                factRequest.setOperation(FactServiceOperation.valueOf(invocation.getMethod().getName()));
                log.info("Invoking {} {}", invocation.getMethod(), invocation.getArgs());
                for (int i = 0; i < invocation.getArgs().length; i++) {
                    final String simpleExpr = invocation.getMethod().getParameters()[i].getAnnotation(Simple.class).value();
                    final String paramName = StringUtils.substringAfter(simpleExpr, "body.");
                    PropertyUtils.setProperty(factRequest, paramName, invocation.getArgs()[i]);
                }
                exchange.getIn().setBody(factRequest);
                exchange.setPattern(ExchangePattern.InOut);
                //exchange.getIn().setHeader(RabbitMQConstants.REPLY_TO, );
            })
            .bean(toJson)
            .to("rabbitmq://localhost/amq.topic?connectionFactory=#amqpConnFactory&exchangeType=topic&autoDelete=false&skipQueueDeclare=true&routingKey=" +
                    LumenChannel.PERSISTENCE_FACT.key())
            .to("log:REPLY." + LumenChannel.PERSISTENCE_FACT.key());
}

@SuppressWarnings("unchecked")
@Override
public <T> T convertTo(Class<T> toType, Exchange exchange, Object value) {

    // should not try to convert Message
    if (Message.class.isAssignableFrom(value.getClass())) {
        return (T) Void.TYPE;
    }

    // should not try to convert future
    if (Future.class.isAssignableFrom(value.getClass())) {
        return (T) Void.TYPE;
    }

    // should not try to convert bean invocations
    if (BeanInvocation.class.isAssignableFrom(value.getClass())) {
        return (T) Void.TYPE;
    }

    // should not try to convert files
    if (WrappedFile.class.isAssignableFrom(value.getClass())) {
        return (T) Void.TYPE;
    }

    if (toType.equals(String.class)) {
        return (T) value.toString();
    }
    return null;
}

/**
 * Returns the expression for the exchanges inbound message body converted
 * to the given type
 *
 * @param type the type
 * @param nullBodyAllowed whether null bodies is allowed and if so a null is returned,
 *                        otherwise an exception is thrown
 */
public static <T> Expression mandatoryBodyExpression(final Class<T> type, final boolean nullBodyAllowed) {
    return new ExpressionAdapter() {
        public Object evaluate(Exchange exchange) {
            if (nullBodyAllowed) {
                if (exchange.getIn().getBody() == null) {
                    return null;
                }

                // if its a bean invocation then if it has no arguments then it should be threaded as null body allowed
                if (exchange.getIn().getBody() instanceof BeanInvocation) {
                    // BeanInvocation would be stored directly as the message body
                    // do not force any type conversion attempts as it would just be unnecessary and cost a bit performance
                    // so a regular instanceof check is sufficient
                    BeanInvocation bi = (BeanInvocation) exchange.getIn().getBody();
                    if (bi.getArgs() == null || bi.getArgs().length == 0 || bi.getArgs()[0] == null) {
                        return null;
                    }
                }
            }

            try {
                return exchange.getIn().getMandatoryBody(type);
            } catch (InvalidPayloadException e) {
                throw ObjectHelper.wrapCamelExecutionException(exchange, e);
            }
        }

        @Override
        public String toString() {
            return "mandatoryBodyAs[" + type.getName() + "]";
        }
    };
}

public <T> T convertTo(Class<T> type, Exchange exchange, Object value) {
    if (BeanInvocation.class.isAssignableFrom(type) || Processor.class.isAssignableFrom(type)) {
        // JAXB cannot convert to a BeanInvocation / Processor, so we need to indicate this
        // to avoid Camel trying to do this when using beans with JAXB payloads
        return null;
    }

    try {
        if (isJaxbType(type)) {
            return unmarshall(type, exchange, value);
        }
        if (value != null && isNotStreamCacheType(type)) {
            if (hasXmlRootElement(value.getClass())) {
                return marshall(type, exchange, value, null);
            }
            if (isObjectFactory()) {
                CamelContext context = exchange != null ? exchange.getContext() : camelContext;
                Method objectFactoryMethod = JaxbHelper.getJaxbElementFactoryMethod(context, value.getClass());
                if (objectFactoryMethod != null) {
                    return marshall(type, exchange, value, objectFactoryMethod);
                }
            }
        }
    } catch (Exception e) {
        throw new TypeConversionException(value, type, e);
    }

    // should return null if didn't even try to convert at all or for whatever reason the conversion is failed
    return null;
}

public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
    if (!isStarted()) {
        throw new IllegalStateException("The endpoint is not active: " + getEndpoint().getEndpointUri());
    }
    BeanInvocation invocation = new BeanInvocation(method, args);
    Exchange exchange = getEndpoint().createExchange();
    exchange.getIn().setBody(invocation);
    try {
        log.debug("Invoking {} with args {}", method, args);
        getProcessor().process(exchange);
    } catch (Exception e) {
        exchange.setException(e);
    }

    // is there a matching exception from the signature, then throw that
    // or fallback and ensure the exception is thrown as a RemoteException
    Throwable fault = exchange.getException();
    if (fault != null) {
        Object match = null;
        for (Class<?> type : method.getExceptionTypes()) {
            Object found = exchange.getException(type);
            if (found != null) {
                match = found;
                break;
            }
        }
        if (match != null && match instanceof Throwable) {
            // we have a match
            throw (Throwable) match;
        } else {
            throw new RemoteException("Error invoking " + method, fault);
        }
    }
    return exchange.getOut().getBody();
}

/**
 * Creates a dynamic context for the given exchange
 */
protected DynamicQueryContext createDynamicContext(Exchange exchange) throws Exception {
    Configuration config = getConfiguration();
    DynamicQueryContext dynamicQueryContext = new DynamicQueryContext(config);

    Message in = exchange.getIn();
    Item item = null;
    if (ObjectHelper.isNotEmpty(getHeaderName())) {
        item = in.getHeader(getHeaderName(), Item.class);
    } else {
        item = in.getBody(Item.class);
    }
    if (item != null) {
        dynamicQueryContext.setContextItem(item);
    } else {
        Object body = null;
        if (ObjectHelper.isNotEmpty(getHeaderName())) {
            body = in.getHeader(getHeaderName());
        } else {
            body = in.getBody();
        }

        // the underlying input stream, which we need to close to avoid locking files or other resources
        InputStream is = null;
        try {
            Source source;
            // only convert to input stream if really needed
            if (isInputStreamNeeded(exchange)) {
                if (ObjectHelper.isNotEmpty(getHeaderName())) {
                    is = exchange.getIn().getHeader(getHeaderName(), InputStream.class);
                } else {
                    is = exchange.getIn().getBody(InputStream.class);
                }
                source = getSource(exchange, is);
            } else {
                source = getSource(exchange, body);
            }

            // special for bean invocation
            if (source == null) {
                if (body instanceof BeanInvocation) {
                    // if its a null bean invocation then handle that
                    BeanInvocation bi = exchange.getContext().getTypeConverter().convertTo(BeanInvocation.class, body);
                    if (bi.getArgs() != null && bi.getArgs().length == 1 && bi.getArgs()[0] == null) {
                        // its a null argument from the bean invocation so use null as answer
                        source = null;
                    }
                }
            }

            if (source == null) {
                // indicate it was not possible to convert to a Source type
                throw new NoTypeConversionAvailableException(body, Source.class);
            }

            DocumentInfo doc = config.buildDocument(source);
            dynamicQueryContext.setContextItem(doc);
        } finally {
            // can deal if is is null
            IOHelper.close(is);
        }
    }

    configureQuery(dynamicQueryContext, exchange);
    // call the reset if the in message body is StreamCache
    MessageHelper.resetStreamCache(exchange.getIn());
    return dynamicQueryContext;
}