Java 类org.apache.camel.InvalidPayloadException 实例源码

项目：Camel 文件：TokenXMLExpressionIterator.java

/**
 * Strategy to evaluate the exchange
 *
 * @param exchange   the exchange
 * @param closeStream whether to close the stream before returning from this method.
 * @return the evaluated value
 */
protected Object doEvaluate(Exchange exchange, boolean closeStream) {
    InputStream in = null;
    try {
        in = exchange.getIn().getMandatoryBody(InputStream.class);
        // we may read from a file, and want to support custom charset defined on the exchange
        String charset = IOHelper.getCharsetName(exchange);
        return createIterator(exchange, in, charset);
    } catch (InvalidPayloadException e) {
        exchange.setException(e);
        // must close input stream
        IOHelper.close(in);
        return null;
    } finally {
        if (closeStream) {
            IOHelper.close(in);
        }
    }
}

项目：Camel 文件：TokenPairExpressionIterator.java

/**
 * Strategy to evaluate the exchange
 *
 * @param exchange   the exchange
 * @param closeStream whether to close the stream before returning from this method.
 * @return the evaluated value
 */
protected Object doEvaluate(Exchange exchange, boolean closeStream) {
    InputStream in = null;
    try {
        in = exchange.getIn().getMandatoryBody(InputStream.class);
        // we may read from a file, and want to support custom charset defined on the exchange
        String charset = IOHelper.getCharsetName(exchange);
        return createIterator(exchange, in, charset);
    } catch (InvalidPayloadException e) {
        exchange.setException(e);
        // must close input stream
        IOHelper.close(in);
        return null;
    } finally {
        if (closeStream) {
            IOHelper.close(in);
        }
    }
}

项目：Camel 文件：MessageSupport.java

public <T> T getMandatoryBody(Class<T> type) throws InvalidPayloadException {
    // eager same instance type test to avoid the overhead of invoking the type converter
    // if already same type
    if (type.isInstance(body)) {
        return type.cast(body);
    }

    Exchange e = getExchange();
    if (e != null) {
        TypeConverter converter = e.getContext().getTypeConverter();
        try {
            return converter.mandatoryConvertTo(type, e, getBody());
        } catch (Exception cause) {
            throw new InvalidPayloadException(e, type, this, cause);
        }
    }
    throw new InvalidPayloadException(e, type, this);
}

项目：Camel 文件：AbstractCamelInvocationHandler.java

private static Object getBody(Exchange exchange, Class<?> type) throws InvalidPayloadException {
    // get the body from the Exchange from either OUT or IN
    if (exchange.hasOut()) {
        if (exchange.getOut().getBody() != null) {
            return exchange.getOut().getMandatoryBody(type);
        } else {
            return null;
        }
    } else {
        if (exchange.getIn().getBody() != null) {
            return exchange.getIn().getMandatoryBody(type);
        } else {
            return null;
        }
    }
}

项目：Camel 文件：JpaEndpoint.java

protected Expression createProducerExpression() {
    return new ExpressionAdapter() {
        public Object evaluate(Exchange exchange) {
            Object answer;

            // must have a body
            try {
                if (getEntityType() == null) {
                    answer = exchange.getIn().getMandatoryBody();
                } else {
                    answer = exchange.getIn().getMandatoryBody(getEntityType());
                }
            } catch (InvalidPayloadException e) {
                throw new InvalidPayloadRuntimeException(exchange, getEntityType(), e.getCause());
            }
            // is never null
            return answer;
        }
    };
}

项目：Camel 文件：DefaultHttpBinding.java

protected void doWriteGZIPResponse(Message message, HttpServletResponse response, Exchange exchange) throws IOException {
    byte[] bytes;
    try {
        bytes = message.getMandatoryBody(byte[].class);
    } catch (InvalidPayloadException e) {
        throw ObjectHelper.wrapRuntimeCamelException(e);
    }

    byte[] data = GZIPHelper.compressGZIP(bytes);
    ServletOutputStream os = response.getOutputStream();
    try {
        if (LOG.isDebugEnabled()) {
            LOG.debug("Streaming response as GZIP in non-chunked mode with content-length {} and buffer size: {}", data.length, response.getBufferSize());
        }
        response.setContentLength(data.length);
        os.write(data);
        os.flush();
    } finally {
        IOHelper.close(os);
    }
}

项目：Camel 文件：IronMQProducer.java

public void process(Exchange exchange) throws Exception {
    IronMQConfiguration configuration = getEndpoint().getConfiguration();
    if (IronMQConstants.CLEARQUEUE.equals(exchange.getIn().getHeader(IronMQConstants.OPERATION, String.class))) {
        this.ironQueue.clear();
    } else {
        Object messageId = null;
        Object body = exchange.getIn().getBody();
        if (body instanceof String[]) {
            messageId = this.ironQueue.pushMessages((String[])body, configuration.getVisibilityDelay());
        } else if (body instanceof String) {
            if (configuration.isPreserveHeaders()) {
                body = GsonUtil.getBodyFromMessage(exchange.getIn());
            }
            messageId = this.ironQueue.push((String)body, configuration.getVisibilityDelay());
        } else {
            throw new InvalidPayloadException(exchange, String.class);
        }
        LOG.trace("Send request [{}] from exchange [{}]...", body, exchange);
        LOG.trace("Received messageId [{}]", messageId);
        Message message = getMessageForResponse(exchange);
        message.setHeader(IronMQConstants.MESSAGE_ID, messageId);
    }
}

项目：Camel 文件：MongoDbProducer.java

private Function<Exchange, Object> createDoFindOneByQuery() {
    return exch -> {
        try {
            MongoCollection<BasicDBObject> dbCol = calculateCollection(exch);
            BasicDBObject o = exch.getIn().getMandatoryBody(BasicDBObject.class);

            BasicDBObject sortBy = exch.getIn().getHeader(MongoDbConstants.SORT_BY, BasicDBObject.class);
            BasicDBObject fieldFilter = exch.getIn().getHeader(MongoDbConstants.FIELDS_FILTER, BasicDBObject.class);

            if (fieldFilter == null) {
                fieldFilter = new BasicDBObject();
            }

            if (sortBy == null) {
                sortBy = new BasicDBObject();
            }

            BasicDBObject ret = dbCol.find(o).projection(fieldFilter).sort(sortBy).first();
            exch.getOut().setHeader(MongoDbConstants.RESULT_TOTAL_SIZE, ret == null ? 0 : 1);
            return ret;
        } catch (InvalidPayloadException e) {
            throw new CamelMongoDbException("Payload is no BasicDBObject", e);
        }
    };
}

项目：Camel 文件：MongoDbProducer.java

private Function<Exchange, Object> createDoFindById() {
    return exchange1 -> {
        try {
            MongoCollection<BasicDBObject> dbCol = calculateCollection(exchange1);
            String id = exchange1.getIn().getMandatoryBody(String.class);
            BasicDBObject o = new BasicDBObject("_id", id);
            DBObject ret;

            BasicDBObject fieldFilter = exchange1.getIn().getHeader(MongoDbConstants.FIELDS_FILTER, BasicDBObject.class);
            if (fieldFilter == null) {
                fieldFilter = new BasicDBObject();
            }
            ret = dbCol.find(o).projection(fieldFilter).first();
            exchange1.getOut().setHeader(MongoDbConstants.RESULT_TOTAL_SIZE, ret == null ? 0 : 1);
            return ret;
        } catch (InvalidPayloadException e) {
            throw new CamelMongoDbException("Invalid payload for findById", e);
        }
    };
}

项目：Camel 文件：MongoDbProducer.java

private Function<Exchange, Object> createDoSave() {
    return exchange1 -> {
        try {
            MongoCollection<BasicDBObject> dbCol = calculateCollection(exchange1);
            BasicDBObject saveObj = exchange1.getIn().getMandatoryBody(BasicDBObject.class);

            UpdateOptions options = new UpdateOptions().upsert(true);
            BasicDBObject queryObject = new BasicDBObject("_id", saveObj.get("_id"));
            UpdateResult result = dbCol.replaceOne(queryObject, saveObj, options);
            exchange1.getIn().setHeader(MongoDbConstants.OID, saveObj.get("_id"));
            return result;
        } catch (InvalidPayloadException e) {
            throw new CamelMongoDbException("Body incorrect type for save", e);
        }
    };
}

项目：Camel 文件：FlatpackEndpoint.java

public Parser createDelimitedParser(Exchange exchange) throws InvalidPayloadException, IOException {
    Reader bodyReader = exchange.getIn().getMandatoryBody(Reader.class);

    Parser parser;
    if (ObjectHelper.isEmpty(getResourceUri())) {
        parser = getParserFactory().newDelimitedParser(bodyReader, delimiter, textQualifier);
    } else {
        InputStream is = ResourceHelper.resolveMandatoryResourceAsInputStream(getCamelContext(), resourceUri);
        InputStreamReader reader = new InputStreamReader(is, IOHelper.getCharsetName(exchange));
        parser = getParserFactory().newDelimitedParser(reader, bodyReader, delimiter, textQualifier, ignoreFirstRecord);
    }

    if (isAllowShortLines()) {
        parser.setHandlingShortLines(true);
        parser.setIgnoreParseWarnings(true);
    }
    if (isIgnoreExtraColumns()) {
        parser.setIgnoreExtraColumns(true);
        parser.setIgnoreParseWarnings(true);
    }

    return parser;
}

项目：camel-c24io 文件：C24IOSource.java

protected ComplexDataObject parseDataObject(Exchange exchange) throws InvalidPayloadException, IOException {
    Source source = getSource();

    // lets set the input stream
    Reader reader = exchange.getIn().getBody(Reader.class);
    if (reader != null) {
        source.setReader(reader);
    } else {
        // TODO have some SAXSource handling code here?

        InputStream inStream = ExchangeHelper.getMandatoryInBody(exchange, InputStream.class);
        source.setInputStream(inStream);
    }
    ComplexDataObject object = source.readObject(element);
    return object;
}

项目：camel-c24io 文件：C24IOTransform.java

public void process(Exchange exchange) throws Exception {
    ComplexDataObject[][] objects = null;
    ComplexDataObject dataObject = null;
    try {
        dataObject = exchange.getIn().getMandatoryBody(ComplexDataObject.class);
    } catch (InvalidPayloadException e1) {
        try {
            objects = exchange.getIn().getMandatoryBody(ComplexDataObject[][].class);
        } catch (InvalidPayloadException e2) {
            objects = getInBodyAsArray(exchange, objects);
        }
    }

    if (objects == null) {
        if (dataObject == null) {
            dataObject = unmarshalDataObject(exchange);
        }
        objects = new ComplexDataObject[][]{{dataObject}};
    }
    Object result = transform(objects);

    Message out = exchange.getOut();
    out.setBody(result);
}

项目：camel-ldpath 文件：LDPathEngine.java

private static InputStream getBodyAsInputStream(Exchange exchange) throws IOException, InvalidPayloadException {
    Object rdf = exchange.getIn().getBody();
    if(rdf instanceof WrappedFile) {
        rdf = ((WrappedFile<?>) rdf).getFile();
    }

    if (rdf instanceof String) {
        return IOConverter.toInputStream((String) rdf, exchange);
    } else if (rdf instanceof InputStream) {
        return (InputStream) rdf;
    } else if (rdf instanceof File) {
        return IOConverter.toInputStream((File) rdf);
    } else if (rdf instanceof URL) {
        return IOConverter.toInputStream((URL) rdf);
    }
    // default to an InputStream
    return exchange.getIn().getMandatoryBody(InputStream.class);
}

项目：Camel 文件：MessageSupport.java

public Object getMandatoryBody() throws InvalidPayloadException {
    Object answer = getBody();
    if (answer == null) {
        throw new InvalidPayloadException(getExchange(), Object.class, this);
    }
    return answer;
}

项目：Camel 文件：ExpressionBuilder.java

/**
 * 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() + "]";
        }
    };
}

项目：Camel 文件：ConvertBodyTest.java

public void testConvertFailed() throws Exception {
    getMockEndpoint("mock:result").expectedMessageCount(0);

    try {
        template.sendBody("direct:invalid", "11");
        fail("Should have thrown an exception");
    } catch (RuntimeCamelException e) {
        assertTrue(e.getCause() instanceof InvalidPayloadException);
    }

    assertMockEndpointsSatisfied();
}

项目：Camel 文件：SimulatorTest.java

protected void assertRespondsWith(final String value, String containedText)
    throws InvalidPayloadException {
    Exchange response = template.request("direct:a", new Processor() {
        public void process(Exchange exchange) throws Exception {
            Message in = exchange.getIn();
            in.setBody("answer");
            in.setHeader("cheese", value);
        }
    });

    assertNotNull("Should receive a response!", response);

    String text = response.getOut().getMandatoryBody(String.class);
    assertStringContains(text, containedText);
}

项目：Camel 文件：MessageSupportTest.java

public void testGetMandatoryBody() throws Exception {
    Exchange exchange = new DefaultExchange(context);
    Message in = exchange.getIn();

    try {
        in.getMandatoryBody();
        fail("Should have thrown an exception");
    } catch (InvalidPayloadException e) {
        // expected
    }

    in.setBody("Hello World");

    assertEquals("Hello World", in.getMandatoryBody());
}

项目：Camel 文件：BeanProxyNoBindingTest.java

public void testBeanProxyFailureInvalidReturnType() throws Exception {
    Endpoint endpoint = context.getEndpoint("direct:start");
    OrderService service = ProxyHelper.createProxy(endpoint, false, OrderService.class);

    try {
        service.invalidReturnType("<order type=\"beer\">Carlsberg</order>");
        fail("Should have thrown exception");
    } catch (Exception e) {
        // expected
        InvalidPayloadException cause = assertIsInstanceOf(InvalidPayloadException.class, e.getCause());
        assertEquals(Integer.class, cause.getType());
    }
}

项目：Camel 文件：BeanProxyTest.java

public void testBeanProxyFailureInvalidReturnType() throws Exception {
    Endpoint endpoint = context.getEndpoint("direct:start");
    OrderService service = ProxyHelper.createProxy(endpoint, OrderService.class);

    try {
        service.invalidReturnType("<order type=\"beer\">Carlsberg</order>");
        fail("Should have thrown exception");
    } catch (Exception e) {
        // expected
        InvalidPayloadException cause = assertIsInstanceOf(InvalidPayloadException.class, e.getCause());
        assertEquals(Integer.class, cause.getType());
    }
}

项目：Camel 文件：VelocitySetHeaderTest.java

protected void assertRespondsWith(final String value, String expectedBody) throws InvalidPayloadException, InterruptedException {
    MockEndpoint mock = getMockEndpoint("mock:result");
    mock.expectedMessageCount(1);
    mock.expectedHeaderReceived("fruit", value);
    mock.expectedBodiesReceived(expectedBody);
    template.request("direct:start", new Processor() {
        public void process(Exchange exchange) throws Exception {
            Message in = exchange.getIn();
            in.setBody(value);

        }
    });
    mock.assertIsSatisfied();
}

项目：Camel 文件：HipchatProducer.java

private StatusLine sendRoomMessage(String room, Exchange exchange) throws IOException, InvalidPayloadException {
    String urlPath = String.format(getConfig().withAuthToken(HipchatApiConstants.URI_PATH_ROOM_NOTIFY), room);
    String backGroundColor = exchange.getIn().getHeader(HipchatConstants.MESSAGE_BACKGROUND_COLOR, String.class);
    Map<String, String> jsonParam = getCommonHttpPostParam(exchange);
    if (backGroundColor != null) {
        jsonParam.put(HipchatApiConstants.API_MESSAGE_COLOR, backGroundColor);
    }
    LOG.info("Sending message to room: " + room + ", " + MAPPER.writeValueAsString(jsonParam));
    StatusLine statusLine = post(urlPath, jsonParam);
    LOG.debug("Response status for send room message: " + statusLine);
    return statusLine;
}

项目：Camel 文件：HipchatProducer.java

private StatusLine sendUserMessage(String user, Exchange exchange) throws IOException, InvalidPayloadException {
    String urlPath = String.format(getConfig().withAuthToken(HipchatApiConstants.URI_PATH_USER_MESSAGE), user);
    Map<String, String> jsonParam = getCommonHttpPostParam(exchange);
    LOG.info("Sending message to user: " + user + ", " + MAPPER.writeValueAsString(jsonParam));
    StatusLine statusLine = post(urlPath, jsonParam);
    LOG.debug("Response status for send user message: " + statusLine);
    return statusLine;
}

项目：Camel 文件：HipchatProducer.java

private Map<String, String> getCommonHttpPostParam(Exchange exchange) throws InvalidPayloadException {
    String format = exchange.getIn().getHeader(HipchatConstants.MESSAGE_FORMAT, "text", String.class);
    String notify = exchange.getIn().getHeader(HipchatConstants.TRIGGER_NOTIFY, String.class);
    Map<String, String> jsonMap = new HashMap<String, String>(4);
    jsonMap.put(HipchatApiConstants.API_MESSAGE, exchange.getIn().getMandatoryBody(String.class));
    if (notify != null) {
        jsonMap.put(HipchatApiConstants.API_MESSAGE_NOTIFY, notify);
    }
    jsonMap.put(HipchatApiConstants.API_MESSAGE_FORMAT, format);
    return jsonMap;
}

项目：Camel 文件：TestSupport.java

/**
 * Asserts that the given exchange has an OUT message of the given body value
 *
 * @param exchange the exchange which should have an OUT message
 * @param expected the expected value of the OUT message
 * @throws InvalidPayloadException is thrown if the payload is not the expected class type
 */
public static void assertInMessageBodyEquals(Exchange exchange, Object expected) throws InvalidPayloadException {
    assertNotNull("Should have a response exchange!", exchange);

    Object actual;
    if (expected == null) {
        actual = exchange.getIn().getMandatoryBody();
        assertEquals("in body of: " + exchange, expected, actual);
    } else {
        actual = exchange.getIn().getMandatoryBody(expected.getClass());
    }
    assertEquals("in body of: " + exchange, expected, actual);

    LOG.debug("Received response: " + exchange + " with in: " + exchange.getIn());
}

项目：Camel 文件：TestSupport.java

/**
 * Asserts that the given exchange has an OUT message of the given body value
 *
 * @param exchange the exchange which should have an OUT message
 * @param expected the expected value of the OUT message
 * @throws InvalidPayloadException is thrown if the payload is not the expected class type
 */
public static void assertOutMessageBodyEquals(Exchange exchange, Object expected) throws InvalidPayloadException {
    assertNotNull("Should have a response exchange!", exchange);

    Object actual;
    if (expected == null) {
        actual = exchange.getOut().getMandatoryBody();
        assertEquals("output body of: " + exchange, expected, actual);
    } else {
        actual = exchange.getOut().getMandatoryBody(expected.getClass());
    }
    assertEquals("output body of: " + exchange, expected, actual);

    LOG.debug("Received response: " + exchange + " with out: " + exchange.getOut());
}

项目：Camel 文件：TestSupport.java

/**
 * Asserts that the given exchange has an OUT message of the given body value
 *
 * @param exchange the exchange which should have an OUT message
 * @param expected the expected value of the OUT message
 * @throws InvalidPayloadException is thrown if the payload is not the expected class type
 */
public static void assertInMessageBodyEquals(Exchange exchange, Object expected) throws InvalidPayloadException {
    assertNotNull("Should have a response exchange!", exchange);

    Object actual;
    if (expected == null) {
        actual = ExchangeHelper.getMandatoryInBody(exchange);
        assertEquals("in body of: " + exchange, expected, actual);
    } else {
        actual = ExchangeHelper.getMandatoryInBody(exchange, expected.getClass());
    }
    assertEquals("in body of: " + exchange, expected, actual);

    LOG.debug("Received response: " + exchange + " with in: " + exchange.getIn());
}

项目：Camel 文件：TestSupport.java

/**
 * Asserts that the given exchange has an OUT message of the given body value
 *
 * @param exchange the exchange which should have an OUT message
 * @param expected the expected value of the OUT message
 * @throws InvalidPayloadException is thrown if the payload is not the expected class type
 */
public static void assertOutMessageBodyEquals(Exchange exchange, Object expected) throws InvalidPayloadException {
    assertNotNull("Should have a response exchange!", exchange);

    Object actual;
    if (expected == null) {
        actual = ExchangeHelper.getMandatoryOutBody(exchange);
        assertEquals("output body of: " + exchange, expected, actual);
    } else {
        actual = ExchangeHelper.getMandatoryOutBody(exchange, expected.getClass());
    }
    assertEquals("output body of: " + exchange, expected, actual);

    LOG.debug("Received response: " + exchange + " with out: " + exchange.getOut());
}

项目：Camel 文件：FreemarkerSetHeaderTest.java

protected void assertRespondsWith(final String value, String expectedBody) throws InvalidPayloadException, InterruptedException {
    MockEndpoint mock = getMockEndpoint("mock:result");
    mock.expectedMessageCount(1);
    //mock.expectedHeaderReceived("fruit", value);
    mock.expectedBodiesReceived(expectedBody);
    template.request("direct:start", new Processor() {
        public void process(Exchange exchange) throws Exception {
            Message in = exchange.getIn();
            in.setBody(value);

        }
    });
    mock.assertIsSatisfied();
}

项目：Camel 文件：SimulatorTest.java

protected void assertRespondsWith(final String value, String containedText) throws InvalidPayloadException {
    Exchange response = template.request("direct:a", new Processor() {
        public void process(Exchange exchange) throws Exception {
            Message in = exchange.getIn();
            in.setBody("answer");
            in.setHeader("cheese", value);
        }
    });

    assertNotNull("Should receive a response!", response);

    String text = response.getOut().getMandatoryBody(String.class);
    assertStringContains(text, containedText);
}

项目：Camel 文件：CouchDbProducer.java

JsonElement getBodyAsJsonElement(Exchange exchange) throws InvalidPayloadException {
    Object body = exchange.getIn().getMandatoryBody();
    if (body instanceof String) {
        try {
            return new JsonParser().parse((String) body);
        } catch (JsonSyntaxException jse) {
            throw new InvalidPayloadException(exchange, body.getClass());
        }
    } else if (body instanceof JsonElement) {
        return (JsonElement) body;
    } else {
        throw new InvalidPayloadException(exchange, body != null ? body.getClass() : null);
    }
}

项目：Camel 文件：CouchDbProducerTest.java

@SuppressWarnings("unchecked")
@Test(expected = InvalidPayloadException.class)
public void testNullSaveResponseThrowsError() throws Exception {
    when(exchange.getIn().getMandatoryBody()).thenThrow(InvalidPayloadException.class);
    when(producer.getBodyAsJsonElement(exchange)).thenThrow(InvalidPayloadException.class);
    producer.process(exchange);
}

项目：Camel 文件：JaxbDataFormatMustBeJAXBElementTest.java

@Test
public void testJaxbMarshalling2() throws InterruptedException {
    getMockEndpoint("mock:result").expectedMessageCount(0);

    try {
        template.sendBody("direct:start2", "<foo><bar>Hello Bar</bar></foo>");
        fail("Should have thrown exception");
    } catch (CamelExecutionException e) {
        InvalidPayloadException ipe = assertIsInstanceOf(InvalidPayloadException.class, e.getCause().getCause());
        assertNotNull(ipe);
        assertEquals(JAXBElement.class, ipe.getType());
    }

    assertMockEndpointsSatisfied();
}

项目：Camel 文件：MongoDbProducer.java

private Function<Exchange, Object> createDoUpdate() {
    return exchange1 -> {
        try {
            MongoCollection<BasicDBObject> dbCol = calculateCollection(exchange1);
            List<BasicDBObject> saveObj = exchange1.getIn().getMandatoryBody((Class<List<BasicDBObject>>) (Class<?>) List.class);
            if (saveObj.size() != 2) {
                throw new CamelMongoDbException("MongoDB operation = insert, failed because body is not a List of DBObject objects with size = 2");
            }

            BasicDBObject updateCriteria = saveObj.get(0);
            BasicDBObject objNew = saveObj.get(1);

            Boolean multi = exchange1.getIn().getHeader(MongoDbConstants.MULTIUPDATE, Boolean.class);
            Boolean upsert = exchange1.getIn().getHeader(MongoDbConstants.UPSERT, Boolean.class);

            UpdateResult result;
            UpdateOptions options = new UpdateOptions();
            if (upsert != null) {
                options.upsert(true);
            }
            if (multi == null) {
                result = dbCol.updateOne(updateCriteria, objNew, options);
            } else {
                result = dbCol.updateMany(updateCriteria, objNew, options);
            }
            return result;
        } catch (InvalidPayloadException e) {
            throw new CamelMongoDbException("Invalid payload for update", e);
        }
    };
}

项目：Camel 文件：MongoDbProducer.java

private Function<Exchange, Object> createDoRemove() {
    return exchange1 -> {
        try {
            MongoCollection<BasicDBObject> dbCol = calculateCollection(exchange1);
            BasicDBObject removeObj = exchange1.getIn().getMandatoryBody(BasicDBObject.class);

            DeleteResult result = dbCol.deleteMany(removeObj);
            return result;
        } catch (InvalidPayloadException e) {
            throw new CamelMongoDbException("Invalid payload for remove", e);
        }
    };
}

项目：Camel 文件：MongoDbProducer.java

private Function<Exchange, Object> createDoCommand() {
    return exchange1 -> {
        try {
            MongoDatabase db = calculateDb(exchange1);
            BasicDBObject cmdObj = exchange1.getIn().getMandatoryBody(BasicDBObject.class);
            return db.runCommand(cmdObj);
        } catch (InvalidPayloadException e) {
            throw new CamelMongoDbException("Invalid payload for command", e);
        }
    };
}

项目：Camel 文件：TestSupport.java

/**
 * Asserts that the given exchange has an OUT message of the given body value
 *
 * @param exchange the exchange which should have an OUT message
 * @param expected the expected value of the OUT message
 * @throws InvalidPayloadException is thrown if the payload is not the expected class type
 */
public static void assertInMessageBodyEquals(Exchange exchange, Object expected) throws InvalidPayloadException {
    assertNotNull(exchange, "Should have a response exchange!");

    Object actual;
    if (expected == null) {
        actual = exchange.getIn().getMandatoryBody();
        assertEquals(actual, expected, "in body of: " + exchange);
    } else {
        actual = exchange.getIn().getMandatoryBody(expected.getClass());
    }
    assertEquals(actual, expected, "in body of: " + exchange);

    LOG.debug("Received response: " + exchange + " with in: " + exchange.getIn());
}

项目：Camel 文件：TestSupport.java

/**
 * Asserts that the given exchange has an OUT message of the given body value
 *
 * @param exchange the exchange which should have an OUT message
 * @param expected the expected value of the OUT message
 * @throws InvalidPayloadException is thrown if the payload is not the expected class type
 */
public static void assertOutMessageBodyEquals(Exchange exchange, Object expected) throws InvalidPayloadException {
    assertNotNull(exchange, "Should have a response exchange!");

    Object actual;
    if (expected == null) {
        actual = exchange.getOut().getMandatoryBody();
        assertEquals(actual, expected, "output body of: " + exchange);
    } else {
        actual = exchange.getOut().getMandatoryBody(expected.getClass());
    }
    assertEquals(actual, expected, "output body of: " + exchange);

    LOG.debug("Received response: " + exchange + " with out: " + exchange.getOut());
}

项目：Camel 文件：KickCommand.java

@Override
public void act(final Client client, final Exchange exchange) throws NoSuchHeaderException, InvalidPayloadException {
    final Integer jobs = exchange.getIn().getMandatoryBody(Integer.class);
    final int result = client.kick(jobs);
    if (LOG.isDebugEnabled()) {
        LOG.debug(String.format("Kick %d jobs. Kicked %d actually.", jobs, result));
    }

    final Message answer = getAnswerMessage(exchange);
    answer.setBody(result, Integer.class);
}