implementation processModifySubscriptionEvent (#757)

* implementation processModifySubscriptionEvent

Signed-off-by: qGYdXbY2 <47661341+qGYdXbY2@users.noreply.github.com>

Author: qGYdXbY2

xyz-psql-connector/src/main/java/com/here/xyz/psql/DatabaseHandler.java

@@ -40,6 +40,7 @@
 import com.here.xyz.events.IterateHistoryEvent;
 import com.here.xyz.events.ModifyFeaturesEvent;
 import com.here.xyz.events.ModifySpaceEvent;
+import com.here.xyz.events.ModifySubscriptionEvent;
 import com.here.xyz.events.SearchForFeaturesEvent;
 import com.here.xyz.models.geojson.coordinates.BBox;
 import com.here.xyz.models.geojson.implementation.Feature;
@@ -450,6 +451,31 @@ else if(event.getOperation() == UPDATE)
         return new SuccessResponse().withStatus("OK");
     }
 
+
+    protected XyzResponse executeModifySubscription(ModifySubscriptionEvent event) throws SQLException {
+    
+        String space = event.getSpace(),
+               tableName  = config.readTableFromEvent(event),
+               schemaName = config.getDatabaseSettings().getSchema();
+     
+        boolean bLastSubscriptionToDelete = event.getHasNoActiveSubscriptions();
+
+        switch(event.getOperation())
+        { case CREATE : 
+          case UPDATE : return new FeatureCollection().withCount((long) executeUpdateWithRetry( SQLQueryBuilder.buildAddSubscriptionQuery(space, schemaName, tableName ) ));
+
+          case DELETE : 
+           if( !bLastSubscriptionToDelete )
+            return new FeatureCollection().withCount( 1l );
+           else  
+            return new FeatureCollection().withCount((long) executeUpdateWithRetry(SQLQueryBuilder.buildRemoveSubscriptionQuery(space, schemaName)));
+
+          default: break;
+        }
+     
+         return null;
+       }
+
     protected XyzResponse executeIterateHistory(IterateHistoryEvent event) throws SQLException {
         if(event.isCompact())
             return executeQueryWithRetry(SQLQueryBuilder.buildSquashHistoryQuery(event), this::compactHistoryResultSetHandler, true);

xyz-psql-connector/src/main/java/com/here/xyz/psql/PSQLXyzConnector.java

@@ -32,6 +32,7 @@
 import com.here.xyz.events.GetFeaturesByIdEvent;
 import com.here.xyz.events.GetFeaturesByTileEvent;
 import com.here.xyz.events.GetFeaturesByTileEvent.ResponseType;
+import com.here.xyz.events.ModifySubscriptionEvent.Operation;
 import com.here.xyz.events.GetHistoryStatisticsEvent;
 import com.here.xyz.events.GetStatisticsEvent;
 import com.here.xyz.events.GetStorageStatisticsEvent;
@@ -553,10 +554,28 @@ protected XyzResponse processModifySpaceEvent(ModifySpaceEvent event) throws Exc
 
   @Override
   protected XyzResponse processModifySubscriptionEvent(ModifySubscriptionEvent event) throws Exception {
-    // Needs further implementation
-    return new SuccessResponse();
+    try{
+      logger.info("{} Received ModifySpaceEvent", traceItem);
+
+      this.validateModifySubscriptionEvent(event);
+
+      return executeModifySubscription(event);
+    }catch (SQLException e){
+      return checkSQLException(e, config.readTableFromEvent(event));
+    }finally {
+      logger.info("{} Finished ModifySpaceEvent", traceItem);
+    }
   }
 
+  private void validateModifySubscriptionEvent(ModifySubscriptionEvent event) throws Exception {
+    
+   switch(event.getOperation())
+   { case CREATE : case UPDATE : case DELETE : break;
+     default:
+      throw new ErrorResponseException(streamId, XyzError.ILLEGAL_ARGUMENT, "Modify Subscription - Operation (" + event.getOperation() + ") not supported" );
+   }
+
+  }
 
   @Override
   protected XyzResponse processIterateHistoryEvent(IterateHistoryEvent event) {
@@ -642,7 +661,6 @@ private void validateModifySpaceEvent(ModifySpaceEvent event) throws Exception{
     }
   }
 
-
   private static final Pattern ERRVALUE_22P02 = Pattern.compile("invalid input syntax for type numeric:\\s+\"([^\"]*)\"\\s+Query:"),
                                ERRVALUE_22P05 = Pattern.compile("ERROR:\\s+(.*)\\s+Detail:\\s+(.*)\\s+Where:");
 

xyz-psql-connector/src/main/java/com/here/xyz/psql/SQLQueryBuilder.java

@@ -890,4 +890,22 @@ protected static String addHistoryTriggerSQL(final String schema, final String t
     public static String getForceMode(boolean isForce2D) {
       return isForce2D ? "ST_Force2D" : "ST_Force3D";
     }
+
+	public static SQLQuery buildAddSubscriptionQuery(String space, String schemaName, String tableName) {
+        String theSql = 
+          "insert into xyz_config.space_meta  ( id, schem, h_id, meta ) values(?,?,?,'{\"subscriptions\":true}' )" 
+         +" on conflict (id,schem) do "
+         +"  update set meta = xyz_config.space_meta.meta || excluded.meta ";
+     
+        return new SQLQuery(theSql, space,schemaName,tableName);
+	}
+
+	public static SQLQuery buildRemoveSubscriptionQuery(String space, String schemaName) {
+        String theSql = 
+          "update xyz_config.space_meta "
+         +" set meta = meta - 'subscriptions' "
+         +"where ( id, schem ) = ( ?, ? ) ";
+            
+        return new SQLQuery(theSql, space,schemaName );
+	}
 }