Skip to content

Commit 2d59a0d

Browse files
olegzolegz
committed
GH-804 Add support for case-insensitive Cloud Event determination
Resolves #804
1 parent 5b4cec0 commit 2d59a0d

File tree

4 files changed

+64
-35
lines changed

4 files changed

+64
-35
lines changed

‎spring-cloud-function-context/src/main/java/org/springframework/cloud/function/cloudevent/CloudEventMessageUtils.java

Lines changed: 14 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,7 @@
2424
import java.util.stream.Collectors;
2525

2626
import org.springframework.cloud.function.context.message.MessageUtils;
27+
import org.springframework.cloud.function.context.message.MessageUtils.MessageStructureWithCaseInsensitiveHeaderKeys;
2728
import org.springframework.lang.Nullable;
2829
import org.springframework.messaging.Message;
2930
import org.springframework.messaging.MessageHeaders;
@@ -331,21 +332,22 @@ else if (Protocols.HTTP.equals(targetProtocol)) {
331332
* @return true if this Message represents Cloud Event in binary-mode
332333
*/
333334
public static boolean isCloudEvent(Message<?> message) {
334-
return (message.getHeaders().containsKey(SPECVERSION)
335-
&& message.getHeaders().containsKey(TYPE)
336-
&& message.getHeaders().containsKey(SOURCE))
335+
MessageStructureWithCaseInsensitiveHeaderKeys _message = MessageUtils.toCaseInsensitiveHeadersStructure(message);
336+
return (_message.getHeaders().containsKey(SPECVERSION)
337+
&& _message.getHeaders().containsKey(TYPE)
338+
&& _message.getHeaders().containsKey(SOURCE))
337339
||
338-
(message.getHeaders().containsKey(_SPECVERSION)
339-
&& message.getHeaders().containsKey(_TYPE)
340-
&& message.getHeaders().containsKey(_SOURCE))
340+
(_message.getHeaders().containsKey(_SPECVERSION)
341+
&& _message.getHeaders().containsKey(_TYPE)
342+
&& _message.getHeaders().containsKey(_SOURCE))
341343
||
342-
(message.getHeaders().containsKey(AMQP_ATTR_PREFIX + _SPECVERSION)
343-
&& message.getHeaders().containsKey(AMQP_ATTR_PREFIX + _TYPE)
344-
&& message.getHeaders().containsKey(AMQP_ATTR_PREFIX + _SOURCE))
344+
(_message.getHeaders().containsKey(AMQP_ATTR_PREFIX + _SPECVERSION)
345+
&& _message.getHeaders().containsKey(AMQP_ATTR_PREFIX + _TYPE)
346+
&& _message.getHeaders().containsKey(AMQP_ATTR_PREFIX + _SOURCE))
345347
||
346-
(message.getHeaders().containsKey(KAFKA_ATTR_PREFIX + _SPECVERSION)
347-
&& message.getHeaders().containsKey(KAFKA_ATTR_PREFIX + _TYPE)
348-
&& message.getHeaders().containsKey(KAFKA_ATTR_PREFIX + _SOURCE));
348+
(_message.getHeaders().containsKey(KAFKA_ATTR_PREFIX + _SPECVERSION)
349+
&& _message.getHeaders().containsKey(KAFKA_ATTR_PREFIX + _TYPE)
350+
&& _message.getHeaders().containsKey(KAFKA_ATTR_PREFIX + _SOURCE));
349351
}
350352

351353
private static boolean isAttribute(String key) {

‎spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/config/RoutingFunction.java

Lines changed: 2 additions & 23 deletions
Original file line numberDiff line numberDiff line change
@@ -16,8 +16,6 @@
1616

1717
package org.springframework.cloud.function.context.config;
1818

19-
import java.util.Map;
20-
import java.util.TreeMap;
2119
import java.util.function.Function;
2220

2321
import org.apache.commons.logging.Log;
@@ -31,6 +29,7 @@
3129
import org.springframework.cloud.function.context.MessageRoutingCallback;
3230
import org.springframework.cloud.function.context.MessageRoutingCallback.FunctionRoutingResult;
3331
import org.springframework.cloud.function.context.catalog.SimpleFunctionRegistry.FunctionInvocationWrapper;
32+
import org.springframework.cloud.function.context.message.MessageUtils;
3433
import org.springframework.context.expression.MapAccessor;
3534
import org.springframework.expression.BeanResolver;
3635
import org.springframework.expression.Expression;
@@ -196,7 +195,7 @@ private FunctionInvocationWrapper functionFromDefinition(String definition) {
196195
private FunctionInvocationWrapper functionFromExpression(String routingExpression, Object input) {
197196
Expression expression = spelParser.parseExpression(routingExpression);
198197
if (input instanceof Message) {
199-
input = new MessageStructureWithCaseInsensitiveHeaderKeys((Message<?>) input);
198+
input = MessageUtils.toCaseInsensitiveHeadersStructure((Message<?>) input);
200199
}
201200

202201
String functionName = expression.getValue(this.evalContext, input, String.class);
@@ -209,24 +208,4 @@ private FunctionInvocationWrapper functionFromExpression(String routingExpressio
209208
}
210209
return function;
211210
}
212-
213-
@SuppressWarnings({"rawtypes", "unused"})
214-
private static class MessageStructureWithCaseInsensitiveHeaderKeys {
215-
private final Object payload;
216-
private final Map headers;
217-
218-
@SuppressWarnings("unchecked")
219-
MessageStructureWithCaseInsensitiveHeaderKeys(Message message) {
220-
this.payload = message.getPayload();
221-
this.headers = new TreeMap<>(String.CASE_INSENSITIVE_ORDER);
222-
this.headers.putAll(message.getHeaders());
223-
}
224-
public Object getPayload() {
225-
return payload;
226-
}
227-
228-
public Map getHeaders() {
229-
return headers;
230-
}
231-
}
232211
}

‎spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/message/MessageUtils.java

Lines changed: 35 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,11 @@
1616

1717
package org.springframework.cloud.function.context.message;
1818

19+
import java.util.Map;
20+
import java.util.TreeMap;
21+
22+
import org.springframework.messaging.Message;
23+
1924
/**
2025
* @author Dave Syer
2126
* @author Oleg Zhurakousky
@@ -36,4 +41,34 @@ public abstract class MessageUtils {
3641
* Value for 'target-protocol' typically use as header key.
3742
*/
3843
public static String SOURCE_TYPE = "source-type";
44+
45+
/**
46+
* Returns (payload, headers) structure identical to `message` while substituting headers with case insensitive map.
47+
*/
48+
public static MessageStructureWithCaseInsensitiveHeaderKeys toCaseInsensitiveHeadersStructure(Message<?> message) {
49+
return new MessageStructureWithCaseInsensitiveHeaderKeys(message);
50+
}
51+
52+
/**
53+
* !!! INTERNAL USE ONLY, MAY CHANGE OR REMOVED WITHOUT NOTICE!!!
54+
*/
55+
@SuppressWarnings({"rawtypes"})
56+
public static class MessageStructureWithCaseInsensitiveHeaderKeys {
57+
private final Object payload;
58+
private final Map headers;
59+
60+
@SuppressWarnings("unchecked")
61+
MessageStructureWithCaseInsensitiveHeaderKeys(Message message) {
62+
this.payload = message.getPayload();
63+
this.headers = new TreeMap<>(String.CASE_INSENSITIVE_ORDER);
64+
this.headers.putAll(message.getHeaders());
65+
}
66+
public Object getPayload() {
67+
return payload;
68+
}
69+
70+
public Map getHeaders() {
71+
return headers;
72+
}
73+
}
3974
}

‎spring-cloud-function-context/src/test/java/org/springframework/cloud/function/cloudevent/CloudEventMessageUtilsAndBuilderTests.java

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -31,6 +31,19 @@
3131
*/
3232
public class CloudEventMessageUtilsAndBuilderTests {
3333

34+
@Test// see https://github.com/spring-cloud/spring-cloud-function/issues/805
35+
public void testHeaderKeyInsensitivity() {
36+
Message<String> httpMessage = MessageBuilder.withPayload("hello")
37+
.setHeader("cE-SoUrCe", "https://foo.bar")
38+
.setHeader("Ce-specVeRsion", "1.0")
39+
.setHeader("Ce-Type", "blah")
40+
.setHeader("x", "x")
41+
.setHeader("zzz", "zzz")
42+
.build();
43+
44+
assertThat(CloudEventMessageUtils.isCloudEvent(httpMessage)).isTrue();
45+
}
46+
3447
@Test// see https://github.com/spring-cloud/spring-cloud-function/issues/680
3548
public void testProperAttributeExtractionRegardlessOfTargetProtocol() {
3649
Message<String> ceMessage = CloudEventMessageBuilder.withData("foo").build();

0 commit comments

Comments
 (0)