Skip to content

Commit

Permalink
chore: Add data-type-action Kamelet
Browse files Browse the repository at this point in the history
- Adds new action Kamelet to leverage Kamelet input/ouput data types
- Enables users to apply data type conversion as part of a KameletBinding
- Introduces new data type implementation representing the CloudEvents Http binding
- Add some YAKS tests to verify data-type-action Kamelet
- Use Kamelets from local directory when running YAKS tests in GitHub workflow
  • Loading branch information
christophd committed Mar 2, 2023
1 parent c403c39 commit 0b3a9bb
Show file tree
Hide file tree
Showing 28 changed files with 843 additions and 28 deletions.
8 changes: 6 additions & 2 deletions .github/workflows/yaks-tests.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -42,7 +42,7 @@ concurrency:

env:
YAKS_VERSION: 0.14.2
YAKS_RUN_OPTIONS: "--timeout=15m --local -e YAKS_CAMELK_MAX_ATTEMPTS=10 -e YAKS_JBANG_CAMEL_VERSION=4.0.0-M1"
YAKS_RUN_OPTIONS: "--timeout=15m --local -e YAKS_CAMELK_MAX_ATTEMPTS=10 -e YAKS_JBANG_CAMEL_VERSION=4.0.0-M1 -e YAKS_JBANG_KAMELETS_LOCAL_DIR=../../kamelets"

jobs:
test:
Expand All @@ -58,6 +58,9 @@ jobs:
distribution: 'temurin'
java-version: 17
cache: 'maven'
- name: Build Kamelet libraries
run: |
./mvnw clean install -DskipTests
- name: Get YAKS CLI
run: |
curl --fail -L --silent https://github.com/citrusframework/yaks/releases/download/v${YAKS_VERSION}/yaks-${YAKS_VERSION}-linux-64bit.tar.gz -o yaks.tar.gz
Expand Down Expand Up @@ -93,11 +96,12 @@ jobs:
yaks run test/mail-sink $YAKS_RUN_OPTIONS
yaks run test/timer-source $YAKS_RUN_OPTIONS
yaks run test/timer-to-http $YAKS_RUN_OPTIONS
yaks run test/data-type-action $YAKS_RUN_OPTIONS
yaks run test/earthquake-source $YAKS_RUN_OPTIONS
yaks run test/rest-openapi-sink $YAKS_RUN_OPTIONS
yaks run test/kafka $YAKS_RUN_OPTIONS
- uses: actions/upload-artifact@v2
if: failure()
with:
name: dumps
path: test/**/.yaks-jbang/*-output.txt
1 change: 1 addition & 0 deletions docs/modules/ROOT/nav.adoc
Original file line number Diff line number Diff line change
Expand Up @@ -51,6 +51,7 @@
* xref:chunk-template-action.adoc[]
* xref:couchbase-sink.adoc[]
* xref:cron-source.adoc[]
* xref:data-type-action.adoc[]
* xref:delay-action.adoc[]
* xref:dns-dig-action.adoc[]
* xref:dns-ip-action.adoc[]
Expand Down
70 changes: 70 additions & 0 deletions kamelets/data-type-action.kamelet.yaml
Original file line number Diff line number Diff line change
@@ -0,0 +1,70 @@
# ---------------------------------------------------------------------------
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements. See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You 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
#
# http://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.
# ---------------------------------------------------------------------------
apiVersion: camel.apache.org/v1alpha1
kind: Kamelet
metadata:
name: data-type-action
annotations:
camel.apache.org/kamelet.support.level: "Stable"
camel.apache.org/catalog.version: "4.0.0-SNAPSHOT"
camel.apache.org/kamelet.icon: "data:image/svg+xml;base64,PD94bWwgdmVyc2lvbj0iMS4wIiBlbmNvZGluZz0iVVRGLTgiIHN0YW5kYWxvbmU9Im5vIj8+CjxzdmcKICAgeG1sbnM6ZGM9Imh0dHA6Ly9wdXJsLm9yZy9kYy9lbGVtZW50cy8xLjEvIgogICB4bWxuczpjYz0iaHR0cDovL2NyZWF0aXZlY29tbW9ucy5vcmcvbnMjIgogICB4bWxuczpyZGY9Imh0dHA6Ly93d3cudzMub3JnLzE5OTkvMDIvMjItcmRmLXN5bnRheC1ucyMiCiAgIHhtbG5zOnN2Zz0iaHR0cDovL3d3dy53My5vcmcvMjAwMC9zdmciCiAgIHhtbG5zPSJodHRwOi8vd3d3LnczLm9yZy8yMDAwL3N2ZyIKICAgeG1sbnM6c29kaXBvZGk9Imh0dHA6Ly9zb2RpcG9kaS5zb3VyY2Vmb3JnZS5uZXQvRFREL3NvZGlwb2RpLTAuZHRkIgogICB4bWxuczppbmtzY2FwZT0iaHR0cDovL3d3dy5pbmtzY2FwZS5vcmcvbmFtZXNwYWNlcy9pbmtzY2FwZSIKICAgdmlld0JveD0iMCAtMjU2IDE3OTIgMTc5MiIKICAgaWQ9InN2ZzMwMjUiCiAgIHZlcnNpb249IjEuMSIKICAgaW5rc2NhcGU6dmVyc2lvbj0iMC40OC4zLjEgcjk4ODYiCiAgIHdpZHRoPSIxMDAlIgogICBoZWlnaHQ9IjEwMCUiCiAgIHNvZGlwb2RpOmRvY25hbWU9ImNvZ19mb250X2F3ZXNvbWUuc3ZnIj4KICA8bWV0YWRhdGEKICAgICBpZD0ibWV0YWRhdGEzMDM1Ij4KICAgIDxyZGY6UkRGPgogICAgICA8Y2M6V29yawogICAgICAgICByZGY6YWJvdXQ9IiI+CiAgICAgICAgPGRjOmZvcm1hdD5pbWFnZS9zdmcreG1sPC9kYzpmb3JtYXQ+CiAgICAgICAgPGRjOnR5cGUKICAgICAgICAgICByZGY6cmVzb3VyY2U9Imh0dHA6Ly9wdXJsLm9yZy9kYy9kY21pdHlwZS9TdGlsbEltYWdlIiAvPgogICAgICA8L2NjOldvcms+CiAgICA8L3JkZjpSREY+CiAgPC9tZXRhZGF0YT4KICA8ZGVmcwogICAgIGlkPSJkZWZzMzAzMyIgLz4KICA8c29kaXBvZGk6bmFtZWR2aWV3CiAgICAgcGFnZWNvbG9yPSIjZmZmZmZmIgogICAgIGJvcmRlcmNvbG9yPSIjNjY2NjY2IgogICAgIGJvcmRlcm9wYWNpdHk9IjEiCiAgICAgb2JqZWN0dG9sZXJhbmNlPSIxMCIKICAgICBncmlkdG9sZXJhbmNlPSIxMCIKICAgICBndWlkZXRvbGVyYW5jZT0iMTAiCiAgICAgaW5rc2NhcGU6cGFnZW9wYWNpdHk9IjAiCiAgICAgaW5rc2NhcGU6cGFnZXNoYWRvdz0iMiIKICAgICBpbmtzY2FwZTp3aW5kb3ctd2lkdGg9IjY0MCIKICAgICBpbmtzY2FwZTp3aW5kb3ctaGVpZ2h0PSI0ODAiCiAgICAgaWQ9Im5hbWVkdmlldzMwMzEiCiAgICAgc2hvd2dyaWQ9ImZhbHNlIgogICAgIGlua3NjYXBlOnpvb209IjAuMTMxNjk2NDMiCiAgICAgaW5rc2NhcGU6Y3g9Ijg5NiIKICAgICBpbmtzY2FwZTpjeT0iODk2IgogICAgIGlua3NjYXBlOndpbmRvdy14PSIwIgogICAgIGlua3NjYXBlOndpbmRvdy15PSIyNSIKICAgICBpbmtzY2FwZTp3aW5kb3ctbWF4aW1pemVkPSIwIgogICAgIGlua3NjYXBlOmN1cnJlbnQtbGF5ZXI9InN2ZzMwMjUiIC8+CiAgPGcKICAgICB0cmFuc2Zvcm09Im1hdHJpeCgxLDAsMCwtMSwxMjEuNDkxNTMsMTI4NS40MjM3KSIKICAgICBpZD0iZzMwMjciPgogICAgPHBhdGgKICAgICAgIGQ9Im0gMTAyNCw2NDAgcSAwLDEwNiAtNzUsMTgxIC03NSw3NSAtMTgxLDc1IC0xMDYsMCAtMTgxLC03NSAtNzUsLTc1IC03NSwtMTgxIDAsLTEwNiA3NSwtMTgxIDc1LC03NSAxODEsLTc1IDEwNiwwIDE4MSw3NSA3NSw3NSA3NSwxODEgeiBtIDUxMiwxMDkgViA1MjcgcSAwLC0xMiAtOCwtMjMgLTgsLTExIC0yMCwtMTMgbCAtMTg1LC0yOCBxIC0xOSwtNTQgLTM5LC05MSAzNSwtNTAgMTA3LC0xMzggMTAsLTEyIDEwLC0yNSAwLC0xMyAtOSwtMjMgLTI3LC0zNyAtOTksLTEwOCAtNzIsLTcxIC05NCwtNzEgLTEyLDAgLTI2LDkgbCAtMTM4LDEwOCBxIC00NCwtMjMgLTkxLC0zOCAtMTYsLTEzNiAtMjksLTE4NiAtNywtMjggLTM2LC0yOCBIIDY1NyBxIC0xNCwwIC0yNC41LDguNSBRIDYyMiwtMTExIDYyMSwtOTggTCA1OTMsODYgcSAtNDksMTYgLTkwLDM3IEwgMzYyLDE2IFEgMzUyLDcgMzM3LDcgMzIzLDcgMzEyLDE4IDE4NiwxMzIgMTQ3LDE4NiBxIC03LDEwIC03LDIzIDAsMTIgOCwyMyAxNSwyMSA1MSw2Ni41IDM2LDQ1LjUgNTQsNzAuNSAtMjcsNTAgLTQxLDk5IEwgMjksNDk1IFEgMTYsNDk3IDgsNTA3LjUgMCw1MTggMCw1MzEgdiAyMjIgcSAwLDEyIDgsMjMgOCwxMSAxOSwxMyBsIDE4NiwyOCBxIDE0LDQ2IDM5LDkyIC00MCw1NyAtMTA3LDEzOCAtMTAsMTIgLTEwLDI0IDAsMTAgOSwyMyAyNiwzNiA5OC41LDEwNy41IDcyLjUsNzEuNSA5NC41LDcxLjUgMTMsMCAyNiwtMTAgbCAxMzgsLTEwNyBxIDQ0LDIzIDkxLDM4IDE2LDEzNiAyOSwxODYgNywyOCAzNiwyOCBoIDIyMiBxIDE0LDAgMjQuNSwtOC41IFEgOTE0LDEzOTEgOTE1LDEzNzggbCAyOCwtMTg0IHEgNDksLTE2IDkwLC0zNyBsIDE0MiwxMDcgcSA5LDkgMjQsOSAxMywwIDI1LC0xMCAxMjksLTExOSAxNjUsLTE3MCA3LC04IDcsLTIyIDAsLTEyIC04LC0yMyAtMTUsLTIxIC01MSwtNjYuNSAtMzYsLTQ1LjUgLTU0LC03MC41IDI2LC01MCA0MSwtOTggbCAxODMsLTI4IHEgMTMsLTIgMjEsLTEyLjUgOCwtMTAuNSA4LC0yMy41IHoiCiAgICAgICBpZD0icGF0aDMwMjkiCiAgICAgICBpbmtzY2FwZTpjb25uZWN0b3ItY3VydmF0dXJlPSIwIgogICAgICAgc3R5bGU9ImZpbGw6Y3VycmVudENvbG9yIiAvPgogIDwvZz4KPC9zdmc+Cg=="
camel.apache.org/provider: "Apache Software Foundation"
camel.apache.org/kamelet.group: "Actions"
camel.apache.org/kamelet.namespace: "Transformation"
labels:
camel.apache.org/kamelet.type: "action"
spec:
definition:
title: "Data Type Action"
description: |-
Applies a given data type with respective data transformation.
required:
- format
type: object
properties:
scheme:
title: Component Scheme
description: The data type component scheme enables users to apply Camel component specific data type conversions.
type: string
default: "camel"
example: "camel"
format:
title: Data Type Format
description: Defines the data type that will be applied by this action. The Kameelet catalog supports different data types and performs automatic message conversion according to the given type.
type: string
dependencies:
- "mvn:org.apache.camel.kamelets:camel-kamelets-utils:4.0.0-SNAPSHOT"
- "camel:kamelet"
- "camel:core"
template:
beans:
- name: dataTypeRegistry
type: "#class:org.apache.camel.kamelets.utils.format.DefaultDataTypeRegistry"
- name: dataTypeProcessor
type: "#class:org.apache.camel.kamelets.utils.format.DataTypeProcessor"
property:
- key: scheme
value: '{{scheme}}'
- key: format
value: '{{format}}'
- key: registry
value: '#bean:{{dataTypeRegistry}}'
from:
uri: "kamelet:source"
steps:
- process:
ref: "{{dataTypeProcessor}}"
Original file line number Diff line number Diff line change
Expand Up @@ -17,14 +17,11 @@

package org.apache.camel.kamelets.utils.format.converter.aws2.s3;

import java.time.Instant;
import java.time.ZoneId;
import java.time.ZonedDateTime;
import java.time.format.DateTimeFormatter;
import java.util.Map;

import org.apache.camel.Exchange;
import org.apache.camel.component.aws2.s3.AWS2S3Constants;
import org.apache.camel.kamelets.utils.format.converter.utils.CloudEvents;
import org.apache.camel.kamelets.utils.format.spi.DataTypeConverter;
import org.apache.camel.kamelets.utils.format.spi.annotations.DataType;

Expand All @@ -35,24 +32,14 @@
@DataType(scheme = "aws2-s3", name = "cloudevents", mediaType = "application/octet-stream")
public class AWS2S3CloudEventOutputType implements DataTypeConverter {

static final String CAMEL_CLOUD_EVENT_TYPE = "CamelCloudEventType";
static final String CAMEL_CLOUD_EVENT_SOURCE = "CamelCloudEventSource";
static final String CAMEL_CLOUD_EVENT_SUBJECT = "CamelCloudEventSubject";
static final String CAMEL_CLOUD_EVENT_TIME = "CamelCloudEventTime";

@Override
public void convert(Exchange exchange) {
final Map<String, Object> headers = exchange.getMessage().getHeaders();

headers.put(CAMEL_CLOUD_EVENT_TYPE, "org.apache.camel.event.aws.s3.getObject");
headers.put(CAMEL_CLOUD_EVENT_SOURCE, "aws.s3.bucket." + exchange.getMessage().getHeader(AWS2S3Constants.BUCKET_NAME, String.class));
headers.put(CAMEL_CLOUD_EVENT_SUBJECT, exchange.getMessage().getHeader(AWS2S3Constants.KEY, String.class));
headers.put(CAMEL_CLOUD_EVENT_TIME, getEventTime(exchange));
}

private String getEventTime(Exchange exchange) {
final ZonedDateTime created
= ZonedDateTime.ofInstant(Instant.ofEpochMilli(exchange.getCreated()), ZoneId.systemDefault());
return DateTimeFormatter.ISO_INSTANT.format(created);
headers.put(CloudEvents.CAMEL_CLOUD_EVENT_ID, exchange.getExchangeId());
headers.put(CloudEvents.CAMEL_CLOUD_EVENT_TYPE, "org.apache.camel.event.aws.s3.getObject");
headers.put(CloudEvents.CAMEL_CLOUD_EVENT_SOURCE, "aws.s3.bucket." + exchange.getMessage().getHeader(AWS2S3Constants.BUCKET_NAME, String.class));
headers.put(CloudEvents.CAMEL_CLOUD_EVENT_SUBJECT, exchange.getMessage().getHeader(AWS2S3Constants.KEY, String.class));
headers.put(CloudEvents.CAMEL_CLOUD_EVENT_TIME, CloudEvents.getEventTime(exchange));
}
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,52 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You 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
*
* http://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 org.apache.camel.kamelets.utils.format.converter.http;

import java.util.Map;

import org.apache.camel.Exchange;
import org.apache.camel.kamelets.utils.format.converter.utils.CloudEvents;
import org.apache.camel.kamelets.utils.format.spi.DataTypeConverter;
import org.apache.camel.kamelets.utils.format.spi.annotations.DataType;

/**
* Output data type represents the CloudEvent V1 Http binding. The data type reads Camel specific
* CloudEvent headers and transforms these to Http headers according to the CloudEvents Http binding specification.
*
* By default, sets the Http content type header to application/json when not set explicitly.
*/
@DataType(scheme = "http", name = "cloudevents", mediaType = "application/json")
public class HttpCloudEventOutputType implements DataTypeConverter {

@Override
public void convert(Exchange exchange) {
final Map<String, Object> headers = exchange.getMessage().getHeaders();

headers.put("ce-id", exchange.getExchangeId());
headers.put("ce-specversion", headers.getOrDefault(CloudEvents.CAMEL_CLOUD_EVENT_VERSION, "1.0"));
headers.put("ce-type", headers.getOrDefault(CloudEvents.CAMEL_CLOUD_EVENT_TYPE, "org.apache.camel.event"));
headers.put("ce-source", headers.getOrDefault(CloudEvents.CAMEL_CLOUD_EVENT_SOURCE, "org.apache.camel"));

if (headers.containsKey(CloudEvents.CAMEL_CLOUD_EVENT_SUBJECT)) {
headers.put("ce-subject", headers.get(CloudEvents.CAMEL_CLOUD_EVENT_SUBJECT));
}

headers.put("ce-time", headers.getOrDefault(CloudEvents.CAMEL_CLOUD_EVENT_TIME, CloudEvents.getEventTime(exchange)));
headers.put(Exchange.CONTENT_TYPE, headers.getOrDefault(CloudEvents.CAMEL_CLOUD_EVENT_CONTENT_TYPE, "application/json"));
}
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,45 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You 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
*
* http://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 org.apache.camel.kamelets.utils.format.converter.utils;

import java.time.Instant;
import java.time.ZoneId;
import java.time.ZonedDateTime;
import java.time.format.DateTimeFormatter;

import org.apache.camel.Exchange;

/**
* Helper class to manage CloudEvents specific Camel message headers and other utilities.
*/
public class CloudEvents {

public static final String CAMEL_CLOUD_EVENT_ID = "CamelCloudEventID";
public static final String CAMEL_CLOUD_EVENT_VERSION = "CamelCloudEventVersion";
public static final String CAMEL_CLOUD_EVENT_TYPE = "CamelCloudEventType";
public static final String CAMEL_CLOUD_EVENT_SOURCE = "CamelCloudEventSource";
public static final String CAMEL_CLOUD_EVENT_SUBJECT = "CamelCloudEventSubject";
public static final String CAMEL_CLOUD_EVENT_TIME = "CamelCloudEventTime";
public static final String CAMEL_CLOUD_EVENT_CONTENT_TYPE = Exchange.CONTENT_TYPE;

public static String getEventTime(Exchange exchange) {
final ZonedDateTime created
= ZonedDateTime.ofInstant(Instant.ofEpochMilli(exchange.getCreated()), ZoneId.systemDefault());
return DateTimeFormatter.ISO_INSTANT.format(created);
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -18,3 +18,4 @@
org.apache.camel.kamelets.utils.format.converter.standard
org.apache.camel.kamelets.utils.format.converter.aws2.s3
org.apache.camel.kamelets.utils.format.converter.aws2.ddb
org.apache.camel.kamelets.utils.format.converter.http
Original file line number Diff line number Diff line change
@@ -0,0 +1,18 @@
#
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements. See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You 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
#
# http://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.
#

class=org.apache.camel.kamelets.utils.format.converter.http.HttpCloudEventOutputType
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,7 @@
import org.apache.camel.component.aws2.s3.AWS2S3Constants;
import org.apache.camel.impl.DefaultCamelContext;
import org.apache.camel.kamelets.utils.format.DefaultDataTypeRegistry;
import org.apache.camel.kamelets.utils.format.converter.utils.CloudEvents;
import org.apache.camel.kamelets.utils.format.spi.DataTypeConverter;
import org.apache.camel.support.DefaultExchange;
import org.junit.jupiter.api.Assertions;
Expand All @@ -52,9 +53,9 @@ void shouldMapToCloudEvent() throws Exception {

Assertions.assertTrue(exchange.getMessage().hasHeaders());
Assertions.assertTrue(exchange.getMessage().getHeaders().containsKey(AWS2S3Constants.KEY));
assertEquals("org.apache.camel.event.aws.s3.getObject", exchange.getMessage().getHeader(AWS2S3CloudEventOutputType.CAMEL_CLOUD_EVENT_TYPE));
assertEquals("test1.txt", exchange.getMessage().getHeader(AWS2S3CloudEventOutputType.CAMEL_CLOUD_EVENT_SUBJECT));
assertEquals("aws.s3.bucket.myBucket", exchange.getMessage().getHeader(AWS2S3CloudEventOutputType.CAMEL_CLOUD_EVENT_SOURCE));
assertEquals("org.apache.camel.event.aws.s3.getObject", exchange.getMessage().getHeader(CloudEvents.CAMEL_CLOUD_EVENT_TYPE));
assertEquals("test1.txt", exchange.getMessage().getHeader(CloudEvents.CAMEL_CLOUD_EVENT_SUBJECT));
assertEquals("aws.s3.bucket.myBucket", exchange.getMessage().getHeader(CloudEvents.CAMEL_CLOUD_EVENT_SOURCE));
}

@Test
Expand All @@ -64,4 +65,4 @@ public void shouldLookupDataType() throws Exception {
Optional<DataTypeConverter> converter = dataTypeRegistry.lookup("aws2-s3", "cloudevents");
Assertions.assertTrue(converter.isPresent());
}
}
}
Loading

0 comments on commit 0b3a9bb

Please sign in to comment.