Add initial version to migrate generic adapters (#1289)

streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/connect/DescriptionResource.java

@@ -26,15 +26,16 @@
 import org.apache.streampipes.model.connect.adapter.AdapterDescription;
 import org.apache.streampipes.rest.shared.annotation.JacksonSerialized;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import jakarta.ws.rs.DELETE;
 import jakarta.ws.rs.GET;
 import jakarta.ws.rs.Path;
 import jakarta.ws.rs.PathParam;
 import jakarta.ws.rs.Produces;
 import jakarta.ws.rs.core.MediaType;
 import jakarta.ws.rs.core.Response;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 import java.util.List;
 import java.util.Optional;

streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/connect/RuntimeResolvableResource.java

@@ -29,15 +29,16 @@
 import org.apache.streampipes.model.runtime.RuntimeOptionsResponse;
 import org.apache.streampipes.rest.shared.annotation.JacksonSerialized;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import jakarta.ws.rs.Consumes;
 import jakarta.ws.rs.POST;
 import jakarta.ws.rs.Path;
 import jakarta.ws.rs.PathParam;
 import jakarta.ws.rs.Produces;
 import jakarta.ws.rs.core.MediaType;
 import jakarta.ws.rs.core.Response;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 
 @Path("/v2/connect/master/resolvable")

streampipes-service-core/src/main/java/org/apache/streampipes/service/core/migrations/v093/format/CsvFormatMigrator.java

@@ -0,0 +1,34 @@
+/*
+ * 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.streampipes.service.core.migrations.v093.format;
+
+import com.google.gson.JsonObject;
+
+public class CsvFormatMigrator implements FormatMigrator {
+
+
+  public CsvFormatMigrator(String csvFormatId,
+                           JsonObject formatDescription) {
+  }
+
+  @Override
+  public void migrate(JsonObject newFormatProperties) {
+
+  }
+}

streampipes-service-core/src/main/java/org/apache/streampipes/service/core/migrations/v093/format/EmptyFormatMigrator.java

@@ -0,0 +1,29 @@
+/*
+ * 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.streampipes.service.core.migrations.v093.format;
+
+import com.google.gson.JsonObject;
+
+public class EmptyFormatMigrator implements FormatMigrator {
+
+  @Override
+  public void migrate(JsonObject newFormatProperties) {
+
+  }
+}

streampipes-service-core/src/main/java/org/apache/streampipes/service/core/migrations/v093/format/FormatMigrator.java

@@ -0,0 +1,26 @@
+/*
+ * 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.streampipes.service.core.migrations.v093.format;
+
+import com.google.gson.JsonObject;
+
+public interface FormatMigrator {
+
+  void migrate(JsonObject newFormatProperties);
+}

streampipes-service-core/src/main/java/org/apache/streampipes/service/core/migrations/v093/format/JsonFormatMigrator.java

@@ -0,0 +1,51 @@
+/*
+ * 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.streampipes.service.core.migrations.v093.format;
+
+import org.apache.streampipes.service.core.migrations.v093.migrator.MigrationHelpers;
+
+import com.google.gson.JsonObject;
+import com.google.gson.JsonPrimitive;
+
+import static org.apache.streampipes.service.core.migrations.v093.utils.DocumentKeys.ALTERNATIVES;
+import static org.apache.streampipes.service.core.migrations.v093.utils.DocumentKeys.INTERNAL_NAME;
+
+public class JsonFormatMigrator implements FormatMigrator {
+
+  private String jsonFormatId;
+
+  public JsonFormatMigrator(String jsonFormatId,
+                            JsonObject existingFormat) {
+    this.jsonFormatId = jsonFormatId;
+  }
+
+  @Override
+  public void migrate(JsonObject newFormatProperties) {
+    newFormatProperties.get(MigrationHelpers.PROPERTIES).getAsJsonObject()
+        .get("staticProperties").getAsJsonArray()
+        .get(0).getAsJsonObject()
+        .get(MigrationHelpers.PROPERTIES).getAsJsonObject()
+        .get(ALTERNATIVES).getAsJsonArray()
+        .forEach(al -> {
+          if (al.getAsJsonObject().get(INTERNAL_NAME).getAsString().equals(jsonFormatId)) {
+            al.getAsJsonObject().add("selected", new JsonPrimitive(true));
+          }
+        });
+  }
+}

streampipes-service-core/src/main/java/org/apache/streampipes/service/core/migrations/v093/format/XmlFormatMigrator.java

@@ -0,0 +1,32 @@
+/*
+ * 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.streampipes.service.core.migrations.v093.format;
+
+import com.google.gson.JsonObject;
+
+public class XmlFormatMigrator implements FormatMigrator {
+  public XmlFormatMigrator(String xmlFormatId,
+                           JsonObject formatDescription) {
+  }
+
+  @Override
+  public void migrate(JsonObject newFormatProperties) {
+
+  }
+}

streampipes-service-core/src/main/java/org/apache/streampipes/service/core/migrations/v093/migrator/GenericAdapterMigrator.java

@@ -18,13 +18,119 @@
 
 package org.apache.streampipes.service.core.migrations.v093.migrator;
 
+import org.apache.streampipes.service.core.migrations.v093.format.CsvFormatMigrator;
+import org.apache.streampipes.service.core.migrations.v093.format.EmptyFormatMigrator;
+import org.apache.streampipes.service.core.migrations.v093.format.JsonFormatMigrator;
+import org.apache.streampipes.service.core.migrations.v093.format.XmlFormatMigrator;
+
+import com.google.gson.JsonArray;
 import com.google.gson.JsonObject;
 import org.lightcouch.CouchDbClient;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import static org.apache.streampipes.service.core.migrations.v093.migrator.MigrationHelpers.PROPERTIES;
+import static org.apache.streampipes.service.core.migrations.v093.utils.FormatIds.CSV;
+import static org.apache.streampipes.service.core.migrations.v093.utils.FormatIds.CSV_FORMAT_ID;
+import static org.apache.streampipes.service.core.migrations.v093.utils.FormatIds.GEOJSON_FORMAT_ID;
+import static org.apache.streampipes.service.core.migrations.v093.utils.FormatIds.IMAGE;
+import static org.apache.streampipes.service.core.migrations.v093.utils.FormatIds.IMAGE_FORMAT_ID;
+import static org.apache.streampipes.service.core.migrations.v093.utils.FormatIds.JSON;
+import static org.apache.streampipes.service.core.migrations.v093.utils.FormatIds.JSON_ARRAY_KEY_FORMAT_ID;
+import static org.apache.streampipes.service.core.migrations.v093.utils.FormatIds.JSON_ARRAY_NO_KEY_FORMAT_ID;
+import static org.apache.streampipes.service.core.migrations.v093.utils.FormatIds.JSON_OBJECT_FORMAT_ID;
+import static org.apache.streampipes.service.core.migrations.v093.utils.FormatIds.JSON_OBJECT_NEW_KEY;
+import static org.apache.streampipes.service.core.migrations.v093.utils.FormatIds.XML;
+import static org.apache.streampipes.service.core.migrations.v093.utils.FormatIds.XML_FORMAT_ID;
+import static org.apache.streampipes.service.core.migrations.v093.utils.GenericAdapterUtils.applyFormat;
+import static org.apache.streampipes.service.core.migrations.v093.utils.GenericAdapterUtils.getFormatTemplate;
 
 public class GenericAdapterMigrator implements AdapterMigrator {
 
+  private static final Logger LOG = LoggerFactory.getLogger(GenericAdapterMigrator.class);
+
+  private static final String PROTOCOL_DESC_KEY = "protocolDescription";
+  private static final String FORMAT_DESC_KEY = "formatDescription";
+  private static final String CONFIG_KEY = "config";
+
+  private final MigrationHelpers helpers;
+
+  public GenericAdapterMigrator(MigrationHelpers helpers) {
+    this.helpers = helpers;
+  }
+
   @Override
   public void migrate(CouchDbClient couchDbClient, JsonObject adapter) {
+    var adapterName = helpers.getAdapterName(adapter);
+    helpers.updateType(adapter);
+    helpers.updateFieldType(adapter);
+
+    JsonObject formatDescription = adapter.get(PROPERTIES).getAsJsonObject().get(FORMAT_DESC_KEY).getAsJsonObject();
+    JsonObject protocolDescription = adapter.get(PROPERTIES).getAsJsonObject().get(PROTOCOL_DESC_KEY).getAsJsonObject();
+
+    migrateProtocolDescription(adapter, protocolDescription);
+    migrateFormatDescription(adapter, formatDescription);
+
+    adapter.get(PROPERTIES).getAsJsonObject().remove(FORMAT_DESC_KEY);
+    adapter.get(PROPERTIES).getAsJsonObject().remove(PROTOCOL_DESC_KEY);
+
+    couchDbClient.update(adapter);
+
+    LOG.info("Successfully migrated adapter {}", adapterName);
+  }
+
+  private void migrateProtocolDescription(JsonObject adapter,
+                                          JsonObject protocolDescription) {
+    JsonArray config = adapter.get(PROPERTIES).getAsJsonObject().get(CONFIG_KEY).getAsJsonArray();
+    JsonArray protocolDescriptionConfig = protocolDescription.get(CONFIG_KEY).getAsJsonArray();
+    protocolDescriptionConfig.forEach(config::add);
+  }
+
+  private void migrateFormatDescription(JsonObject adapter,
+                                        JsonObject formatDescription) {
+    var adapterConfig = adapter
+        .get(PROPERTIES)
+        .getAsJsonObject()
+        .get(CONFIG_KEY)
+        .getAsJsonArray();
+
+    var formatTemplate = getFormatTemplate();
+
+    if (isFormat(formatDescription, JSON_OBJECT_FORMAT_ID)) {
+      var migrator = new JsonFormatMigrator(JSON_OBJECT_NEW_KEY, formatDescription);
+      applyFormat(JSON, formatTemplate, migrator);
+    } else if (isFormat(formatDescription, JSON_ARRAY_KEY_FORMAT_ID)) {
+      var migrator = new JsonFormatMigrator(JSON_ARRAY_KEY_FORMAT_ID, formatDescription);
+      applyFormat(JSON, formatTemplate, migrator);
+    } else if (isFormat(formatDescription, JSON_ARRAY_NO_KEY_FORMAT_ID)) {
+      var migrator = new JsonFormatMigrator(JSON_ARRAY_NO_KEY_FORMAT_ID, formatDescription);
+      applyFormat(JSON, formatTemplate, migrator);
+    } else if (isFormat(formatDescription, CSV_FORMAT_ID)) {
+      var migrator = new CsvFormatMigrator(CSV_FORMAT_ID, formatDescription);
+      applyFormat(CSV, formatTemplate, migrator);
+    } else if (isFormat(formatDescription, GEOJSON_FORMAT_ID)) {
+      var migrator = new JsonFormatMigrator(GEOJSON_FORMAT_ID, formatDescription);
+      applyFormat(JSON, formatTemplate, migrator);
+    } else if (isFormat(formatDescription, XML_FORMAT_ID)) {
+      var migrator = new XmlFormatMigrator(XML_FORMAT_ID, formatDescription);
+      applyFormat(XML, formatTemplate, migrator);
+    } else if (isFormat(formatDescription, IMAGE_FORMAT_ID)) {
+      applyFormat(IMAGE, formatTemplate, new EmptyFormatMigrator());
+    } else {
+      LOG.warn("Found unknown format {}", getAppId(formatDescription));
+    }
+
+    adapterConfig.add(formatTemplate);
+  }
+
+  private boolean isFormat(JsonObject formatDescription,
+                           String format) {
+    return getAppId(formatDescription).equals(format);
+  }
 
+  private String getAppId(JsonObject formatDescription) {
+    return formatDescription
+        .getAsJsonObject()
+        .get(MigrationHelpers.APP_ID).getAsString();
   }
 }

streampipes-service-core/src/main/java/org/apache/streampipes/service/core/migrations/v093/migrator/MigrationHelpers.java

@@ -18,12 +18,19 @@
 
 package org.apache.streampipes.service.core.migrations.v093.migrator;
 
+import org.apache.streampipes.service.core.migrations.v093.utils.AdapterMigrationUtils;
+
 import com.google.gson.JsonObject;
+import com.google.gson.JsonPrimitive;
 
 public class MigrationHelpers {
 
-  private static final String ID = "_id";
-  private static final String REV = "_rev";
+  public static final String ID = "_id";
+  public static final String REV = "_rev";
+
+  public static final String APP_ID = "appId";
+
+  public static final String PROPERTIES = "properties";
 
   public String getDocId(JsonObject adapter) {
     return adapter.get(ID).getAsString();
@@ -32,4 +39,16 @@ public String getDocId(JsonObject adapter) {
   public String getRev(JsonObject adapter) {
     return adapter.get(REV).getAsString();
   }
+
+  public void updateType(JsonObject adapter) {
+    adapter.add("type", new JsonPrimitive(AdapterMigrationUtils.NEW_MODEL));
+  }
+
+  public void updateFieldType(JsonObject adapter) {
+    adapter.add("field_type", new JsonPrimitive(AdapterMigrationUtils.NEW_MODEL));
+  }
+
+  public String getAdapterName(JsonObject adapter) {
+    return adapter.get("properties").getAsJsonObject().get("name").getAsString();
+  }
 }

streampipes-service-core/src/main/java/org/apache/streampipes/service/core/migrations/v093/migrator/SpecificAdapterMigrator.java

@@ -18,10 +18,7 @@
 
 package org.apache.streampipes.service.core.migrations.v093.migrator;
 
-import org.apache.streampipes.service.core.migrations.v093.utils.AdapterMigrationUtils;
-
 import com.google.gson.JsonObject;
-import com.google.gson.JsonPrimitive;
 import org.lightcouch.CouchDbClient;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -38,9 +35,9 @@ public SpecificAdapterMigrator(MigrationHelpers helpers) {
   @Override
   public void migrate(CouchDbClient couchDbClient,
                       JsonObject adapter) {
-    var adapterName = adapter.get("properties").getAsJsonObject().get("name").getAsString();
-    adapter.add("type", new JsonPrimitive(AdapterMigrationUtils.NEW_MODEL));
-    adapter.add("field_type", new JsonPrimitive(AdapterMigrationUtils.NEW_MODEL));
+    var adapterName = helpers.getAdapterName(adapter);
+    helpers.updateType(adapter);
+    helpers.updateFieldType(adapter);
 
     couchDbClient.update(adapter);