TMSS-421: processed review comments. adapted radbpglistener add_notifications...

TMSS-421: processed review comments. adapted radbpglistener add_notifications and handle the change in the payload (from a single id to a jsonblob)

SAS/ResourceAssignment/ResourceAssignmentDatabase/radb/sql/add_notifications.sql

@@ -12,12 +12,18 @@ BEGIN;
 SET LOCAL client_min_messages=warning;
 
 
+DROP TRIGGER IF EXISTS T_NOTIFY_task_INSERT ON resource_allocation.task CASCADE;
+DROP FUNCTION IF EXISTS resource_allocation.NOTIFY_task_INSERT();
+
+
 CREATE OR REPLACE FUNCTION resource_allocation.NOTIFY_task_INSERT()
 RETURNS TRIGGER AS $$
 DECLARE payload text;
 BEGIN
-SELECT CAST(NEW.id AS text) INTO payload;
+
+SELECT '{"id": ' || CAST(NEW.id AS text) || '}' INTO payload;
 PERFORM pg_notify(CAST('task_insert' AS text), payload);
+
 RETURN NEW;
 END;
 $$ LANGUAGE plpgsql;
@@ -29,12 +35,16 @@ FOR EACH ROW
 EXECUTE PROCEDURE resource_allocation.NOTIFY_task_INSERT();
 
 
+DROP TRIGGER IF EXISTS T_NOTIFY_task_UPDATE ON resource_allocation.task CASCADE;
+DROP FUNCTION IF EXISTS resource_allocation.NOTIFY_task_UPDATE();
+
+
 CREATE OR REPLACE FUNCTION resource_allocation.NOTIFY_task_UPDATE()
 RETURNS TRIGGER AS $$
 DECLARE payload text;
 BEGIN
 IF ROW(NEW.*) IS DISTINCT FROM ROW(OLD.*) THEN
-SELECT CAST(NEW.id AS text) INTO payload;
+SELECT '{"id": ' || CAST(NEW.id AS text) || '}' INTO payload;
 PERFORM pg_notify(CAST('task_update' AS text), payload);
 END IF;
 RETURN NEW;
@@ -48,12 +58,18 @@ FOR EACH ROW
 EXECUTE PROCEDURE resource_allocation.NOTIFY_task_UPDATE();
 
 
+DROP TRIGGER IF EXISTS T_NOTIFY_task_DELETE ON resource_allocation.task CASCADE;
+DROP FUNCTION IF EXISTS resource_allocation.NOTIFY_task_DELETE();
+
+
 CREATE OR REPLACE FUNCTION resource_allocation.NOTIFY_task_DELETE()
 RETURNS TRIGGER AS $$
 DECLARE payload text;
 BEGIN
-SELECT CAST(OLD.id AS text) INTO payload;
+
+SELECT '{"id": ' || CAST(OLD.id AS text) || '}' INTO payload;
 PERFORM pg_notify(CAST('task_delete' AS text), payload);
+
 RETURN OLD;
 END;
 $$ LANGUAGE plpgsql;
@@ -65,12 +81,18 @@ FOR EACH ROW
 EXECUTE PROCEDURE resource_allocation.NOTIFY_task_DELETE();
 
 
+DROP TRIGGER IF EXISTS T_NOTIFY_task_predecessor_INSERT_column_task_id ON resource_allocation.task_predecessor CASCADE;
+DROP FUNCTION IF EXISTS resource_allocation.NOTIFY_task_predecessor_INSERT_column_task_id();
+
+
 CREATE OR REPLACE FUNCTION resource_allocation.NOTIFY_task_predecessor_INSERT_column_task_id()
 RETURNS TRIGGER AS $$
 DECLARE payload text;
 BEGIN
-SELECT CAST(NEW.task_id AS text) INTO payload;
+
+SELECT '{"id": ' || CAST(NEW.id AS text) || '}' INTO payload;
 PERFORM pg_notify(CAST('task_predecessor_insert_column_task_id' AS text), payload);
+
 RETURN NEW;
 END;
 $$ LANGUAGE plpgsql;
@@ -82,12 +104,16 @@ FOR EACH ROW
 EXECUTE PROCEDURE resource_allocation.NOTIFY_task_predecessor_INSERT_column_task_id();
 
 
+DROP TRIGGER IF EXISTS T_NOTIFY_task_predecessor_UPDATE_column_task_id ON resource_allocation.task_predecessor CASCADE;
+DROP FUNCTION IF EXISTS resource_allocation.NOTIFY_task_predecessor_UPDATE_column_task_id();
+
+
 CREATE OR REPLACE FUNCTION resource_allocation.NOTIFY_task_predecessor_UPDATE_column_task_id()
 RETURNS TRIGGER AS $$
 DECLARE payload text;
 BEGIN
-IF ROW(NEW.*) IS DISTINCT FROM ROW(OLD.*) THEN
-SELECT CAST(NEW.task_id AS text) INTO payload;
+IF ROW(NEW.task_id) IS DISTINCT FROM ROW(OLD.task_id) THEN
+SELECT '{"id": ' || CAST(NEW.id AS text) || ', "task_id": "' || CAST(NEW.task_id AS text)  || '"}' INTO payload;
 PERFORM pg_notify(CAST('task_predecessor_update_column_task_id' AS text), payload);
 END IF;
 RETURN NEW;
@@ -101,12 +127,18 @@ FOR EACH ROW
 EXECUTE PROCEDURE resource_allocation.NOTIFY_task_predecessor_UPDATE_column_task_id();
 
 
+DROP TRIGGER IF EXISTS T_NOTIFY_task_predecessor_DELETE_column_task_id ON resource_allocation.task_predecessor CASCADE;
+DROP FUNCTION IF EXISTS resource_allocation.NOTIFY_task_predecessor_DELETE_column_task_id();
+
+
 CREATE OR REPLACE FUNCTION resource_allocation.NOTIFY_task_predecessor_DELETE_column_task_id()
 RETURNS TRIGGER AS $$
 DECLARE payload text;
 BEGIN
-SELECT CAST(OLD.task_id AS text) INTO payload;
+
+SELECT '{"id": ' || CAST(OLD.id AS text) || '}' INTO payload;
 PERFORM pg_notify(CAST('task_predecessor_delete_column_task_id' AS text), payload);
+
 RETURN OLD;
 END;
 $$ LANGUAGE plpgsql;
@@ -118,12 +150,18 @@ FOR EACH ROW
 EXECUTE PROCEDURE resource_allocation.NOTIFY_task_predecessor_DELETE_column_task_id();
 
 
+DROP TRIGGER IF EXISTS T_NOTIFY_task_predecessor_INSERT_column_predecessor_id ON resource_allocation.task_predecessor CASCADE;
+DROP FUNCTION IF EXISTS resource_allocation.NOTIFY_task_predecessor_INSERT_column_predecessor_id();
+
+
 CREATE OR REPLACE FUNCTION resource_allocation.NOTIFY_task_predecessor_INSERT_column_predecessor_id()
 RETURNS TRIGGER AS $$
 DECLARE payload text;
 BEGIN
-SELECT CAST(NEW.predecessor_id AS text) INTO payload;
+
+SELECT '{"id": ' || CAST(NEW.id AS text) || '}' INTO payload;
 PERFORM pg_notify(CAST('task_predecessor_insert_column_predecessor_id' AS text), payload);
+
 RETURN NEW;
 END;
 $$ LANGUAGE plpgsql;
@@ -135,12 +173,16 @@ FOR EACH ROW
 EXECUTE PROCEDURE resource_allocation.NOTIFY_task_predecessor_INSERT_column_predecessor_id();
 
 
+DROP TRIGGER IF EXISTS T_NOTIFY_task_predecessor_UPDATE_column_predecessor_id ON resource_allocation.task_predecessor CASCADE;
+DROP FUNCTION IF EXISTS resource_allocation.NOTIFY_task_predecessor_UPDATE_column_predecessor_id();
+
+
 CREATE OR REPLACE FUNCTION resource_allocation.NOTIFY_task_predecessor_UPDATE_column_predecessor_id()
 RETURNS TRIGGER AS $$
 DECLARE payload text;
 BEGIN
-IF ROW(NEW.*) IS DISTINCT FROM ROW(OLD.*) THEN
-SELECT CAST(NEW.predecessor_id AS text) INTO payload;
+IF ROW(NEW.predecessor_id) IS DISTINCT FROM ROW(OLD.predecessor_id) THEN
+SELECT '{"id": ' || CAST(NEW.id AS text) || ', "predecessor_id": "' || CAST(NEW.predecessor_id AS text)  || '"}' INTO payload;
 PERFORM pg_notify(CAST('task_predecessor_update_column_predecessor_id' AS text), payload);
 END IF;
 RETURN NEW;
@@ -154,12 +196,18 @@ FOR EACH ROW
 EXECUTE PROCEDURE resource_allocation.NOTIFY_task_predecessor_UPDATE_column_predecessor_id();
 
 
+DROP TRIGGER IF EXISTS T_NOTIFY_task_predecessor_DELETE_column_predecessor_id ON resource_allocation.task_predecessor CASCADE;
+DROP FUNCTION IF EXISTS resource_allocation.NOTIFY_task_predecessor_DELETE_column_predecessor_id();
+
+
 CREATE OR REPLACE FUNCTION resource_allocation.NOTIFY_task_predecessor_DELETE_column_predecessor_id()
 RETURNS TRIGGER AS $$
 DECLARE payload text;
 BEGIN
-SELECT CAST(OLD.predecessor_id AS text) INTO payload;
+
+SELECT '{"id": ' || CAST(OLD.id AS text) || '}' INTO payload;
 PERFORM pg_notify(CAST('task_predecessor_delete_column_predecessor_id' AS text), payload);
+
 RETURN OLD;
 END;
 $$ LANGUAGE plpgsql;
@@ -171,12 +219,16 @@ FOR EACH ROW
 EXECUTE PROCEDURE resource_allocation.NOTIFY_task_predecessor_DELETE_column_predecessor_id();
 
 
+DROP TRIGGER IF EXISTS T_NOTIFY_specification_UPDATE ON resource_allocation.specification CASCADE;
+DROP FUNCTION IF EXISTS resource_allocation.NOTIFY_specification_UPDATE();
+
+
 CREATE OR REPLACE FUNCTION resource_allocation.NOTIFY_specification_UPDATE()
 RETURNS TRIGGER AS $$
 DECLARE payload text;
 BEGIN
 IF ROW(NEW.*) IS DISTINCT FROM ROW(OLD.*) THEN
-SELECT CAST(NEW.id AS text) INTO payload;
+SELECT '{"id": ' || CAST(NEW.id AS text) || '}' INTO payload;
 PERFORM pg_notify(CAST('specification_update' AS text), payload);
 END IF;
 RETURN NEW;
@@ -190,12 +242,18 @@ FOR EACH ROW
 EXECUTE PROCEDURE resource_allocation.NOTIFY_specification_UPDATE();
 
 
+DROP TRIGGER IF EXISTS T_NOTIFY_resource_claim_INSERT ON resource_allocation.resource_claim CASCADE;
+DROP FUNCTION IF EXISTS resource_allocation.NOTIFY_resource_claim_INSERT();
+
+
 CREATE OR REPLACE FUNCTION resource_allocation.NOTIFY_resource_claim_INSERT()
 RETURNS TRIGGER AS $$
 DECLARE payload text;
 BEGIN
-SELECT CAST(NEW.id AS text) INTO payload;
+
+SELECT '{"id": ' || CAST(NEW.id AS text) || '}' INTO payload;
 PERFORM pg_notify(CAST('resource_claim_insert' AS text), payload);
+
 RETURN NEW;
 END;
 $$ LANGUAGE plpgsql;
@@ -207,12 +265,16 @@ FOR EACH ROW
 EXECUTE PROCEDURE resource_allocation.NOTIFY_resource_claim_INSERT();
 
 
+DROP TRIGGER IF EXISTS T_NOTIFY_resource_claim_UPDATE ON resource_allocation.resource_claim CASCADE;
+DROP FUNCTION IF EXISTS resource_allocation.NOTIFY_resource_claim_UPDATE();
+
+
 CREATE OR REPLACE FUNCTION resource_allocation.NOTIFY_resource_claim_UPDATE()
 RETURNS TRIGGER AS $$
 DECLARE payload text;
 BEGIN
 IF ROW(NEW.*) IS DISTINCT FROM ROW(OLD.*) THEN
-SELECT CAST(NEW.id AS text) INTO payload;
+SELECT '{"id": ' || CAST(NEW.id AS text) || '}' INTO payload;
 PERFORM pg_notify(CAST('resource_claim_update' AS text), payload);
 END IF;
 RETURN NEW;
@@ -226,12 +288,18 @@ FOR EACH ROW
 EXECUTE PROCEDURE resource_allocation.NOTIFY_resource_claim_UPDATE();
 
 
+DROP TRIGGER IF EXISTS T_NOTIFY_resource_claim_DELETE ON resource_allocation.resource_claim CASCADE;
+DROP FUNCTION IF EXISTS resource_allocation.NOTIFY_resource_claim_DELETE();
+
+
 CREATE OR REPLACE FUNCTION resource_allocation.NOTIFY_resource_claim_DELETE()
 RETURNS TRIGGER AS $$
 DECLARE payload text;
 BEGIN
-SELECT CAST(OLD.id AS text) INTO payload;
+
+SELECT '{"id": ' || CAST(OLD.id AS text) || '}' INTO payload;
 PERFORM pg_notify(CAST('resource_claim_delete' AS text), payload);
+
 RETURN OLD;
 END;
 $$ LANGUAGE plpgsql;
@@ -243,12 +311,16 @@ FOR EACH ROW
 EXECUTE PROCEDURE resource_allocation.NOTIFY_resource_claim_DELETE();
 
 
+DROP TRIGGER IF EXISTS T_NOTIFY_resource_availability_UPDATE_column_resource_id ON resource_monitoring.resource_availability CASCADE;
+DROP FUNCTION IF EXISTS resource_monitoring.NOTIFY_resource_availability_UPDATE_column_resource_id();
+
+
 CREATE OR REPLACE FUNCTION resource_monitoring.NOTIFY_resource_availability_UPDATE_column_resource_id()
 RETURNS TRIGGER AS $$
 DECLARE payload text;
 BEGIN
-IF ROW(NEW.*) IS DISTINCT FROM ROW(OLD.*) THEN
-SELECT CAST(NEW.resource_id AS text) INTO payload;
+IF ROW(NEW.resource_id) IS DISTINCT FROM ROW(OLD.resource_id) THEN
+SELECT '{"id": ' || CAST(NEW.id AS text) || ', "resource_id": "' || CAST(NEW.resource_id AS text)  || '"}' INTO payload;
 PERFORM pg_notify(CAST('resource_availability_update_column_resource_id' AS text), payload);
 END IF;
 RETURN NEW;
@@ -262,12 +334,16 @@ FOR EACH ROW
 EXECUTE PROCEDURE resource_monitoring.NOTIFY_resource_availability_UPDATE_column_resource_id();
 
 
+DROP TRIGGER IF EXISTS T_NOTIFY_resource_capacity_UPDATE_column_resource_id ON resource_monitoring.resource_capacity CASCADE;
+DROP FUNCTION IF EXISTS resource_monitoring.NOTIFY_resource_capacity_UPDATE_column_resource_id();
+
+
 CREATE OR REPLACE FUNCTION resource_monitoring.NOTIFY_resource_capacity_UPDATE_column_resource_id()
 RETURNS TRIGGER AS $$
 DECLARE payload text;
 BEGIN
-IF ROW(NEW.*) IS DISTINCT FROM ROW(OLD.*) THEN
-SELECT CAST(NEW.resource_id AS text) INTO payload;
+IF ROW(NEW.resource_id) IS DISTINCT FROM ROW(OLD.resource_id) THEN
+SELECT '{"id": ' || CAST(NEW.id AS text) || ', "resource_id": "' || CAST(NEW.resource_id AS text)  || '"}' INTO payload;
 PERFORM pg_notify(CAST('resource_capacity_update_column_resource_id' AS text), payload);
 END IF;
 RETURN NEW;

SAS/ResourceAssignment/ResourceAssignmentDatabase/radbpglistener.py

@@ -74,7 +74,7 @@ class RADBPGListener(PostgresListener):
 
     def onTaskUpdated(self, payload = None):
         # Send notification for the given updated task
-        task_id = payload        
+        task_id = json.loads(payload)['id']
         task = self.radb.getTask(task_id)
         self._sendNotification('TaskUpdated', task)
 
@@ -86,39 +86,39 @@ class RADBPGListener(PostgresListener):
             self._sendNotification('TaskUpdated', suc_sched_task)
 
     def onTaskInserted(self, payload = None):
-        self._sendNotification('TaskInserted', self.radb.getTask(payload))
+        self._sendNotification('TaskInserted', self.radb.getTask(json.loads(payload)['id']))
 
     def onTaskDeleted(self, payload = None):
-        self._sendNotification('TaskDeleted', payload)
+        self._sendNotification('TaskDeleted', json.loads(payload)['id'])
 
-    def onTaskPredecessorChanged(self, task_id):
-        logger.info('onTaskPredecessorChanged(task_id=%s)', task_id)
-        self._sendNotification('TaskUpdated', self.radb.getTask(task_id))
+    def onTaskPredecessorChanged(self, payload):
+        logger.info('onTaskPredecessorChanged(task_id=%s)', json.loads(payload)['task_id'])
+        self._sendNotification('TaskUpdated', self.radb.getTask(json.loads(payload)['task_id']))
 
-    def onTaskSuccessorChanged(self, task_id):
-        logger.info('onTaskSuccessorChanged(task_id=%s)', task_id)
-        self._sendNotification('TaskUpdated', self.radb.getTask(task_id))
+    def onTaskSuccessorChanged(self, payload):
+        logger.info('onTaskSuccessorChanged(task_id=%s)', json.loads(payload)['task_id'])
+        self._sendNotification('TaskUpdated', self.radb.getTask(json.loads(payload)['task_id']))
 
     def onSpecificationUpdated(self, payload = None):
         # when the specification starttime and endtime are updated, then that effects the task as well
-        self._sendNotification('TaskUpdated', self.radb.getTask(specification_id=payload))
+        self._sendNotification('TaskUpdated', self.radb.getTask(specification_id=json.loads(payload)['id']))
 
     def onResourceClaimUpdated(self, payload = None):
-        self._sendNotification('ResourceClaimUpdated', self.radb.getResourceClaim(payload))
+        self._sendNotification('ResourceClaimUpdated', self.radb.getResourceClaim(json.loads(payload)['id']))
 
     def onResourceClaimInserted(self, payload = None):
-        self._sendNotification('ResourceClaimInserted', self.radb.getResourceClaim(payload))
+        self._sendNotification('ResourceClaimInserted', self.radb.getResourceClaim(json.loads(payload)['id']))
 
     def onResourceClaimDeleted(self, payload = None):
-        self._sendNotification('ResourceClaimDeleted', payload)
+        self._sendNotification('ResourceClaimDeleted', json.loads(payload)['id'])
 
     def onResourceAvailabilityUpdated(self, payload = None):
-        r = self.radb.getResources(resource_ids=[payload], include_availability=True)[0]
+        r = self.radb.getResources(resource_ids=[json.loads(payload)['id']], include_availability=True)[0]
         r = {k:r[k] for k in ['id', 'active']}
         self._sendNotification('ResourceAvailabilityUpdated', r)
 
     def onResourceCapacityUpdated(self, payload = None):
-        r = self.radb.getResources(resource_ids=[payload], include_availability=True)[0]
+        r = self.radb.getResources(resource_ids=[json.loads(payload)['id']], include_availability=True)[0]
         r = {k:r[k] for k in ['id', 'total_capacity', 'available_capacity', 'used_capacity']}
         self._sendNotification('ResourceCapacityUpdated', r)