changed to a batched platform message

Author: sposerina

src/server/pub_sub/salesforce_message_publisher.py

@@ -1,3 +1,4 @@
+import json
 import time
 import jwt
 import os
@@ -20,7 +21,8 @@
 TENANT_ID = os.getenv("TENANT_ID")
 PLATFORM_MESSAGE_AUTHOR = os.getenv("PLATFORM_MESSAGE_AUTHOR_RECORD_ID")
 
-UPDATE_TOPIC = "/event/Updated_Contacts_From_Pipeline__e"
+UPDATE_TOPIC = "/event/updated_contacts_batched__e"
+BATCH_SIZE = 200
 
 
 def send_pipeline_update_messages(contacts_list):
@@ -57,19 +59,35 @@ def send_pipeline_update_messages(contacts_list):
         schema_id = stub.GetTopic(pb2.TopicRequest(topic_name=UPDATE_TOPIC), metadata=auth_meta_data).schema_id
         schema = stub.GetSchema(pb2.SchemaRequest(schema_id=schema_id), metadata=auth_meta_data).schema_json
 
-        for contact_dict in contacts_list:
-            contact_dict['CreatedDate'] = int(datetime.now().timestamp())
-            contact_dict['CreatedById'] = PLATFORM_MESSAGE_AUTHOR
+        payloads = []
+        while len(contacts_list) > 0:
+            if len(contacts_list) > BATCH_SIZE:
+                current_batch = contacts_list[:BATCH_SIZE]
+                del contacts_list[:BATCH_SIZE]
+            else:
+                current_batch = contacts_list
+                contacts_list = []
 
+            root_object = {
+                "updatedContactsJson" : current_batch
+            }
+
+            message = {
+                "CreatedById": "0052g000003G926AAC",
+                "CreatedDate": int(datetime.now().timestamp()),
+                "updated_contacts_json__c": json.dumps(root_object)
+            }
             buf = io.BytesIO()
             encoder = avro.io.BinaryEncoder(buf)
             writer = avro.io.DatumWriter(avro.schema.parse(schema))
-            writer.write(contact_dict, encoder)
+            writer.write(message, encoder)
             payload = {
                 "schema_id": schema_id,
                 "payload": buf.getvalue()
             }
-            stub.Publish(pb2.PublishRequest(topic_name=UPDATE_TOPIC, events=[payload]), metadata=auth_meta_data)
-            logger.info('Pipeline update message sent')
+            payloads.append(payload)
+
+        stub.Publish(pb2.PublishRequest(topic_name=UPDATE_TOPIC, events=payloads), metadata=auth_meta_data)
+
+    logger.info("%s total pipeline update messages sent", len(payloads))
 
-    logger.info("%s total pipeline update messages sent", len(contacts_list))