code refactoring

Author: rjtokenring

app/exporter/exporter.go

@@ -28,74 +28,87 @@ import (
 	"github.com/sirupsen/logrus"
 )
 
-func StartExporter(
-	ctx context.Context,
-	logger *logrus.Entry,
-	key, secret, orgid string,
-	tagsF *string,
-	resolution, timeWindowMinutes int,
-	destinationS3Bucket string,
-	aggregationStat string,
-	compress, enableAlignTimeWindow bool) error {
+type samplesExporter struct {
+	iotClient             *iot.Client
+	logger                *logrus.Entry
+	tagsF                 *string
+	compress              bool
+	enableAlignTimeWindow bool
+}
 
-	// Init client
+func New(key, secret, orgid string, tagsF *string, compress, enableAlignTimeWindow bool, logger *logrus.Entry) (*samplesExporter, error) {
 	iotcl, err := iot.NewClient(key, secret, orgid)
 	if err != nil {
-		return err
+		return nil, err
 	}
 
-	if tagsF != nil {
-		logger.Infoln("Filtering things linked to configured account using tags: ", *tagsF)
+	return &samplesExporter{
+		iotClient:             iotcl,
+		logger:                logger,
+		tagsF:                 tagsF,
+		compress:              compress,
+		enableAlignTimeWindow: enableAlignTimeWindow,
+	}, nil
+}
+
+func (s *samplesExporter) StartExporter(
+	ctx context.Context,
+	resolution, timeWindowMinutes int,
+	destinationS3Bucket string,
+	aggregationStat string) error {
+
+	if s.tagsF != nil {
+		s.logger.Infoln("Filtering things linked to configured account using tags: ", *s.tagsF)
 	} else {
-		logger.Infoln("Importing all things linked to configured account")
+		s.logger.Infoln("Importing all things linked to configured account")
 	}
 
-	things, err := iotcl.ThingList(ctx, nil, nil, true, utils.ParseTags(tagsF))
+	things, err := s.iotClient.ThingList(ctx, nil, nil, true, utils.ParseTags(s.tagsF))
 	if err != nil {
 		return err
 	}
 	thingsMap := make(map[string]iotclient.ArduinoThing, len(things))
 	for _, thing := range things {
-		logger.Infoln("  Thing: ", thing.Id, thing.Name)
+		s.logger.Infoln("  Thing: ", thing.Id, thing.Name)
 		thingsMap[thing.Id] = thing
 	}
 
 	// Extract data points from thing and push to S3
-	tsextractorClient := tsextractor.New(iotcl, logger)
+	tsextractorClient := tsextractor.New(s.iotClient, s.logger)
 
 	// Open s3 output writer
 	s3cl, err := s3.NewS3Client(destinationS3Bucket)
 	if err != nil {
 		return err
 	}
 
-	if writer, from, err := tsextractorClient.ExportTSToFile(ctx, timeWindowMinutes, thingsMap, resolution, aggregationStat, enableAlignTimeWindow); err != nil {
+	if writer, from, err := tsextractorClient.ExportTSToFile(ctx, timeWindowMinutes, thingsMap, resolution, aggregationStat, s.enableAlignTimeWindow); err != nil {
 		if writer != nil {
 			writer.Close()
 			defer writer.Delete()
 		}
-		logger.Error("Error aligning time series samples: ", err)
+		s.logger.Error("Error aligning time series samples: ", err)
 		return err
 	} else {
 		writer.Close()
 		defer writer.Delete()
 
 		fileToUpload := writer.GetFilePath()
 		destinationKeyFormat := "%s/%s.csv"
-		if compress {
-			logger.Infof("Compressing file: %s\n", fileToUpload)
+		if s.compress {
+			s.logger.Infof("Compressing file: %s\n", fileToUpload)
 			compressedFile, err := utils.GzipFileCompression(fileToUpload)
 			if err != nil {
 				return err
 			}
 			fileToUpload = compressedFile
-			logger.Infof("Generated compressed file: %s\n", fileToUpload)
+			s.logger.Infof("Generated compressed file: %s\n", fileToUpload)
 			destinationKeyFormat = "%s/%s.csv.gz"
 			defer func(f string) { os.Remove(f) }(fileToUpload)
 		}
 
 		destinationKey := fmt.Sprintf(destinationKeyFormat, from.Format("2006-01-02"), from.Format("2006-01-02-15-04"))
-		logger.Infof("Uploading file %s to bucket %s/%s\n", fileToUpload, s3cl.DestinationBucket(), destinationKey)
+		s.logger.Infof("Uploading file %s to bucket %s/%s\n", fileToUpload, s3cl.DestinationBucket(), destinationKey)
 		if err := s3cl.WriteFile(ctx, destinationKey, fileToUpload); err != nil {
 			return err
 		}

business/tsextractor/tsextractor.go

@@ -360,7 +360,7 @@ func (a *TsExtractor) populateStringTSDataIntoS3(
 			if !slices.Contains(populatedProperties, propertyID) {
 				populatedProperties = append(populatedProperties, propertyID)
 			}
-			samples = append(samples, composeRow(ts, thing.Id, thing.Name, propertyID, propertyName, propertyType, a.interfaceToString(value), ""))
+			samples = append(samples, composeRow(ts, thing.Id, thing.Name, propertyID, propertyName, propertyType, a.interfaceToString(value), "SAMPLED"))
 		}
 	}
 

lambda.go

@@ -188,7 +188,11 @@ func HandleRequest(ctx context.Context, event *AWSS3ImportTrigger) (*string, err
 	logger.Infoln("file compression enabled:", enabledCompression)
 	logger.Infoln("align time window:", enableAlignTimeWindow)
 
-	err = exporter.StartExporter(ctx, logger, *apikey, *apiSecret, organizationId, tags, *resolution, *extractionWindowMinutes, *destinationS3Bucket, *aggregationStat, enabledCompression, enableAlignTimeWindow)
+	tsExporter, err := exporter.New(*apikey, *apiSecret, organizationId, tags, enabledCompression, enableAlignTimeWindow, logger)
+	if err != nil {
+		return nil, err
+	}
+	err = tsExporter.StartExporter(ctx, *resolution, *extractionWindowMinutes, *destinationS3Bucket, *aggregationStat)
 	if err != nil {
 		message := "Error detected during data export"
 		return &message, err

resources/test/localexecution.go

@@ -89,10 +89,15 @@ func HandleRequest(ctx context.Context, dev bool) (*string, error) {
 		logger.Infoln("tags:", *tags)
 	}
 
-	err = exporter.StartExporter(ctx, logger, *apikey, *apiSecret, organizationId, tags, *resolution, TimeExtractionWindowMinutes, *destinationS3Bucket, "MAX", true, true)
+	tsExporter, err := exporter.New(*apikey, *apiSecret, organizationId, tags, true, true, logger)
 	if err != nil {
 		return nil, err
 	}
+	err = tsExporter.StartExporter(ctx, *resolution, TimeExtractionWindowMinutes, *destinationS3Bucket, "MAX")
+	if err != nil {
+		message := "Error detected during data export"
+		return &message, err
+	}
 
 	message := "Data exported successfully"
 	return &message, nil