Catalog: expose an API for consuming connectors catalog metadata and options

camel-kafka-connector-catalog/pom.xml

@@ -44,6 +44,40 @@
             <artifactId>camel-tooling-model</artifactId>
             <version>${camel.version}</version>
         </dependency>
+        <dependency>
+            <groupId>org.apache.camel.kafkaconnector</groupId>
+            <artifactId>camel-kafka-connector-model</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.camel</groupId>
+            <artifactId>camel-core-catalog</artifactId>
+            <version>${camel.version}</version>
+        </dependency>
+        <!-- logging -->
+        <dependency>
+            <groupId>org.apache.logging.log4j</groupId>
+            <artifactId>log4j-api</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.logging.log4j</groupId>
+            <artifactId>log4j-core</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.logging.log4j</groupId>
+            <artifactId>log4j-slf4j-impl</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>org.junit.jupiter</groupId>
+            <artifactId>junit-jupiter-api</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.junit.jupiter</groupId>
+            <artifactId>junit-jupiter-engine</artifactId>
+            <scope>test</scope>
+        </dependency>
 
     </dependencies>
 

camel-kafka-connector-catalog/src/main/java/org/apache/camel/kafkaconnector/catalog/CamelKafkaConnectorCatalog.java

@@ -0,0 +1,138 @@
+/*
+ * 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.catalog;
+
+import java.io.BufferedReader;
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.nio.file.Files;
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import org.apache.camel.kafkaconnector.model.CamelKafkaConnectorModel;
+import org.apache.camel.kafkaconnector.model.CamelKafkaConnectorOptionModel;
+import org.apache.camel.tooling.model.JsonMapper;
+import org.apache.camel.util.json.JsonObject;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class CamelKafkaConnectorCatalog {
+
+    static List<String> connectorsName = new ArrayList<String>();
+    static Map<String, CamelKafkaConnectorModel> connectorsModel = new HashMap<String, CamelKafkaConnectorModel>();
+    private static final Logger LOG = LoggerFactory.getLogger(CamelKafkaConnectorCatalog.class);
+    private static final String CONNECTORS_DIR = "src/generated/resources/connectors";
+    private static final String DESCRIPTORS_DIR = "src/generated/resources/descriptors";
+    private static final String CONNECTORS_PROPERTIES = "connectors.properties";
+
+    public CamelKafkaConnectorCatalog() {
+        initCatalog();
+        generateModel();
+    }
+
+    private void generateModel() {
+        for (String connector : connectorsName) {
+            connectorsModel.put(connector, getConnectorModel(connector));
+        }
+    }
+
+    private void initCatalog() {
+        try (FileInputStream input = new FileInputStream(DESCRIPTORS_DIR + File.separator + CONNECTORS_PROPERTIES)) {
+
+            BufferedReader reader = new BufferedReader(new InputStreamReader(input));
+
+            while (reader.ready()) {
+                String connector = reader.readLine();
+                if (connector.equalsIgnoreCase("camel-coap-tcp-source")) {
+                    connectorsName.add("camel-coap+tcp-source");
+                } else if (connector.equalsIgnoreCase("camel-coaps-tcp-source")) {
+                    connectorsName.add("camel-coaps+tcp-source");
+                } else if (connector.equalsIgnoreCase("camel-coaps-tcp-sink")) {
+                    connectorsName.add("camel-coaps+tcp-sink");
+                } else if (connector.equalsIgnoreCase("camel-coap-tcp-sink")) {
+                    connectorsName.add("camel-coap+tcp-sink");
+                } else {
+                    connectorsName.add(connector);
+                }
+            }
+        } catch (FileNotFoundException e) {
+            LOG.error("Cannot find file: {}", e.getMessage(), e);
+        } catch (IOException e) {
+            LOG.error("IO Exception: {}", e.getMessage(), e);
+        }
+    }
+
+    public String getConnectorAsJson(String connectorName) {
+        String result = null;
+        try {
+            result = Files.lines(Paths.get(CONNECTORS_DIR + File.separator + connectorName + ".json")).parallel() // for
+                                                                                                                        // parallel
+                                                                                                                        // processing
+                .map(String::trim) // to change line
+                .collect(Collectors.joining());
+        } catch (IOException e) {
+            LOG.error("IO Exception: {}", e.getMessage(), e);
+        }
+        return result;
+    }
+
+    private CamelKafkaConnectorModel getConnectorModel(String connectorName) {
+        CamelKafkaConnectorModel model = new CamelKafkaConnectorModel();
+        String json = getConnectorAsJson(connectorName);
+        JsonObject obj = JsonMapper.deserialize(json);
+        JsonObject wrapper = (JsonObject)obj.get("connector");
+        model.setConnectorClass((String)wrapper.get("class"));
+        model.setArtifactId((String)wrapper.get("artifactId"));
+        model.setGroupId((String)wrapper.get("groupId"));
+        model.setType((String)wrapper.get("type"));
+        model.setVersion((String)wrapper.get("version"));
+        model.setOptions((List<CamelKafkaConnectorOptionModel>)getConnectorOptionModel(obj));
+        return model;
+    }
+
+    private List<CamelKafkaConnectorOptionModel> getConnectorOptionModel(JsonObject obj) {
+        List<CamelKafkaConnectorOptionModel> model = new ArrayList<CamelKafkaConnectorOptionModel>();
+        JsonObject wrapper = (JsonObject)obj.get("properties");
+        Set<String> options = wrapper.keySet();
+        for (String string : options) {
+            JsonObject object = (JsonObject)wrapper.get(string);
+            CamelKafkaConnectorOptionModel singleModel = new CamelKafkaConnectorOptionModel();
+            singleModel.setDefaultValue((String)object.get("defaultValue"));
+            singleModel.setPriority((String)object.get("priority"));
+            singleModel.setDescription((String)object.get("description"));
+            singleModel.setName((String)object.get("name"));
+            model.add(singleModel);
+        }
+        return model;
+    }
+
+    public List<String> getConnectorsName() {
+        return connectorsName;
+    }
+
+    public static Map<String, CamelKafkaConnectorModel> getConnectorsModel() {
+        return connectorsModel;
+    }
+}

camel-kafka-connector-catalog/src/test/java/org/apache/camel/kafkaconnector/catalog/CamelKafkaConnectorCatalogTest.java

@@ -0,0 +1,58 @@
+/*
+ * 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.catalog;
+
+import java.util.List;
+import java.util.Map;
+
+import org.apache.camel.kafkaconnector.model.CamelKafkaConnectorModel;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class CamelKafkaConnectorCatalogTest {
+
+    static CamelKafkaConnectorCatalog catalog;
+
+    @BeforeAll
+    public static void createCamelCatalog() {
+        catalog = new CamelKafkaConnectorCatalog();
+    }
+
+    @Test
+    public void testConnectors() throws Exception {
+        List<String> list = catalog.getConnectorsName();
+        assertTrue(list.contains("camel-aws2-s3-sink"));
+        assertTrue(list.contains("camel-aws2-s3-source"));
+    }
+
+    @Test
+    public void testOptions() throws Exception {
+        Map<String, CamelKafkaConnectorModel> p = catalog.getConnectorsModel();
+        CamelKafkaConnectorModel model = p.get("camel-aws2-s3-sink");
+        assertEquals("org.apache.camel.kafkaconnector", model.getGroupId());
+        assertEquals("sink", model.getType());
+        assertEquals("org.apache.camel.kafkaconnector.aws2s3.CamelAws2s3SinkConnector", model.getConnectorClass());
+        assertEquals(61, model.getOptions().size());
+        assertEquals("camel.sink.path.bucketNameOrArn", model.getOptions().get(0).getName());
+        assertEquals("camel.sink.endpoint.amazonS3Client", model.getOptions().get(1).getName());
+        assertEquals("camel.sink.endpoint.autoCreateBucket", model.getOptions().get(2).getName());
+    }
+
+}