Java 类org.apache.camel.spi.HeaderFilterStrategy 实例源码

/**
 * Copies the headers from the source to the target message.
 * 
 * @param source the source message
 * @param target the target message
 * @param strategy the header filter strategy which could help us to filter the protocol message headers
 * @param override whether to override existing headers
 */
public static void copyHeaders(Message source, Message target, HeaderFilterStrategy strategy, boolean override) {
    if (!source.hasHeaders()) {
        return;
    }

    for (Map.Entry<String, Object> entry : source.getHeaders().entrySet()) {
        String key = entry.getKey();
        Object value = entry.getValue();

        if (target.getHeader(key) == null || override) {
            if (strategy == null) {
                target.setHeader(key, value);
            } else if (!strategy.applyFilterToExternalHeaders(key, value, target.getExchange())) {
                // Just make sure we don't copy the protocol headers to target
                target.setHeader(key, value);
            }
        }
    }
}

private Exchange createExchange(ExchangePattern pattern, com.amazonaws.services.sqs.model.Message msg) {
    Exchange exchange = super.createExchange(pattern);
    Message message = exchange.getIn();
    message.setBody(msg.getBody());
    message.setHeaders(new HashMap<String, Object>(msg.getAttributes()));
    message.setHeader(SqsConstants.MESSAGE_ID, msg.getMessageId());
    message.setHeader(SqsConstants.MD5_OF_BODY, msg.getMD5OfBody());
    message.setHeader(SqsConstants.RECEIPT_HANDLE, msg.getReceiptHandle());
    message.setHeader(SqsConstants.ATTRIBUTES, msg.getAttributes());
    message.setHeader(SqsConstants.MESSAGE_ATTRIBUTES, msg.getMessageAttributes());

    //Need to apply the SqsHeaderFilterStrategy this time
    HeaderFilterStrategy headerFilterStrategy = getHeaderFilterStrategy();
    //add all sqs message attributes as camel message headers so that knowledge of 
    //the Sqs class MessageAttributeValue will not leak to the client
    for (Entry<String, MessageAttributeValue> entry : msg.getMessageAttributes().entrySet()) {
        String header = entry.getKey();
        Object value = translateValue(entry.getValue());
        if (!headerFilterStrategy.applyFilterToExternalHeaders(header, value, exchange)) {
            message.setHeader(header, value);
        }
    }
    return exchange;
}

public CamelConduit(CamelContext context, Bus b, EndpointInfo epInfo, EndpointReferenceType targetReference,
        HeaderFilterStrategy headerFilterStrategy) {
    super(getTargetReference(epInfo, targetReference, b));
    String address = epInfo.getAddress();
    if (address != null) {
        targetCamelEndpointUri = address.substring(CamelTransportConstants.CAMEL_TRANSPORT_PREFIX.length());
        if (targetCamelEndpointUri.startsWith("//")) {
            targetCamelEndpointUri = targetCamelEndpointUri.substring(2);
        }
    }
    camelContext = context;
    endpointInfo = epInfo;
    bus = b;
    initConfig();
    this.headerFilterStrategy = headerFilterStrategy;
    Endpoint target = getCamelContext().getEndpoint(targetCamelEndpointUri);
    try {
        producer = target.createProducer();
        producer.start();
    } catch (Exception e) {
        throw new RuntimeCamelException("Cannot create the producer rightly", e);
    }
}

@Before
public void setup() throws Exception {
    underTest = new SqsProducer(sqsEndpoint);
    sendMessageResult = new SendMessageResult().withMD5OfMessageBody(MESSAGE_MD5).withMessageId(MESSAGE_ID);
    sqsConfiguration = new SqsConfiguration();
    HeaderFilterStrategy headerFilterStrategy = new SqsHeaderFilterStrategy();
    sqsConfiguration.setDelaySeconds(Integer.valueOf(0));
    when(sqsEndpoint.getClient()).thenReturn(amazonSQSClient);
    when(sqsEndpoint.getConfiguration()).thenReturn(sqsConfiguration);
    when(amazonSQSClient.sendMessage(any(SendMessageRequest.class))).thenReturn(sendMessageResult);
    when(exchange.getOut()).thenReturn(outMessage);
    when(exchange.getIn()).thenReturn(inMessage);
    when(exchange.getPattern()).thenReturn(ExchangePattern.InOnly);
    when(inMessage.getBody(String.class)).thenReturn(SAMPLE_MESSAGE_BODY);
    when(sqsEndpoint.getQueueUrl()).thenReturn(QUEUE_URL);
    when(sqsEndpoint.getHeaderFilterStrategy()).thenReturn(headerFilterStrategy);
}

@Test
public void testContentType() {

    Exchange camelExchange = EasyMock.createMock(Exchange.class);
    HeaderFilterStrategy strategy = setupHeaderStrategy(camelExchange);
    Message cxfMessage = new MessageImpl();
    CxfHeaderHelper.propagateCamelToCxf(strategy, 
        Collections.<String, Object>singletonMap("Content-Type", "text/xml"), cxfMessage, camelExchange);

    Map<String, List<String>> cxfHeaders = CastUtils.cast((Map<?, ?>)cxfMessage.get(Message.PROTOCOL_HEADERS)); 
    assertEquals(1, cxfHeaders.size());
    assertEquals(1, cxfHeaders.get("Content-Type").size());
    assertEquals("text/xml", cxfHeaders.get("Content-Type").get(0)); 

    assertEquals("text/xml", cxfMessage.get(Message.CONTENT_TYPE));   
}

protected void populateResponse(Exchange exchange, JettyContentExchange httpExchange,
                                Message in, HeaderFilterStrategy strategy, int responseCode) throws IOException {
    Message answer = exchange.getOut();

    answer.setHeader(Exchange.HTTP_RESPONSE_CODE, responseCode);

    // must use response fields to get the http headers as
    // httpExchange.getHeaders() does not work well with multi valued headers
    Map<String, Collection<String>> headers = httpExchange.getResponseHeaders();
    for (Map.Entry<String, Collection<String>> ent : headers.entrySet()) {
        String name = ent.getKey();
        Collection<String> values = ent.getValue();
        for (String value : values) {
            if (name.toLowerCase().equals("content-type")) {
                name = Exchange.CONTENT_TYPE;
                exchange.setProperty(Exchange.CHARSET_NAME, IOHelper.getCharsetNameFromContentType(value));
            }
            if (strategy != null && !strategy.applyFilterToExternalHeaders(name, value, exchange)) {
                HttpHelper.appendHeader(answer.getHeaders(), name, value);
            }
        }
    }

    // preserve headers from in by copying any non existing headers
    // to avoid overriding existing headers with old values
    // We also need to apply the httpProtocolHeaderFilterStrategy to filter the http protocol header
    MessageHelper.copyHeaders(exchange.getIn(), answer, httpProtocolHeaderFilterStrategy, false);

    // extract body after headers has been set as we want to ensure content-type from Jetty HttpExchange
    // has been populated first
    answer.setBody(extractResponseBody(exchange, httpExchange));
}

public void testHeaderFilterStrategyAware() {
    MyComponent comp = new MyComponent(MyEndpoint.class);
    assertNull(comp.getHeaderFilterStrategy());

    HeaderFilterStrategy strategy = new DefaultHeaderFilterStrategy();
    comp.setHeaderFilterStrategy(strategy);

    MyEndpoint my = new MyEndpoint();
    comp.setEndpointHeaderFilterStrategy(my);

    assertSame(strategy, my.getHeaderFilterStrategy());
    assertSame(strategy, comp.getHeaderFilterStrategy());
}

/**
 * To use a custom HeaderFilterStrategy to filter header to and from Camel message.
 */
public void setHeaderFilterStrategy(HeaderFilterStrategy headerFilterStrategy) {
    this.headerFilterStrategy = headerFilterStrategy;
    if (restletBinding instanceof HeaderFilterStrategyAware) {
        ((HeaderFilterStrategyAware) restletBinding).setHeaderFilterStrategy(headerFilterStrategy);
    }
}

/**
 * To use a custom HeaderFilterStrategy to filter header to and from Camel message.
 */
public void setHeaderFilterStrategy(HeaderFilterStrategy headerFilterStrategy) {
    this.headerFilterStrategy = headerFilterStrategy;
    if (cxfBinding instanceof HeaderFilterStrategyAware) {
        ((HeaderFilterStrategyAware) cxfBinding).setHeaderFilterStrategy(headerFilterStrategy);
    }
}

@Test
public void testSetGetHeaderFilterStrategy() {
    DefaultCxfBinding cxfBinding = new DefaultCxfBinding();
    HeaderFilterStrategy hfs = new DefaultHeaderFilterStrategy();

    cxfBinding.setHeaderFilterStrategy(hfs);        
    assertSame("The header filter strategy is set", hfs, cxfBinding.getHeaderFilterStrategy());
}

@Test
public void testFilterCamelHeaders() throws Exception {
    HeaderFilterStrategy filter = new CxfRsHeaderFilterStrategy();
    assertTrue("Get a wrong filtered result", filter.applyFilterToCamelHeaders(Exchange.CHARSET_NAME, "just a test", null));
    assertTrue("Get a wrong filtered result", filter.applyFilterToCamelHeaders(CxfConstants.CAMEL_CXF_RS_RESPONSE_CLASS, "just a test", null));
    assertTrue("Get a wrong filtered result", filter.applyFilterToCamelHeaders("org.apache.camel.such.Header", "just a test", null));
    assertFalse("Get a wrong filtered result", filter.applyFilterToCamelHeaders("camel.result", "just a test", null));
}

public CamelDestination(CamelContext camelContext, Bus bus, ConduitInitiator ci, EndpointInfo info,
                        HeaderFilterStrategy headerFilterStrategy, boolean checkException) throws IOException {
    super(bus, getTargetReference(info, bus), info);
    this.camelContext = camelContext;
    conduitInitiator = ci;
    camelDestinationUri = endpointInfo.getAddress().substring(CamelTransportConstants.CAMEL_TRANSPORT_PREFIX.length());
    if (camelDestinationUri.startsWith("//")) {
        camelDestinationUri = camelDestinationUri.substring(2);
    }
    initConfig();
    this.headerFilterStrategy = headerFilterStrategy;
    this.checkException = checkException;
}

public static org.apache.cxf.message.Message getCxfInMessage(HeaderFilterStrategy headerFilterStrategy,
                                                             org.apache.camel.Exchange exchange,
                                                             boolean isClient) {
    MessageImpl answer = new MessageImpl();
    org.apache.cxf.message.Exchange cxfExchange = exchange
        .getProperty(CamelTransportConstants.CXF_EXCHANGE, org.apache.cxf.message.Exchange.class);
    org.apache.camel.Message message;
    if (isClient && exchange.hasOut()) {
        message = exchange.getOut();
    } else {
        message = exchange.getIn();
    }
    assert message != null;
    if (cxfExchange == null) {
        cxfExchange = new ExchangeImpl();
        exchange.setProperty(CamelTransportConstants.CXF_EXCHANGE, cxfExchange);
    }

    CxfHeaderHelper.propagateCamelToCxf(headerFilterStrategy, message.getHeaders(), answer, exchange);

    // body can be empty in case of GET etc.
    InputStream body = message.getBody(InputStream.class);
    if (body != null) {
        answer.setContent(InputStream.class, body);
    } else if (message.getBody() != null) {
        // fallback and set the body as what it is
        answer.setContent(Object.class, body);
    }

    answer.putAll(message.getHeaders());
    answer.setExchange(cxfExchange);
    cxfExchange.setInMessage(answer);
    return answer;
}

public void propagateResponseHeadersToCamel(Message cxfMessage, Exchange exchange,
                                            HeaderFilterStrategy strategy) {

    LOG.trace("Propagating response headers from CXF message {}", cxfMessage);

    if (strategy == null) {
        return;
    }

    Map<String, Object> camelHeaders = exchange.getOut().getHeaders();
    // copy the in message header to out message
    camelHeaders.putAll(exchange.getIn().getHeaders());

    Map<String, List<String>> cxfHeaders =
        CastUtils.cast((Map<?, ?>)cxfMessage.get(Message.PROTOCOL_HEADERS));

    if (cxfHeaders != null) {
        for (Map.Entry<String, List<String>> entry : cxfHeaders.entrySet()) {
            if (!strategy.applyFilterToExternalHeaders(entry.getKey(), entry.getValue(), exchange)) {
                camelHeaders.put(entry.getKey(), entry.getValue().get(0));

                LOG.trace("Populate header from CXF header={} value={}", entry.getKey(), entry.getValue());
            }
        }
    }

    // propagate HTTP RESPONSE_CODE
    String key = Message.RESPONSE_CODE;
    Object value = cxfMessage.get(key);
    if (value != null && !strategy.applyFilterToExternalHeaders(key, value, exchange)) {
        camelHeaders.put(Exchange.HTTP_RESPONSE_CODE, value);
        LOG.trace("Populate header from CXF header={} value={} as {}",
                new Object[]{key, value, Exchange.HTTP_RESPONSE_CODE}); 
    }

    // propagate HTTP CONTENT_TYPE
    if (cxfMessage.get(Message.CONTENT_TYPE) != null) {
        camelHeaders.put(Exchange.CONTENT_TYPE, cxfMessage.get(Message.CONTENT_TYPE));
    }
}

/**
 * Propagates Camel headers to CXF message.
 *
 * @param strategy header filter strategy
 * @param headers Camel header
 * @param message CXF message
 * @param exchange provides context for filtering
 */
public static void propagateCamelToCxf(HeaderFilterStrategy strategy,
        Map<String, Object> headers, Message message, Exchange exchange) {

    Map<String, List<String>> cxfHeaders =
        CastUtils.cast((Map<?, ?>)message.get(Message.PROTOCOL_HEADERS));

    if (cxfHeaders == null) {
        // use a treemap to keep ordering and ignore key case
        cxfHeaders = new TreeMap<String, List<String>>(String.CASE_INSENSITIVE_ORDER);
        message.put(Message.PROTOCOL_HEADERS, cxfHeaders);
    }

    for (Map.Entry<String, Object> entry : headers.entrySet()) {
        if (strategy != null
                && !strategy.applyFilterToCamelHeaders(entry.getKey(), entry.getValue(), exchange)) {

            if (Exchange.CONTENT_TYPE.equals(entry.getKey())) {
                message.put(Message.CONTENT_TYPE, entry.getValue());
            }
            if (Client.REQUEST_CONTEXT.equals(entry.getKey())
                || Client.RESPONSE_CONTEXT.equals(entry.getKey())
                || Message.RESPONSE_CODE.equals(entry.getKey())) {
                message.put(entry.getKey(), entry.getValue());
            } else {
                Object values = entry.getValue();
                if (values instanceof List<?>) {
                    cxfHeaders.put(entry.getKey(), CastUtils.cast((List<?>)values, String.class));
                } else {
                    List<String> listValue = new ArrayList<String>();
                    listValue.add(entry.getValue().toString());
                    cxfHeaders.put(entry.getKey(), listValue);
                }
            }
        }
    }
}

private HeaderFilterStrategy setupHeaderStrategy(Exchange exchange) {

        HeaderFilterStrategy strategy = EasyMock.createMock(HeaderFilterStrategy.class);
        strategy.applyFilterToCamelHeaders("Content-Type", "text/xml", exchange);
        EasyMock.expectLastCall().andReturn(false);
        EasyMock.replay(strategy);
        return strategy;
    }

protected void populateResponse(Exchange exchange, HttpMethod method, Message in, HeaderFilterStrategy strategy, int responseCode) throws IOException, ClassNotFoundException {
    //We just make the out message is not create when extractResponseBody throws exception,
    Object response = extractResponseBody(method, exchange, getEndpoint().isIgnoreResponseBody());
    Message answer = exchange.getOut();

    answer.setHeader(Exchange.HTTP_RESPONSE_CODE, responseCode);
    answer.setHeader(Exchange.HTTP_RESPONSE_TEXT, method.getStatusText());
    answer.setBody(response);

    // propagate HTTP response headers
    Header[] headers = method.getResponseHeaders();
    for (Header header : headers) {
        String name = header.getName();
        String value = header.getValue();
        if (name.toLowerCase().equals("content-type")) {
            name = Exchange.CONTENT_TYPE;
            exchange.setProperty(Exchange.CHARSET_NAME, IOHelper.getCharsetNameFromContentType(value));
        }

        // use http helper to extract parameter value as it may contain multiple values
        Object extracted = HttpHelper.extractHttpParameterValue(value);
        if (strategy != null && !strategy.applyFilterToExternalHeaders(name, extracted, exchange)) {
            HttpHelper.appendHeader(answer.getHeaders(), name, extracted);
        }
    }

    // endpoint might be configured to copy headers from in to out
    // to avoid overriding existing headers with old values just
    // filter the http protocol headers
    if (getEndpoint().isCopyHeaders()) {
        MessageHelper.copyHeaders(exchange.getIn(), answer, httpProtocolHeaderFilterStrategy, false);
    }
}

@Override
protected void populateResponse(Exchange exchange, JettyContentExchange httpExchange, Message in,
                                HeaderFilterStrategy strategy, int responseCode) throws IOException {

    Message answer = exchange.getOut();

    answer.setHeaders(in.getHeaders());
    answer.setHeader(Exchange.HTTP_RESPONSE_CODE, responseCode);
    answer.setBody("Not exactly the message the server returned.");
}

public HeaderFilterStrategy getHeaderFilterStrategy() {
    return headerFilterStrategy;
}

public void setHeaderFilterStrategy(HeaderFilterStrategy headerFilterStrategy) {
    this.headerFilterStrategy = headerFilterStrategy;
}

/**
 * To use a custom HeaderFilterStrategy to filter header to and from Camel message.
 */
public void setHeaderFilterStrategy(HeaderFilterStrategy strategy) {
    this.headerFilterStrategy = strategy;
}

public HeaderFilterStrategy getHeaderFilterStrategy() {
    return headerFilterStrategy;
}

public MailBinding(HeaderFilterStrategy headerFilterStrategy, ContentTypeResolver contentTypeResolver) {
    this.headerFilterStrategy = headerFilterStrategy;
    this.contentTypeResolver = contentTypeResolver;
}

/**
 * To use a custom {@link org.apache.camel.spi.HeaderFilterStrategy} to filter header to and from Camel message.
 */
public void setHeaderFilterStrategy(HeaderFilterStrategy strategy) {
    headerFilterStrategy = strategy;
}

public void setHeaderFilterStrategy(
        HeaderFilterStrategy headerFilterStrategy) {
    this.headerFilterStrategy = headerFilterStrategy;
}

public HeaderFilterStrategy getHeaderFilterStrategy() {
    if (headerFilterStrategy == null) {
        headerFilterStrategy = new SjmsHeaderFilterStrategy(isIncludeAllJMSXProperties());
    }
    return headerFilterStrategy;
}

public HeaderFilterStrategy getHeaderFilterStrategy() {
    return headerFilterStrategy;
}

public HeaderFilterStrategy getHeaderFilterStrategy() {
    return headerFilterStrategy;
}

public void setHeaderFilterStrategy(
        HeaderFilterStrategy headerFilterStrategy) {
    this.headerFilterStrategy = headerFilterStrategy;
}

public void setHeaderFilterStrategy(HeaderFilterStrategy strategy) {
    this.strategy = strategy;
}

public HeaderFilterStrategy getHeaderFilterStrategy() {
    return headerFilterStrategy;
}

public HeaderFilterStrategy getHeaderFilterStrategy() {
    return headerFilterStrategy;
}

public HeaderFilterStrategy getHeaderFilterStrategy() {
    return headerFilterStrategy;
}

public void setHeaderFilterStrategy(
        HeaderFilterStrategy headerFilterStrategy) {
    this.headerFilterStrategy = headerFilterStrategy;
}

public DefaultNettyHttpBinding(HeaderFilterStrategy headerFilterStrategy) {
    this.headerFilterStrategy = headerFilterStrategy;
}

@Override
public HeaderFilterStrategy getHeaderFilterStrategy() {
    return headerFilterStrategy;
}

@Override
public void setHeaderFilterStrategy(HeaderFilterStrategy headerFilterStrategy) {
    this.headerFilterStrategy = headerFilterStrategy;
}

public HeaderFilterStrategy getHeaderFilterStrategy() {
    return headerFilterStrategy;
}

public HeaderFilterStrategy getHeaderFilterStrategy() {
    return headerFilterStrategy;
}

public HeaderFilterStrategy getHeaderFilterStrategy() {
    return headerFilterStrategy;
}