Heartbeat activities

activities/geoparquet_to_geojson.go

@@ -10,6 +10,7 @@ import (
 	"github.com/apache/arrow/go/v14/parquet/file"
 	"github.com/onaio/akuko-temporal-go-tooling/internal/geojson"
 	"github.com/onaio/akuko-temporal-go-tooling/internal/geoparquet"
+	"go.temporal.io/sdk/activity"
 )
 
 // ReadFileBytes reads a file from a filepath and returns its contents as a byte slice
@@ -53,12 +54,18 @@ type ConvertGeoParquetToGeoJSONActivityReturnType struct {
 }
 
 func ConvertGeoParquetToGeoJSONActivity(ctx context.Context, params *ConvertGeoParquetToGeoJSONActivityParams) (*ConvertGeoParquetToGeoJSONActivityReturnType, error) {
+	message := "Reading geoparquet file bytes"
+	fmt.Println(message)
+	activity.RecordHeartbeat(ctx, message)
 	fileBytes, err := ReadFileBytes(params.GeoParquetFilePath)
 	if err != nil {
 		fmt.Println("Error:", err)
 		return nil, err
 	}
 
+	message = "Converting geoparquet to geojson"
+	fmt.Println(message)
+	activity.RecordHeartbeat(ctx, message)
 	// Convert from GeoParquet to GeoJSON
 	geoJSONBuffer := &bytes.Buffer{}
 	err = geojson.FromParquet(bytes.NewReader(fileBytes), geoJSONBuffer)
@@ -74,6 +81,9 @@ func ConvertGeoParquetToGeoJSONActivity(ctx context.Context, params *ConvertGeoP
 	}
 	defer reader.Close()
 
+	message = "Getting geoparquet metadata"
+	fmt.Println(message)
+	activity.RecordHeartbeat(ctx, message)
 	metadata, metadataErr := geoparquet.GetMetadata(reader.MetaData().GetKeyValueMetadata())
 	if metadataErr != nil {
 		fmt.Println("Error converting from Parquet to GeoJSON: %v", metadataErr)
@@ -82,6 +92,9 @@ func ConvertGeoParquetToGeoJSONActivity(ctx context.Context, params *ConvertGeoP
 
 	fmt.Println("MetaData: %s", &metadata)
 
+	message = "Writting geojson data to disk"
+	fmt.Println(message)
+	activity.RecordHeartbeat(ctx, message)
 	err = WriteFileBytes(params.GeoJSONFilePath, geoJSONBuffer.Bytes())
 	if err != nil {
 		fmt.Println("Error writing file to disk: %v", err)

activities/sanitize_geojson_feature_properties.go

@@ -7,6 +7,8 @@ import (
 	"io/ioutil"
 	"regexp"
 	"strings"
+
+	"go.temporal.io/sdk/activity"
 )
 
 // sanitizeStringToCubeSyntax sanitizes a string to conform to Cube syntax rules.
@@ -96,6 +98,8 @@ type SanitizeGeoJSONFeaturePropertiesActivityReturnType struct {
 }
 
 func SanitizeGeoJSONFeaturePropertiesActivity(ctx context.Context, params *SanitizeGeoJSONFeaturePropertiesActivityParams) (*SanitizeGeoJSONFeaturePropertiesActivityReturnType, error) {
+	message := "Sanitizing geojson file"
+	activity.RecordHeartbeat(ctx, message)
 	inputFile := params.GeoJSONFilePath
 	outputFile := params.GeoJSONFilePath
 
@@ -105,6 +109,8 @@ func SanitizeGeoJSONFeaturePropertiesActivity(ctx context.Context, params *Sanit
 		return nil, err
 	}
 
+	message = "GeoJSON file sanitized successfully."
+	activity.RecordHeartbeat(ctx, message)
 	fmt.Println("GeoJSON file sanitized successfully.")
 	var data = SanitizeGeoJSONFeaturePropertiesActivityReturnType{
 		FilePath: outputFile,