chore: remove unnessessary heartbeating readable code

supaglue-labs · Dec 12, 2023 · c8b0607 · c8b0607
1 parent 7dd1471
commit c8b0607
Show file tree

Hide file tree

Showing 2 changed files with 6 additions and 52 deletions.
diff --git a/packages/sync-workflows/activities/sync_entity_records.ts b/packages/sync-workflows/activities/sync_entity_records.ts
@@ -71,7 +71,7 @@ export function createSyncEntityRecords(
           return await writer.writeEntityRecords(
             connection,
             entity.name,
-            toHeartbeatingReadable(toMappedPropertiesReadable(stream, fieldMappingConfig)),
+            toMappedPropertiesReadable(stream, fieldMappingConfig),
             heartbeat,
             /* diffAndDeleteRecords */ shouldDeleteRecords(!updatedAfterMs, connection.providerName)
           );
@@ -81,7 +81,7 @@ export function createSyncEntityRecords(
           return await writer.writeEntityRecords(
             connection,
             entity.name,
-            toHeartbeatingReadable(toMappedPropertiesReadable(stream, fieldMappingConfig)),
+            toMappedPropertiesReadable(stream, fieldMappingConfig),
             heartbeat,
             /* diffAndDeleteRecords */ shouldDeleteRecords(!updatedAfterMs, connection.providerName)
           );
@@ -116,29 +116,6 @@ export function createSyncEntityRecords(
   };
 }
 
-function toHeartbeatingReadable(readable: Readable): Readable {
-  // TODO: While this ensures rescheduling of this activity if the process dies,
-  // it does not ensure that we stop the stream processing.
-  // We need to include a timeout here to clean up the pipeline when we
-  // exceed the heartbeat timeout.
-  return pipeline(
-    readable,
-    new Transform({
-      objectMode: true,
-      transform: (chunk, encoding, callback) => {
-        Context.current().heartbeat();
-        try {
-          callback(null, chunk);
-        } catch (e: any) {
-          return callback(e);
-        }
-      },
-    }),
-    // eslint-disable-next-line @typescript-eslint/no-empty-function
-    () => {}
-  );
-}
-
 function toMappedPropertiesReadable(readable: Readable, fieldMappingConfig: FieldMappingConfig): Readable {
   return pipeline(
     readable,

diff --git a/packages/sync-workflows/activities/sync_object_records.ts b/packages/sync-workflows/activities/sync_object_records.ts
@@ -105,7 +105,7 @@ export function createSyncObjectRecords(
                 return await writer.writeCommonObjectRecords(
                   connection,
                   object as CRMCommonObjectType,
-                  toHeartbeatingReadable(readable),
+                  readable,
                   heartbeat,
                   /* diffAndDeleteRecords */ shouldDeleteRecords(!updatedAfterMs, connection.providerName)
                 );
@@ -120,7 +120,7 @@ export function createSyncObjectRecords(
                 return await writer.writeCommonObjectRecords(
                   connection,
                   object as EngagementCommonObjectType,
-                  toHeartbeatingReadable(readable),
+                  readable,
                   heartbeat,
                   /* diffAndDeleteRecords */ shouldDeleteRecords(!updatedAfterMs, connection.providerName)
                 );
@@ -150,7 +150,7 @@ export function createSyncObjectRecords(
             return await writer.writeObjectRecords(
               connection,
               object,
-              toHeartbeatingReadable(toMappedPropertiesReadable(stream, fieldMappingConfig)),
+              toMappedPropertiesReadable(stream, fieldMappingConfig),
               heartbeat,
               /* diffAndDeleteRecords */ shouldDeleteRecords(!updatedAfterMs, connection.providerName),
               'standard'
@@ -174,7 +174,7 @@ export function createSyncObjectRecords(
             return await writer.writeObjectRecords(
               connection,
               object,
-              toHeartbeatingReadable(toMappedPropertiesReadable(stream, fieldMappingConfig)),
+              toMappedPropertiesReadable(stream, fieldMappingConfig),
               heartbeat,
               /* diffAndDeleteRecords */ shouldDeleteRecords(!updatedAfterMs, connection.providerName),
               'custom'
@@ -215,29 +215,6 @@ export function createSyncObjectRecords(
   };
 }
 
-function toHeartbeatingReadable(readable: Readable): Readable {
-  // TODO: While this ensures rescheduling of this activity if the process dies,
-  // it does not ensure that we stop the stream processing.
-  // We need to include a timeout here to clean up the pipeline when we
-  // exceed the heartbeat timeout.
-  return pipeline(
-    readable,
-    new Transform({
-      objectMode: true,
-      transform: (chunk, encoding, callback) => {
-        Context.current().heartbeat();
-        try {
-          callback(null, chunk);
-        } catch (e: any) {
-          return callback(e);
-        }
-      },
-    }),
-    // eslint-disable-next-line @typescript-eslint/no-empty-function
-    () => {}
-  );
-}
-
 function toMappedPropertiesReadable(readable: Readable, fieldMappingConfig: FieldMappingConfig): Readable {
   return pipeline(
     readable,