This is an automated email from the ASF dual-hosted git repository.
acosentino pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel-kamelets.git
The following commit(s) were added to refs/heads/main by this push:
new d096fd2 Message Timestamp Router added as bean in library
d096fd2 is described below
commit d096fd24e6d91a1a7d79ac05532ac02e04642885
Author: Andrea Cosentino <[email protected]>
AuthorDate: Tue May 18 16:41:21 2021 +0200
Message Timestamp Router added as bean in library
---
.../utils/transform/MessageTimestampRouter.java | 90 ++++++++++++++++++++++
1 file changed, 90 insertions(+)
diff --git
a/library/camel-kamelets-utils/src/main/java/org/apache/camel/kamelets/utils/transform/MessageTimestampRouter.java
b/library/camel-kamelets-utils/src/main/java/org/apache/camel/kamelets/utils/transform/MessageTimestampRouter.java
new file mode 100644
index 0000000..234350a
--- /dev/null
+++
b/library/camel-kamelets-utils/src/main/java/org/apache/camel/kamelets/utils/transform/MessageTimestampRouter.java
@@ -0,0 +1,90 @@
+/*
+ * 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.transform;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.camel.Exchange;
+import org.apache.camel.ExchangeProperty;
+import org.apache.camel.component.kafka.KafkaConstants;
+import org.apache.camel.util.ObjectHelper;
+
+import java.text.SimpleDateFormat;
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Date;
+import java.util.List;
+import java.util.Map;
+import java.util.TimeZone;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+import java.util.stream.Collectors;
+
+public class MessageTimestampRouter {
+
+ public void process(@ExchangeProperty("topicFormat") String topicFormat,
@ExchangeProperty("timestampFormat") String timestampFormat,
@ExchangeProperty("timestampKeys") String timestampKeys,
@ExchangeProperty("timestampKeyFormat") String timestampKeyFormat, Exchange ex)
{
+ final Pattern TOPIC = Pattern.compile("$[topic]", Pattern.LITERAL);
+
+ final Pattern TIMESTAMP = Pattern.compile("$[timestamp]",
Pattern.LITERAL);
+
+ final SimpleDateFormat fmt = new SimpleDateFormat(timestampFormat);
+ fmt.setTimeZone(TimeZone.getTimeZone("UTC"));
+
+ ObjectMapper mapper = new ObjectMapper();
+ List<String> splittedKeys = new ArrayList<>();
+ JsonNode jsonNodeBody = ex.getMessage().getBody(JsonNode.class);
+ Map<Object, Object> body = mapper.convertValue(jsonNodeBody, new
TypeReference<Map<Object, Object>>(){});
+ if (ObjectHelper.isNotEmpty(timestampKeys)) {
+ splittedKeys =
Arrays.stream(timestampKeys.split(",")).collect(Collectors.toList());
+ }
+
+ String rawTimestamp = null;
+ String topicName = ex.getMessage().getHeader(KafkaConstants.TOPIC,
String.class);
+ for (String key:
+ splittedKeys) {
+ if (ObjectHelper.isNotEmpty(key)) {
+ rawTimestamp = (String) body.get(key);
+ break;
+ }
+ }
+ String timestamp;
+ if (ObjectHelper.isNotEmpty(timestampKeyFormat) &&
ObjectHelper.isNotEmpty(rawTimestamp)) {
+ final SimpleDateFormat timestampKeyFmt = new
SimpleDateFormat(timestampKeyFormat);
+ fmt.setTimeZone(TimeZone.getTimeZone("UTC"));
+ timestamp = timestampKeyFmt.format(new Date(rawTimestamp));
+ } else {
+ timestamp = rawTimestamp;
+ }
+ if (ObjectHelper.isNotEmpty(timestamp)) {
+ final String formattedTimestamp = fmt.format(new Date(timestamp));
+ String replace1;
+ String updatedTopic;
+
+ if (ObjectHelper.isNotEmpty(topicName)) {
+ replace1 =
TOPIC.matcher(topicFormat).replaceAll(Matcher.quoteReplacement(topicName));
+ updatedTopic =
TIMESTAMP.matcher(replace1).replaceAll(Matcher.quoteReplacement(formattedTimestamp));
+ } else {
+ replace1 =
TOPIC.matcher(topicFormat).replaceAll(Matcher.quoteReplacement(""));
+ updatedTopic =
TIMESTAMP.matcher(replace1).replaceAll(Matcher.quoteReplacement(formattedTimestamp));
+ }
+ ex.getMessage().setHeader(KafkaConstants.OVERRIDE_TOPIC,
updatedTopic);
+ }
+ }
+
+}
