Fixes

sensorbucket · Jan 3, 2025 · 7cf0b09 · 7cf0b09
1 parent a643378
commit 7cf0b09
Show file tree

Hide file tree

Showing 3 changed files with 12 additions and 7 deletions.
diff --git a/services/core/main.go b/services/core/main.go
@@ -47,7 +47,7 @@ var (
 	AUTH_JWKS_URL                = env.Could("AUTH_JWKS_URL", "http://oathkeeper:4456/.well-known/jwks.json")
 	SYS_ARCHIVE_TIME             = env.Could("SYS_ARCHIVE_TIME", "30")
 	MEASUREMENT_BATCH_SIZE       = env.CouldInt("MEASUREMENT_BATCH_SIZE", 1024)
-	MEASUREMENT_COMMIT_INTERVAL  = env.CouldInt("MEASUREMENT_COMMIT_INTERVAL", 1)
+	MEASUREMENT_COMMIT_INTERVAL  = env.CouldInt("MEASUREMENT_COMMIT_INTERVAL", 1000)
 )
 
 func main() {
@@ -97,7 +97,7 @@ func Run(cleanup cleanupper.Cleanupper) error {
 	}
 	measurementstore := measurementsinfra.NewPSQL(db)
 	measurementservice := measurements.New(measurementstore, sysArchiveTime, keyClient)
-	cleanup.Add(measurementservice.StartMeasurementBatchStorer(MEASUREMENT_BATCH_SIZE, time.Duration(MEASUREMENT_COMMIT_INTERVAL)*time.Second))
+	cleanup.Add(measurementservice.StartMeasurementBatchStorer(MEASUREMENT_BATCH_SIZE, time.Duration(MEASUREMENT_COMMIT_INTERVAL)*time.Millisecond))
 
 	processingstore := processinginfra.NewPSQLStore(db)
 	processingPipelinePublisher := processinginfra.NewPipelineMessagePublisher(amqpConn, AMQP_XCHG_PIPELINE_MESSAGES)

diff --git a/services/core/measurements/application.go b/services/core/measurements/application.go
@@ -51,9 +51,10 @@ func (s *Service) StartMeasurementBatchStorer(batchSize int, interval time.Durat
 	t := time.NewTicker(interval)
 
 	commit := func() {
-		if len(measurements) > 0 {
-			log.Printf("Committing %d measurements\n", len(measurements))
+		if len(measurements) == 0 {
+			return
 		}
+		log.Printf("Committing %d measurements\n", len(measurements))
 		err := s.store.StoreMeasurements(measurements)
 		if err != nil {
 			log.Printf("Error storing measurements: %s\n", err.Error())
@@ -118,8 +119,7 @@ func (s *Service) ProcessPipelineMessage(pmsg pipeline.Message) error {
 		CreatedAt:                 time.Now(),
 	}
 
-	measurements := make([]Measurement, len(msg.Measurements))
-	for ix, m := range msg.Measurements {
+	for _, m := range msg.Measurements {
 
 		sensor, err := dev.GetSensorByExternalIDOrFallback(m.SensorExternalID)
 		if err != nil {
@@ -161,7 +161,7 @@ func (s *Service) ProcessPipelineMessage(pmsg pipeline.Message) error {
 			measurement.MeasurementAltitude = m.Altitude
 		}
 
-		measurements[ix] = measurement
+		s.measurementBatch <- measurement
 	}
 
 	return nil

diff --git a/services/core/measurements/infra/store_psql.go b/services/core/measurements/infra/store_psql.go
@@ -447,5 +447,10 @@ func (s *MeasurementStorePSQL) StoreMeasurements(measurements []measurements.Mea
 		)
 	}
 
+	_, err := q.RunWith(s.db).Exec()
+	if err != nil {
+		return fmt.Errorf("could not insert measurements: %w", err)
+	}
+
 	return nil
 }