From 97347bf30d84f2f059032e30495e7c010597cba2 Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Wed, 11 Nov 2020 09:18:51 +0100 Subject: [PATCH] GH-422 Improvements in cloud event samples Added initial README Polished tests --- .../catalog/SimpleFunctionRegistry.java | 9 +-- .../CloudEventJsonMessageConverter.java | 4 ++ .../function-sample-cloudevent/README.adoc | 71 +++++++++++++++++++ .../function-sample-cloudevent/pom.xml | 22 +++++- .../cloudevent/CloudeventDemoApplication.java | 59 +++++---------- ...> CloudeventDemoApplicationRESTTests.java} | 33 ++++++++- .../CloudeventDemoApplicationStreamTests.java | 27 +++++++ 7 files changed, 179 insertions(+), 46 deletions(-) create mode 100644 spring-cloud-function-samples/function-sample-cloudevent/README.adoc rename spring-cloud-function-samples/function-sample-cloudevent/src/test/java/io/spring/cloudevent/{CloudeventDemoApplicationTests.java => CloudeventDemoApplicationRESTTests.java} (87%) create mode 100644 spring-cloud-function-samples/function-sample-cloudevent/src/test/java/io/spring/cloudevent/CloudeventDemoApplicationStreamTests.java diff --git a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/SimpleFunctionRegistry.java b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/SimpleFunctionRegistry.java index 5b2e91419..c07e71d9c 100644 --- a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/SimpleFunctionRegistry.java +++ b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/SimpleFunctionRegistry.java @@ -806,7 +806,7 @@ public class SimpleFunctionRegistry implements FunctionRegistry, FunctionInspect else if (input instanceof Message) { convertedInput = this.convertInputMessageIfNecessary((Message) input, type); if (convertedInput == null) { // give ConversionService a chance - convertedInput = this.convertNonMessageInputIfNecessary(type, ((Message) input).getPayload()); + convertedInput = this.convertNonMessageInputIfNecessary(type, ((Message) input).getPayload(), false); } if (convertedInput != null && !FunctionTypeUtils.isMultipleArgumentType(this.inputType)) { convertedInput = !convertedInput.equals(input) @@ -818,7 +818,7 @@ public class SimpleFunctionRegistry implements FunctionRegistry, FunctionInspect } } else { - convertedInput = this.convertNonMessageInputIfNecessary(type, input); + convertedInput = this.convertNonMessageInputIfNecessary(type, input, JsonMapper.isJsonString(input)); if (convertedInput != null && logger.isDebugEnabled()) { logger.debug("Converted input: " + input + " to: " + convertedInput); } @@ -827,6 +827,7 @@ public class SimpleFunctionRegistry implements FunctionRegistry, FunctionInspect if (this.isWrapConvertedInputInMessage(convertedInput)) { convertedInput = MessageBuilder.withPayload(convertedInput).build(); } + Assert.notNull(convertedInput, "Failed to convert input: " + input + " to " + type); return convertedInput; } @@ -897,13 +898,13 @@ public class SimpleFunctionRegistry implements FunctionRegistry, FunctionInspect /* * */ - private Object convertNonMessageInputIfNecessary(Type inputType, Object input) { + private Object convertNonMessageInputIfNecessary(Type inputType, Object input, boolean maybeJson) { Object convertedInput = null; Class rawInputType = this.isTypePublisher(inputType) || this.isInputTypeMessage() ? FunctionTypeUtils.getRawType(FunctionTypeUtils.getGenericType(inputType)) : this.getRawClassFor(inputType); - if (JsonMapper.isJsonString(input) && !Message.class.isAssignableFrom(rawInputType)) { + if (maybeJson && !Message.class.isAssignableFrom(rawInputType)) { if (FunctionTypeUtils.isMessage(inputType)) { inputType = FunctionTypeUtils.getGenericType(inputType); } diff --git a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/config/CloudEventJsonMessageConverter.java b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/config/CloudEventJsonMessageConverter.java index a4c7e0612..373e49678 100644 --- a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/config/CloudEventJsonMessageConverter.java +++ b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/config/CloudEventJsonMessageConverter.java @@ -17,6 +17,7 @@ package org.springframework.cloud.function.context.config; import java.lang.reflect.Type; +import java.util.Collection; import java.util.Map; import org.springframework.cloud.function.json.JsonMapper; @@ -50,6 +51,9 @@ public class CloudEventJsonMessageConverter extends JsonMessageConverter { return super.convertFromInternal(message, targetClass, conversionHint); } else { + if (targetClass.isInstance(message.getPayload()) && !(message.getPayload() instanceof Collection)) { + return message.getPayload(); + } Type convertToType = conversionHint == null ? targetClass : (Type) conversionHint; String jsonString = (String) message.getPayload(); Map mapEvent = this.mapper.fromJson(jsonString, Map.class); diff --git a/spring-cloud-function-samples/function-sample-cloudevent/README.adoc b/spring-cloud-function-samples/function-sample-cloudevent/README.adoc new file mode 100644 index 000000000..a9d65a584 --- /dev/null +++ b/spring-cloud-function-samples/function-sample-cloudevent/README.adoc @@ -0,0 +1,71 @@ +## Cloud Events with Spring samples + +### Introduction +The current example uses spring-cloud-function framework as its core which allows users to only worry about functional aspects of +their requirement while taking care-off non-functional aspects. For more information on Spring Cloud Function please visit +our https://spring.io/projects/spring-cloud-function[project page]. +The example provides dependency and instructions to demonstrate several distinct invocation models: + + - Direct function invocation + - Function as a REST endpoint + - Function as message handler (e.g., Kafka, RabbitMQ etc) + - Function invocation via RSocket + +The POM file defines all the necessary dependency in a segregated way, so you can choose the one you're interested in. + +#### Direct function invocation + +#### Function as a REST endpoint + +Given that SCF allows function to be exposed as REST endpoints, you can post cloud event to any of the +functions by using function name as path (e.g., localhost:8080/) + +Here is an example of curl command posting a cloud event in binary-mode: + +[source, text] +---- +curl -w'\n' localhost:8080/asPOJO \ + -H "ce-Specversion: 1.0" \ + -H "ce-Type: com.example.springevent" \ + -H "ce-Source: spring.io/spring-event" \ + -H "Content-Type: application/json" \ + -H "ce-Id: 0001" \ + -d '{"releaseDate":"24-03-2004", "releaseName":"Spring Framework", "version":"1.0"}' +---- + +And here is an example of curl command posting a cloud event in structured-mode: + +[source, text] +---- +curl -w'\n' localhost:8080/asString \ + -H "ce-Specversion: 1.0" \ + -H "ce-Type: com.example.springevent" \ + -H "ce-Source: spring.io/spring-event" \ + -H "Content-Type: application/cloudevents+json" \ + -H "ce-Id: 0001" \ + -d '{ + "specversion" : "1.0", + "type" : "org.springframework", + "source" : "https://spring.io/", + "id" : "A234-1234-1234", + "datacontenttype" : "application/json", + "data" : { + "version" : "1.0", + "releaseName" : "Spring Framework", + "releaseDate" : "24-03-2004" + } +}' +---- + +#### Function as message handler (e.g., Kafka, RabbitMQ etc) + +Streaming support for Kafka and Rabbit is provided via Spring Cloud Stream framework (link). In fact we're only mentioning Kafka and Rabbit here as an example. +Streaming support is automatically provided for any existing binders (e.g., Solace, GCP, AWS etc) (link) +Binders are components of SCSt responsible to bind user code (e.g., function) to broker destinations so execution is triggered +by messages on broker destination and results of execution are sent to broker destinations. Binders also provide support consumer +groups and partitioning for both Kafka and RabbitMQ messaging systems. + + +#### Function invocation via RSocket + +TBD \ No newline at end of file diff --git a/spring-cloud-function-samples/function-sample-cloudevent/pom.xml b/spring-cloud-function-samples/function-sample-cloudevent/pom.xml index 90da5d0cc..52a3567cf 100644 --- a/spring-cloud-function-samples/function-sample-cloudevent/pom.xml +++ b/spring-cloud-function-samples/function-sample-cloudevent/pom.xml @@ -23,22 +23,42 @@ org.springframework.boot spring-boot-starter + + org.springframework.boot spring-boot-starter-web - org.springframework.cloud spring-cloud-function-web 3.1.0-SNAPSHOT + + + + + + + + + + + + + + + + + + + org.springframework.boot diff --git a/spring-cloud-function-samples/function-sample-cloudevent/src/main/java/io/spring/cloudevent/CloudeventDemoApplication.java b/spring-cloud-function-samples/function-sample-cloudevent/src/main/java/io/spring/cloudevent/CloudeventDemoApplication.java index b8578a5ce..c50848174 100644 --- a/spring-cloud-function-samples/function-sample-cloudevent/src/main/java/io/spring/cloudevent/CloudeventDemoApplication.java +++ b/spring-cloud-function-samples/function-sample-cloudevent/src/main/java/io/spring/cloudevent/CloudeventDemoApplication.java @@ -43,59 +43,38 @@ public class CloudeventDemoApplication { SpringApplication.run(CloudeventDemoApplication.class, args); } - /* - * curl -w'\n' localhost:8080/asStringMessage \ - * -H "Ce-Specversion: 1.0" \ - * -H "Ce-Type: com.example.springevent" \ - * -H "Ce-Source: spring.io/spring-event" \ - * -H "Content-Type: application/json" \ - * -H "Ce-Id: 0001" \ - * -d '{"releaseDate":"2004-03-24", "releaseName":"Spring Framework", "version":"1.0"}' - */ @Bean public Function, String> asStringMessage() { - return v -> v.getPayload().toString(); + return v -> { + System.out.println("Received Cloud Event with raw data: " + v); + return v.getPayload(); + }; } - /* - * curl -w'\n' localhost:8080/asString \ - * -H "Ce-Specversion: 1.0" \ - * -H "Ce-Type: com.example.springevent" \ - * -H "Ce-Source: spring.io/spring-event" \ - * -H "Content-Type: application/json" \ - * -H "Ce-Id: 0001" \ - * -d '{"releaseDate":"2004-03-24", "releaseName":"Spring Framework", "version":"1.0"}' - */ + @Bean public Function asString() { - return v -> v; + return v -> { + System.out.println("Received raw Cloud Event data: " + v); + return v; + }; } - /* - * curl -w'\n' localhost:8080/asPOJOMessage \ - * -H "Ce-Specversion: 1.0" \ - * -H "Ce-Type: com.example.springevent" \ - * -H "Ce-Source: spring.io/spring-event" \ - * -H "Content-Type: application/json" \ - * -H "Ce-Id: 0001" \ - * -d '{"releaseDate":"2004-03-24", "releaseName":"Spring Framework", "version":"1.0"}' - */ + @Bean public Function, String> asPOJOMessage() { - return v -> v.getPayload().toString(); + return v -> { + System.out.println("Received Cloud Event with POJO data: " + v); + return v.getPayload().toString(); + }; } - /* - * curl -w'\n' localhost:8080/asPOJO \ - * -H "Ce-Specversion: 1.0" \ - * -H "Ce-Type: com.example.springevent" \ - * -H "Ce-Source: spring.io/spring-event" \ - * -H "Content-Type: application/json" \ - * -H "Ce-Id: 0001" \ - * -d '{"releaseDate":"2004-03-24", "releaseName":"Spring Framework", "version":"1.0"}' - */ + @Bean public Function asPOJO() { - return v -> v.toString(); + return v -> { + System.out.println("Received POJO Cloud Event data: " + v); + return v.toString(); + }; } } diff --git a/spring-cloud-function-samples/function-sample-cloudevent/src/test/java/io/spring/cloudevent/CloudeventDemoApplicationTests.java b/spring-cloud-function-samples/function-sample-cloudevent/src/test/java/io/spring/cloudevent/CloudeventDemoApplicationRESTTests.java similarity index 87% rename from spring-cloud-function-samples/function-sample-cloudevent/src/test/java/io/spring/cloudevent/CloudeventDemoApplicationTests.java rename to spring-cloud-function-samples/function-sample-cloudevent/src/test/java/io/spring/cloudevent/CloudeventDemoApplicationRESTTests.java index 459d7e17e..1815e29b8 100644 --- a/spring-cloud-function-samples/function-sample-cloudevent/src/test/java/io/spring/cloudevent/CloudeventDemoApplicationTests.java +++ b/spring-cloud-function-samples/function-sample-cloudevent/src/test/java/io/spring/cloudevent/CloudeventDemoApplicationRESTTests.java @@ -47,7 +47,7 @@ import org.springframework.util.SocketUtils; * @author Oleg Zhurakousky * */ -public class CloudeventDemoApplicationTests { +public class CloudeventDemoApplicationRESTTests { private TestRestTemplate testRestTemplate = new TestRestTemplate(); @@ -167,6 +167,37 @@ public class CloudeventDemoApplicationTests { assertThat(response.getBody()).isEqualTo("releaseDate:24-03-2004; releaseName:Spring Framework; version:1.0"); } + @Test + public void testAsStracturalFormatToString() throws Exception { + SpringApplication.run(CloudeventDemoApplication.class); + + String payload = "{\n" + + " \"specversion\" : \"1.0\",\n" + + " \"type\" : \"org.springframework\",\n" + + " \"source\" : \"https://spring.io/\",\n" + + " \"id\" : \"A234-1234-1234\",\n" + + " \"datacontenttype\" : \"application/json\",\n" + + " \"data\" : {\n" + + " \"version\" : \"1.0\",\n" + + " \"releaseName\" : \"Spring Framework\",\n" + + " \"releaseDate\" : \"24-03-2004\"\n" + + " }\n" + + "}"; + + HttpHeaders headers = new HttpHeaders(); + headers.setContentType(MediaType.valueOf("application/cloudevents+json;charset=utf-8")); + + RequestEntity re = new RequestEntity<>(payload, headers, HttpMethod.POST, this.constructURI("/asStringMessage")); + ResponseEntity response = testRestTemplate.exchange(re, String.class); + + assertThat(response.getBody()).isEqualTo(payload); + + re = new RequestEntity<>(payload, headers, HttpMethod.POST, this.constructURI("/asString")); + response = testRestTemplate.exchange(re, String.class); + + assertThat(response.getBody()).isEqualTo(payload); + } + @Configuration public static class FooBarConverterConfiguration { diff --git a/spring-cloud-function-samples/function-sample-cloudevent/src/test/java/io/spring/cloudevent/CloudeventDemoApplicationStreamTests.java b/spring-cloud-function-samples/function-sample-cloudevent/src/test/java/io/spring/cloudevent/CloudeventDemoApplicationStreamTests.java new file mode 100644 index 000000000..b976f3ff6 --- /dev/null +++ b/spring-cloud-function-samples/function-sample-cloudevent/src/test/java/io/spring/cloudevent/CloudeventDemoApplicationStreamTests.java @@ -0,0 +1,27 @@ +/* + * Copyright 2020-2020 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package io.spring.cloudevent; + +/** + * + * @author Oleg Zhurakousky + * + */ +public class CloudeventDemoApplicationStreamTests { + + +}