Java 类org.apache.camel.builder.PredicateBuilder 实例源码

@Override
public void configure() throws Exception {
    from(String.format("cxfrs://bean://%s", this.rsComponentId))
    .routeId("warehouse-rs-main-route")
    .choice()
        .when(PredicateBuilder.and(header("CamelHttpMethod").isEqualTo("POST"), header("CamelHttpUri").isEqualTo("/warehouse/product")))
            .beanRef(this.productDaoId, "createProduct(${body})")
            .beanRef(this.eventHandlerId, "notifyProductAdded(${body})")
        .when(PredicateBuilder.and(header("CamelHttpMethod").isEqualTo("GET"), header("CamelHttpUri").isEqualTo("/warehouse/product")))
            .beanRef(this.productDaoId, "readProducts()")
        .when(PredicateBuilder.and(header("CamelHttpMethod").isEqualTo("PUT"), header("CamelHttpUri").regex("/warehouse/product/[0-9]+")))
            .beanRef(this.productDaoId, "updateProduct(${body[0]}, ${body[1]})")
            .beanRef(this.eventHandlerId, "notifyProductModified(${body})")
         .when(PredicateBuilder.and(header("CamelHttpMethod").isEqualTo("DELETE"), header("CamelHttpUri").regex("/warehouse/product/[0-9]+")))
            .beanRef(this.productDaoId, "deleteProduct(${body})")
            .beanRef(this.eventHandlerId, "notifyProductDeleted(${body})")
    .end();
}

private Expression createAndExpression(final Expression leftExp, final Expression rightExp) {
    return new Expression() {
        @Override
        public <T> T evaluate(Exchange exchange, Class<T> type) {
            Predicate predicate = ExpressionToPredicateAdapter.toPredicate(leftExp);
            predicate = PredicateBuilder.and(predicate, ExpressionToPredicateAdapter.toPredicate(rightExp));

            boolean answer = predicate.matches(exchange);
            return exchange.getContext().getTypeConverter().convertTo(type, answer);
        }

        @Override
        public String toString() {
            return left + " " + token.getText() + " " + right;
        }
    };
}

private Expression createOrExpression(final Expression leftExp, final Expression rightExp) {
    return new Expression() {
        @Override
        public <T> T evaluate(Exchange exchange, Class<T> type) {
            Predicate predicate = ExpressionToPredicateAdapter.toPredicate(leftExp);
            predicate = PredicateBuilder.or(predicate, ExpressionToPredicateAdapter.toPredicate(rightExp));

            boolean answer = predicate.matches(exchange);
            return exchange.getContext().getTypeConverter().convertTo(type, answer);
        }

        @Override
        public String toString() {
            return left + " " + token.getText() + " " + right;
        }
    };
}

private Expression createRegexExpression(final Expression leftExp, final Expression rightExp) {
    return new Expression() {
        @Override
        public <T> T evaluate(Exchange exchange, Class<T> type) {
            // reg ex should use String pattern, so we evaluate the right hand side as a String
            Predicate predicate = PredicateBuilder.regex(leftExp, rightExp.evaluate(exchange, String.class));
            if (operator == BinaryOperatorType.NOT_REGEX) {
                predicate = PredicateBuilder.not(predicate);
            }
            boolean answer = predicate.matches(exchange);
            return exchange.getContext().getTypeConverter().convertTo(type, answer);
        }

        @Override
        public String toString() {
            return left + " " + token.getText() + " " + right;
        }
    };
}

@Override
protected RouteBuilder createRouteBuilder() throws Exception {
    return new RouteBuilder() {
        @Override
        public void configure() throws Exception {
            // build a compound predicate using the PredicateBuilder
            Predicate valid = PredicateBuilder.and(
                    // this xpath must return true
                    xpath("/book/title = 'Camel in Action'"),
                    // this simple must return true
                    simple("${header.source} == 'batch'"),
                    // this method call predicate must return false (as we use not)
                    not(method(CompoundPredicateTest.class, "isAuthor")));

            // use the predicate in the route using the validate eip
            from("direct:start")
                .validate(valid)
                .to("mock:valid");
        }
    };
}

private Expression createIsExpression(final String expression, final Expression leftExp, final Expression rightExp) {
    return new Expression() {
        @Override
        public <T> T evaluate(Exchange exchange, Class<T> type) {
            Predicate predicate;
            String name = rightExp.evaluate(exchange, String.class);
            if (name == null || "null".equals(name)) {
                throw new SimpleIllegalSyntaxException(expression, right.getToken().getIndex(), operator + " operator cannot accept null. A class type must be provided.");
            }
            Class<?> rightType = exchange.getContext().getClassResolver().resolveClass(name);
            if (rightType == null) {
                throw new SimpleIllegalSyntaxException(expression, right.getToken().getIndex(), operator + " operator cannot find class with name: " + name);
            }

            predicate = PredicateBuilder.isInstanceOf(leftExp, rightType);
            if (operator == BinaryOperatorType.NOT_IS) {
                predicate = PredicateBuilder.not(predicate);
            }
            boolean answer = predicate.matches(exchange);

            return exchange.getContext().getTypeConverter().convertTo(type, answer);
        }

        @Override
        public String toString() {
            return left + " " + token.getText() + " " + right;
        }
    };
}

private Expression createInExpression(final Expression leftExp, final Expression rightExp) {
    return new Expression() {
        @Override
        public <T> T evaluate(Exchange exchange, Class<T> type) {
            // okay the in operator is a bit more complex as we need to build a list of values
            // from the right hand side expression.
            // each element on the right hand side must be separated by comma (default for create iterator)
            Iterator<Object> it = ObjectHelper.createIterator(rightExp.evaluate(exchange, Object.class));
            List<Object> values = new ArrayList<Object>();
            while (it.hasNext()) {
                values.add(it.next());
            }
            // then reuse value builder to create the in predicate with the list of values
            ValueBuilder vb = new ValueBuilder(leftExp);
            Predicate predicate = vb.in(values.toArray());
            if (operator == BinaryOperatorType.NOT_IN) {
                predicate = PredicateBuilder.not(predicate);
            }
            boolean answer = predicate.matches(exchange);
            return exchange.getContext().getTypeConverter().convertTo(type, answer);
        }

        @Override
        public String toString() {
            return left + " " + token.getText() + " " + right;
        }
    };
}

private Expression createRangeExpression(final String expression, final Expression leftExp, final Expression rightExp) {
    return new Expression() {
        @Override
        public <T> T evaluate(Exchange exchange, Class<T> type) {
            Predicate predicate;

            String range = rightExp.evaluate(exchange, String.class);
            Matcher matcher = RANGE_PATTERN.matcher(range);
            if (matcher.matches()) {
                // wrap as constant expression for the from and to values
                Expression from = ExpressionBuilder.constantExpression(matcher.group(1));
                Expression to = ExpressionBuilder.constantExpression(matcher.group(3));

                // build a compound predicate for the range
                predicate = PredicateBuilder.isGreaterThanOrEqualTo(leftExp, from);
                predicate = PredicateBuilder.and(predicate, PredicateBuilder.isLessThanOrEqualTo(leftExp, to));
            } else {
                throw new SimpleIllegalSyntaxException(expression, right.getToken().getIndex(), operator + " operator is not valid. Valid syntax:'from..to' (where from and to are numbers).");
            }
            if (operator == BinaryOperatorType.NOT_RANGE) {
                predicate = PredicateBuilder.not(predicate);
            }

            boolean answer = predicate.matches(exchange);
            return exchange.getContext().getTypeConverter().convertTo(type, answer);
        }

        @Override
        public String toString() {
            return left + " " + token.getText() + " " + right;
        }
    };
}

public void testIssue() throws Exception {
    final Predicate fail = PredicateBuilder.or(
        header(Exchange.REDELIVERY_COUNTER).isNull(),
        header(Exchange.REDELIVERY_COUNTER).isLessThan(5));

    RouteDefinition route = context.getRouteDefinitions().get(0);
    route.adviceWith(context, new RouteBuilder() {
        @Override
        public void configure() throws Exception {
            interceptSendToEndpoint("seda:*")
                .skipSendToOriginalEndpoint()
                .process(new Processor() {
                    public void process(Exchange exchange) throws Exception {
                        invoked.incrementAndGet();

                        if (fail.matches(exchange)) {
                            throw new ConnectException("Forced");
                        }
                    }
                }).to("mock:ok");
        }
    });

    getMockEndpoint("mock:global").expectedMessageCount(0);
    getMockEndpoint("mock:ok").expectedMessageCount(1);
    getMockEndpoint("mock:exhausted").expectedMessageCount(0);

    template.sendBody("direct:start", "Hello World");

    assertMockEndpointsSatisfied();

    // 5 retry + 1 ok
    assertEquals(6, invoked.get());
}

public void testIssue() throws Exception {
    final Predicate fail = PredicateBuilder.or(
        header(Exchange.REDELIVERY_COUNTER).isNull(),
        header(Exchange.REDELIVERY_COUNTER).isLessThan(5));

    RouteDefinition route = context.getRouteDefinitions().get(0);
    route.adviceWith(context, new RouteBuilder() {
        @Override
        public void configure() throws Exception {
            interceptSendToEndpoint("seda:*")
                .skipSendToOriginalEndpoint()
                .process(new Processor() {
                    public void process(Exchange exchange) throws Exception {
                        invoked.incrementAndGet();

                        if (fail.matches(exchange)) {
                            throw new ConnectException("Forced");
                        }
                    }
                }).to("mock:ok");
        }
    });

    getMockEndpoint("mock:global").expectedMessageCount(0);
    getMockEndpoint("mock:ok").expectedMessageCount(1);
    getMockEndpoint("mock:exhausted").expectedMessageCount(0);

    template.sendBody("direct:start", "Hello World");

    assertMockEndpointsSatisfied();

    // 5 retry + 1 ok
    assertEquals(6, invoked.get());
}

public void testPredicateAssertHelper() throws Exception {
    Exchange exchange = new DefaultExchange(context);
    Predicate notNull = PredicateBuilder.isNotNull(constant("foo"));

    PredicateAssertHelper.assertMatches(notNull, "foo is not null", exchange);
    PredicateAssertHelper.assertMatches(notNull, null, exchange);
}

@Override
public void assertMessageExpected(DataSetEndpoint dataSetEndpoint, Exchange expected, Exchange actual, long index) throws Exception {
    // lets compare the XPath result
    Predicate predicate = PredicateBuilder.isEqualTo(expression, ExpressionBuilder.constantExpression(index));
    log.debug("evaluating predicate: " + predicate);
    PredicateAssertHelper.assertMatches(predicate, "Actual: " + actual, actual);
}

@Override
protected RouteBuilder createRouteBuilder() throws Exception {
    return new RouteBuilder() {
        @Override
        public void configure() throws Exception {
            from("direct:start")
                .choice()
                    .when().jsonpath("$.store.book[?(@.price < 10)]")
                        .to("mock:cheap")
                    .when().jsonpath("$.store.book[?(@.price < 30)]")
                        .to("mock:average")
                    .otherwise()
                        .to("mock:expensive");

            from("direct:bicycle")
                .choice()
                    .when().method(new BeanPredicate())
                        .to("mock:cheap")
                    .otherwise()
                        .to("mock:expensive");

            from("direct:bicycle2")
                .choice()
                .when(PredicateBuilder.isLessThan(ExpressionBuilder.languageExpression("jsonpath", "$.store.bicycle.price"), ExpressionBuilder.constantExpression(20)))
                    .to("mock:cheap")
                .otherwise()
                    .to("mock:expensive");
        }
    };
}

/**
 * Configures the route for output to websockets
 * @param routeBuilder
 * @param input
 */
public static void configureWebsocketTxRoute(RouteBuilder routeBuilder ,String input, int port){
    Predicate p1 = routeBuilder.header(ConfigConstants.OUTPUT_TYPE).isEqualTo(ConfigConstants.OUTPUT_WS);
    Predicate p2 = routeBuilder.header(WebsocketConstants.CONNECTION_KEY).isEqualTo(WebsocketConstants.SEND_TO_ALL);
    //from SEDA_WEBSOCKETS
        routeBuilder.from(input).id(getName("Websocket Tx"))
            .onException(Exception.class)
            .handled(true)
            .maximumRedeliveries(0)
            .to("log:nz.co.fortytwo.signalk.model.websocket.tx?level=ERROR&showException=true&showStackTrace=true")
            .end()
        .filter(PredicateBuilder.or(p1, p2))
        .to("skWebsocket://0.0.0.0:"+port+SignalKConstants.SIGNALK_WS).id(getName("Websocket Client"));

}

public static void configureTcpServerRoute(RouteBuilder routeBuilder ,String input, NettyServer nettyServer, String outputType) throws Exception{
    // push out via TCPServer.
    Predicate p1 = routeBuilder.header(ConfigConstants.OUTPUT_TYPE).isEqualTo(outputType);
    Predicate p2 = routeBuilder.header(WebsocketConstants.CONNECTION_KEY).isEqualTo(WebsocketConstants.SEND_TO_ALL);
    routeBuilder.from(input).id(getName("Netty "+outputType+" Server"))
        .onException(Exception.class)
        .handled(true)
        .maximumRedeliveries(0)
        .end()
    .filter(PredicateBuilder.or(p1, p2))
    .process((Processor) nettyServer).id(getName(NettyServer.class.getSimpleName())).end();

}

@Override
  public Predicate shouldProcessMessage() {

    // MSH-3
    Predicate sendingApplicationFilterPredicate = buildPredicate(HL7Constants.HL7_SENDING_APPLICATION, sendingApplicationFilterList);

// MSH-4
Predicate sendingFacilityFilterPredicate = buildPredicate(HL7Constants.HL7_SENDING_FACILITY, sendingFacilityFilterList);

// MSH-5
Predicate receivingApplicationFilterPredicate = buildPredicate(HL7Constants.HL7_RECEIVING_APPLICATION, receivingApplicationFilterList);

// MSH-6
Predicate receivingFacilityFilterPredicate = buildPredicate(HL7Constants.HL7_RECEIVING_FACILITY, receivingFacilityFilterList);

// MSH-9-1
Predicate messageTypeFilterPredicate = buildPredicate(HL7Constants.HL7_MESSAGE_TYPE, messageTypeFilterList);

// MSH-9-2
Predicate triggerEventPredicate = buildPredicate(HL7Constants.HL7_TRIGGER_EVENT, triggerEventFilterList);

      ArrayList<Predicate> predicates = new ArrayList<Predicate>();
      predicates.add(sendingApplicationFilterPredicate);
      predicates.add(sendingFacilityFilterPredicate);
      predicates.add(receivingApplicationFilterPredicate);
      predicates.add(receivingFacilityFilterPredicate);
      predicates.add(messageTypeFilterPredicate);
      predicates.add(triggerEventPredicate);
      return PredicateBuilder.and(predicates);
  }

@Override
public void configure() {

    final Predicate is412 = PredicateBuilder.toPredicate(simple("${exception.statusCode} == 412"));

    onException(HttpOperationFailedException.class)
            .onWhen(is412)
            .useOriginalMessage()
            .handled(true)
            .log(
                    INFO,
                    LOGGER,
                    "Received 412 from Milliner, skipping indexing."
            );

    onException(Exception.class)
            .maximumRedeliveries(maxRedeliveries)
            .log(
                    ERROR,
                    LOGGER,
                    "Error indexing resource in fcrepo: ${exception.message}\n\n${exception.stacktrace}"
            );

    from("{{content.stream}}")
            .routeId("FcrepoIndexerContent")
            .removeHeaders("*", "Authorization")
            .setHeader(Exchange.CONTENT_TYPE, constant("application/ld+json"))
            .setHeader(Exchange.HTTP_METHOD, constant("POST"))
            .toD(getMillinerBaseUrl() + "content");

    from("{{file.stream}}")
            .routeId("FcrepoIndexerFile")
            .removeHeaders("*", "Authorization")
            .setHeader(Exchange.CONTENT_TYPE, constant("application/ld+json"))
            .setHeader(Exchange.HTTP_METHOD, constant("POST"))
            .to(getMillinerBaseUrl() + "file");

    from("{{media.stream}}")
            .routeId("FcrepoIndexerMedia")
            .removeHeaders("*", "Authorization")
            .setHeader(Exchange.CONTENT_TYPE, constant("application/ld+json"))
            .setHeader(Exchange.HTTP_METHOD, constant("POST"))
            .to(getMillinerBaseUrl() + "media");

    from("{{delete.stream}}")
            .routeId("FcrepoIndexerDelete")
            .setProperty("urn").jsonpath("$.object.id")
            .setProperty("uuid").simple("${exchangeProperty.urn.replaceAll(\"urn:uuid:\",\"\")}")
            .removeHeaders("*", "Authorization")
            .setHeader(Exchange.HTTP_METHOD, constant("DELETE"))
            .toD(getMillinerBaseUrl() + "resource/${exchangeProperty.uuid}");
}

protected Predicate doParsePredicate() {

        // parse using the following grammar
        nextToken();
        while (!token.getType().isEol()) {
            // predicate supports quotes, functions, operators and whitespaces
            //CHECKSTYLE:OFF
            if (!singleQuotedLiteralWithFunctionsText()
                    && !doubleQuotedLiteralWithFunctionsText()
                    && !functionText()
                    && !unaryOperator()
                    && !binaryOperator()
                    && !logicalOperator()
                    && !isBooleanValue()
                    && !token.getType().isWhitespace()
                    && !token.getType().isEol()) {
                // okay the symbol was not one of the above, so its not supported
                // use the previous index as that is where the problem is
                throw new SimpleParserException("Unexpected token " + token, previousIndex);
            }
            //CHECKSTYLE:ON
            // take the next token
            nextToken();
        }

        // now after parsing we need a bit of work to do, to make it easier to turn the tokens
        // into and ast, and then from the ast, to Camel predicate(s).
        // hence why there is a number of tasks going on below to accomplish this

        // remove any ignorable white space tokens
        removeIgnorableWhiteSpaceTokens();
        // turn the tokens into the ast model
        parseTokensAndCreateNodes();
        // compact and stack blocks (eg function start/end, quotes start/end, etc.)
        prepareBlocks();
        // compact and stack unary expressions
        prepareUnaryExpressions();
        // compact and stack binary expressions
        prepareBinaryExpressions();
        // compact and stack logical expressions
        prepareLogicalExpressions();

        // create and return as a Camel predicate
        List<Predicate> predicates = createPredicates();
        if (predicates.isEmpty()) {
            // return a false predicate as response as there was nothing to parse
            return PredicateBuilder.constant(false);
        } else if (predicates.size() == 1) {
            return predicates.get(0);
        } else {
            return PredicateBuilder.and(predicates);
        }
    }

@Override
public Expression createExpression(String expression) {
    ObjectHelper.notNull(left, "left node", this);
    ObjectHelper.notNull(right, "right node", this);

    final Expression leftExp = left.createExpression(expression);
    final Expression rightExp = right.createExpression(expression);

    if (operator == BinaryOperatorType.EQ) {
        return createExpression(leftExp, rightExp, PredicateBuilder.isEqualTo(leftExp, rightExp));
    } else if (operator == BinaryOperatorType.EQ_IGNORE) {
        return createExpression(leftExp, rightExp, PredicateBuilder.isEqualToIgnoreCase(leftExp, rightExp));
    } else if (operator == BinaryOperatorType.GT) {
        return createExpression(leftExp, rightExp, PredicateBuilder.isGreaterThan(leftExp, rightExp));
    } else if (operator == BinaryOperatorType.GTE) {
        return createExpression(leftExp, rightExp, PredicateBuilder.isGreaterThanOrEqualTo(leftExp, rightExp));
    } else if (operator == BinaryOperatorType.LT) {
        return createExpression(leftExp, rightExp, PredicateBuilder.isLessThan(leftExp, rightExp));
    } else if (operator == BinaryOperatorType.LTE) {
        return createExpression(leftExp, rightExp, PredicateBuilder.isLessThanOrEqualTo(leftExp, rightExp));
    } else if (operator == BinaryOperatorType.NOT_EQ) {
        return createExpression(leftExp, rightExp, PredicateBuilder.isNotEqualTo(leftExp, rightExp));
    } else if (operator == BinaryOperatorType.CONTAINS) {
        return createExpression(leftExp, rightExp, PredicateBuilder.contains(leftExp, rightExp));
    } else if (operator == BinaryOperatorType.NOT_CONTAINS) {
        return createExpression(leftExp, rightExp, PredicateBuilder.not(PredicateBuilder.contains(leftExp, rightExp)));
    } else if (operator == BinaryOperatorType.IS || operator == BinaryOperatorType.NOT_IS) {
        return createIsExpression(expression, leftExp, rightExp);
    } else if (operator == BinaryOperatorType.REGEX || operator == BinaryOperatorType.NOT_REGEX) {
        return createRegexExpression(leftExp, rightExp);
    } else if (operator == BinaryOperatorType.IN || operator == BinaryOperatorType.NOT_IN) {
        return createInExpression(leftExp, rightExp);
    } else if (operator == BinaryOperatorType.RANGE || operator == BinaryOperatorType.NOT_RANGE) {
        return createRangeExpression(expression, leftExp, rightExp);
    } else if (operator == BinaryOperatorType.STARTS_WITH) {
        return createExpression(leftExp, rightExp, PredicateBuilder.startsWith(leftExp, rightExp));
    } else if (operator == BinaryOperatorType.ENDS_WITH) {
        return createExpression(leftExp, rightExp, PredicateBuilder.endsWith(leftExp, rightExp));
    }

    throw new SimpleParserException("Unknown binary operator " + operator, token.getIndex());
}

@Override
public Predicate createPredicate(String expression) {
    return PredicateBuilder.constant(true);

}

public void sendMessage() throws Exception {        
    DefaultCamelContext context = new DefaultCamelContext();
    context.addComponent("trade-executor", new TradeExecutorComponent());

    final CountDownLatch logonLatch = new CountDownLatch(2);
    final CountDownLatch executionReportLatch = new CountDownLatch(2);

    RouteBuilder routes = new RouteBuilder() {
        @Override
        public void configure() throws Exception {
            // Release latch when session logon events are received
            from("quickfix:examples/inprocess.cfg").
                filter(header(QuickfixjEndpoint.EVENT_CATEGORY_KEY).isEqualTo(QuickfixjEventCategory.SessionLogon)).
                bean(new CountDownLatchDecrementer("logon", logonLatch));

            from("quickfix:examples/inprocess.cfg?sessionID=FIX.4.2:MARKET->TRADER").
                filter(header(QuickfixjEndpoint.EVENT_CATEGORY_KEY).isEqualTo(QuickfixjEventCategory.AppMessageReceived)).
                to("trade-executor:market");

            from("trade-executor:market").to("quickfix:examples/inprocess.cfg");

            // Logger app messages as JSON
            from("quickfix:examples/inprocess.cfg").
                filter(PredicateBuilder.or(
                    header(QuickfixjEndpoint.EVENT_CATEGORY_KEY).isEqualTo(QuickfixjEventCategory.AppMessageReceived),
                    header(QuickfixjEndpoint.EVENT_CATEGORY_KEY).isEqualTo(QuickfixjEventCategory.AppMessageSent))).
                bean(new QuickfixjMessageJsonPrinter());

            // Release latch when trader receives execution report
            from("quickfix:examples/inprocess.cfg?sessionID=FIX.4.2:TRADER->MARKET").
                filter(PredicateBuilder.and(
                    header(QuickfixjEndpoint.EVENT_CATEGORY_KEY).isEqualTo(QuickfixjEventCategory.AppMessageReceived),
                    header(QuickfixjEndpoint.MESSAGE_TYPE_KEY).isEqualTo(MsgType.EXECUTION_REPORT))).
                bean(new CountDownLatchDecrementer("execution report", executionReportLatch));
        }
    };

    context.addRoutes(routes);

    LOG.info("Starting Camel context");
    context.start();

    // This is not strictly necessary, but it prevents the need for session
    // synchronization due to app messages being sent before being logged on
    if (!logonLatch.await(5, TimeUnit.SECONDS)) {
        throw new IllegalStateException("Logon did not complete");
    }

    String gatewayUri = "quickfix:examples/inprocess.cfg?sessionID=FIX.4.2:TRADER->MARKET";
    Endpoint gatewayEndpoint = context.getEndpoint(gatewayUri);
    Producer producer = gatewayEndpoint.createProducer();

    LOG.info("Sending order");

    NewOrderSingle order = createNewOrderMessage();
    Exchange exchange = producer.createExchange(ExchangePattern.InOnly);
    exchange.getIn().setBody(order);
    producer.process(exchange);            

    if (!executionReportLatch.await(5, TimeUnit.SECONDS)) {
        throw new IllegalStateException("Did not receive execution reports");
    }

    LOG.info("Message received, shutting down Camel context");

    context.stop();

    LOG.info("Order execution example complete");
}

public void sendMessage() throws Exception {        
    DefaultCamelContext context = new DefaultCamelContext();

    final CountDownLatch logonLatch = new CountDownLatch(2);
    final CountDownLatch receivedMessageLatch = new CountDownLatch(1);

    RouteBuilder routes = new RouteBuilder() {
        @Override
        public void configure() throws Exception {
            // Release latch when session logon events are received
            // We expect two events, one for the trader session and one for the market session
            from("quickfix:examples/inprocess.cfg").
                filter(header(QuickfixjEndpoint.EVENT_CATEGORY_KEY).isEqualTo(QuickfixjEventCategory.SessionLogon)).
                bean(new CountDownLatchDecrementer("logon", logonLatch));

            // For all received messages, print the JSON-formatted message to stdout
            from("quickfix:examples/inprocess.cfg").
                filter(PredicateBuilder.or(
                        header(QuickfixjEndpoint.EVENT_CATEGORY_KEY).isEqualTo(QuickfixjEventCategory.AdminMessageReceived),
                        header(QuickfixjEndpoint.EVENT_CATEGORY_KEY).isEqualTo(QuickfixjEventCategory.AppMessageReceived))).
                bean(new QuickfixjMessageJsonPrinter());

            // If the market session receives an email then release the latch
            from("quickfix:examples/inprocess.cfg?sessionID=FIX.4.2:MARKET->TRADER").
                filter(header(QuickfixjEndpoint.MESSAGE_TYPE_KEY).isEqualTo(MsgType.EMAIL)).
                bean(new CountDownLatchDecrementer("message", receivedMessageLatch));
        }
    };

    context.addRoutes(routes);

    LOG.info("Starting Camel context");
    context.start();

    if (!logonLatch.await(5L, TimeUnit.SECONDS)) {
        throw new IllegalStateException("Logon did not succeed");
    }

    String marketUri = "quickfix:examples/inprocess.cfg?sessionID=FIX.4.2:TRADER->MARKET";
    Producer producer = context.getEndpoint(marketUri).createProducer();

    Email email = TestSupport.createEmailMessage("Example");
    Exchange exchange = producer.createExchange(ExchangePattern.InOnly);
    exchange.getIn().setBody(email);
    producer.process(exchange);            

    if (!receivedMessageLatch.await(5L, TimeUnit.SECONDS)) {
        throw new IllegalStateException("Message did not reach market");
    }

    LOG.info("Message received, shutting down Camel context");

    context.stop();

    LOG.info("Example complete");
}

public void run() throws Exception {        
    DefaultCamelContext context = new DefaultCamelContext();

    final CountDownLatch logoutLatch = new CountDownLatch(1);

    RouteBuilder routes = new RouteBuilder() {
        @Override
        public void configure() throws Exception {
            // Modify the outgoing logon message to add a password
            // The modified message will be sent from the FIX engine when the message exchange completes
            from("quickfix:examples/inprocess.cfg?sessionID=FIX.4.2:TRADER->MARKET").
                filter(PredicateBuilder.and(
                        header(QuickfixjEndpoint.EVENT_CATEGORY_KEY).isEqualTo(QuickfixjEventCategory.AdminMessageSent),
                        header(QuickfixjEndpoint.MESSAGE_TYPE_KEY).isEqualTo(MsgType.LOGON))).
                bean(new CredentialInjector("PASSWORD"));

            // Release latch when the trader received a logout message
            from("quickfix:examples/inprocess.cfg?sessionID=FIX.4.2:TRADER->MARKET").
                filter(header(QuickfixjEndpoint.EVENT_CATEGORY_KEY).isEqualTo(QuickfixjEventCategory.SessionLogoff)).
                bean(new CountDownLatchDecrementer("logout", logoutLatch));

            // Reject all logons on market side
            // Demonstrates how to validate logons
            from("quickfix:examples/inprocess.cfg?sessionID=FIX.4.2:MARKET->TRADER").
                filter(PredicateBuilder.and(
                        header(QuickfixjEndpoint.EVENT_CATEGORY_KEY).isEqualTo(QuickfixjEventCategory.AdminMessageReceived),
                        header(QuickfixjEndpoint.MESSAGE_TYPE_KEY).isEqualTo(MsgType.LOGON))).
                bean(new LogonAuthenticator());
        }
    };

    context.addRoutes(routes);

    LOG.info("Starting Camel context");
    context.start();

    if (!logoutLatch.await(5L, TimeUnit.SECONDS)) {
        throw new IllegalStateException("Logout was not received");
    }

    context.stop();

    LOG.info("Example complete");
}

@Test
public void sendMessage() throws Exception {

    final CountDownLatch logonLatch = new CountDownLatch(2);
    final CountDownLatch receivedMessageLatch = new CountDownLatch(1);

    CamelContext camelctx = new DefaultCamelContext();
    camelctx.addRoutes(new RouteBuilder() {
        @Override
        public void configure() throws Exception {
            // Release latch when session logon events are received
            // We expect two events, one for the trader session and one for the market session
            from("quickfix:quickfix/inprocess.cfg").
                filter(header(QuickfixjEndpoint.EVENT_CATEGORY_KEY).isEqualTo(QuickfixjEventCategory.SessionLogon)).
                bean(new CountDownLatchDecrementer("logon", logonLatch));

            // For all received messages, print the JSON-formatted message to stdout
            from("quickfix:quickfix/inprocess.cfg").
                filter(PredicateBuilder.or(
                        header(QuickfixjEndpoint.EVENT_CATEGORY_KEY).isEqualTo(QuickfixjEventCategory.AdminMessageReceived),
                        header(QuickfixjEndpoint.EVENT_CATEGORY_KEY).isEqualTo(QuickfixjEventCategory.AppMessageReceived))).
                bean(new QuickfixjMessageJsonPrinter());

            // If the market session receives an email then release the latch
            from("quickfix:quickfix/inprocess.cfg?sessionID=FIX.4.2:MARKET->TRADER").
                filter(header(QuickfixjEndpoint.MESSAGE_TYPE_KEY).isEqualTo(MsgType.EMAIL)).
                bean(new CountDownLatchDecrementer("message", receivedMessageLatch));
        }
    });

    camelctx.start();
    try {
        Assert.assertTrue("Logon succeed", logonLatch.await(5L, TimeUnit.SECONDS));

        String marketUri = "quickfix:quickfix/inprocess.cfg?sessionID=FIX.4.2:TRADER->MARKET";
        Producer producer = camelctx.getEndpoint(marketUri).createProducer();

        Email email = createEmailMessage("Example");
        Exchange exchange = producer.createExchange(ExchangePattern.InOnly);
        exchange.getIn().setBody(email);
        producer.process(exchange);

        Assert.assertTrue("Message reached market", receivedMessageLatch.await(5L, TimeUnit.SECONDS));
    } finally {
        camelctx.stop();
    }
}

public void testToProcessorPredicate() throws Exception {
    Predicate pred = PredicateBuilder.isEqualTo(headerExpression("foo"), constant("bar"));

    Exchange exchange = new DefaultExchange(context);
    exchange.getIn().setHeader("foo", "bar");
    exchange.getIn().setBody("Hello World");

    Processor pro = CamelConverter.toProcessor(pred);

    pro.process(exchange);

    assertEquals(true, exchange.getOut().getBody());
}

public void testMockPredicateAsParameter() throws Exception {
    MockEndpoint mock = getMockEndpoint("mock:foo");
    mock.message(0).predicate(PredicateBuilder.isNotNull(header("foo")));

    mock.expectedMessageCount(1);

    template.sendBodyAndHeader("direct:start", "Hello World", "foo", "bar");

    assertMockEndpointsSatisfied();
}