You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by ac...@apache.org on 2020/11/26 12:59:56 UTC
[camel-kafka-connector] 02/03: Move SshTransforms from example/it
test class directly in the camel-ssh-kafka-connector,
removed the transforms from itest
This is an automated email from the ASF dual-hosted git repository.
acosentino pushed a commit to branch ssh-transforms
in repository https://gitbox.apache.org/repos/asf/camel-kafka-connector.git
commit c119916712d52e69fe21f35a5239ba3b0864daac
Author: Andrea Cosentino <an...@gmail.com>
AuthorDate: Thu Nov 26 13:40:36 2020 +0100
Move SshTransforms from example/it test class directly in the camel-ssh-kafka-connector, removed the transforms from itest
---
tests/itests-ssh/pom.xml | 6 --
.../ssh/source/CamelSourceSshITCase.java | 2 +-
.../kafkaconnector/ssh/source/SshTransforms.java | 76 ----------------------
3 files changed, 1 insertion(+), 83 deletions(-)
diff --git a/tests/itests-ssh/pom.xml b/tests/itests-ssh/pom.xml
index 087df1a..b2d2388 100644
--- a/tests/itests-ssh/pom.xml
+++ b/tests/itests-ssh/pom.xml
@@ -41,12 +41,6 @@
<groupId>org.apache.camel</groupId>
<artifactId>camel-ssh</artifactId>
</dependency>
-
- <dependency>
- <groupId>commons-io</groupId>
- <artifactId>commons-io</artifactId>
- <version>${commons-io-version}</version>
- </dependency>
</dependencies>
<build>
diff --git a/tests/itests-ssh/src/test/java/org/apache/camel/kafkaconnector/ssh/source/CamelSourceSshITCase.java b/tests/itests-ssh/src/test/java/org/apache/camel/kafkaconnector/ssh/source/CamelSourceSshITCase.java
index 8cd9abf..6316e01 100644
--- a/tests/itests-ssh/src/test/java/org/apache/camel/kafkaconnector/ssh/source/CamelSourceSshITCase.java
+++ b/tests/itests-ssh/src/test/java/org/apache/camel/kafkaconnector/ssh/source/CamelSourceSshITCase.java
@@ -76,7 +76,7 @@ public class CamelSourceSshITCase extends AbstractKafkaTest {
ConnectorPropertyFactory connectorPropertyFactory = CamelSshPropertyFactory.basic().withKafkaTopic(topic).withHost(sshService.getSshHost())
.withPort(Integer.toString(sshService.getSshPort())).withDelay(Integer.toString(10000)).withUsername("root").withPassword("root").withPollcommand("date")
- .withTransformsConfig("SshTransforms").withEntry("type", "org.apache.camel.kafkaconnector.ssh.source.SshTransforms").end();
+ .withTransformsConfig("SshTransforms").withEntry("type", "org.apache.camel.kafkaconnector.ssh.transformers.SshTransforms").end();
runTest(connectorPropertyFactory);
}
diff --git a/tests/itests-ssh/src/test/java/org/apache/camel/kafkaconnector/ssh/source/SshTransforms.java b/tests/itests-ssh/src/test/java/org/apache/camel/kafkaconnector/ssh/source/SshTransforms.java
deleted file mode 100644
index 6097d75..0000000
--- a/tests/itests-ssh/src/test/java/org/apache/camel/kafkaconnector/ssh/source/SshTransforms.java
+++ /dev/null
@@ -1,76 +0,0 @@
-/*
- * 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.kafkaconnector.ssh.source;
-
-import java.io.ByteArrayInputStream;
-import java.io.IOException;
-import java.nio.charset.Charset;
-import java.util.Map;
-
-import org.apache.camel.kafkaconnector.utils.SchemaHelper;
-import org.apache.commons.io.IOUtils;
-import org.apache.kafka.common.config.ConfigDef;
-import org.apache.kafka.connect.connector.ConnectRecord;
-import org.apache.kafka.connect.transforms.Transformation;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class SshTransforms<R extends ConnectRecord<R>> implements Transformation<R> {
- public static final String FIELD_KEY_CONFIG = "key";
- public static final ConfigDef CONFIG_DEF = new ConfigDef().define(FIELD_KEY_CONFIG, ConfigDef.Type.STRING, null, ConfigDef.Importance.MEDIUM,
- "Transforms String-based content from Kafka into a map");
-
- private static final Logger LOG = LoggerFactory.getLogger(SshTransforms.class);
-
- @Override
- public R apply(R r) {
- Object value = r.value();
-
- if (r.value() instanceof ByteArrayInputStream) {
- LOG.debug("Converting record from Ssh Body Result to text");
- ByteArrayInputStream message = (ByteArrayInputStream)r.value();
- String m = null;
- try {
- m = IOUtils.toString(message, Charset.defaultCharset());
- } catch (IOException e) {
- e.printStackTrace();
- }
-
- return r.newRecord(r.topic(), r.kafkaPartition(), null, r.key(), SchemaHelper.buildSchemaBuilderForType(m), m, r.timestamp());
-
- } else {
- LOG.debug("Unexpected message type: {}", r.value().getClass());
-
- return r;
- }
- }
-
- @Override
- public ConfigDef config() {
- return CONFIG_DEF;
- }
-
- @Override
- public void close() {
-
- }
-
- @Override
- public void configure(Map<String, ?> map) {
-
- }
-}