Merge pull request #744 from fabriziosestito/remove_skip_old_event

Remove skipping for an older event

web/datapipeline/projector.go

@@ -53,15 +53,6 @@ func (p *projector) Project(dataCollectedEvent *DataCollectedEvent) error {
 		var subscription Subscription
 		tx.Clauses(clause.Locking{Strength: "UPDATE"}).Where(&Subscription{ProjectorID: p.ID, AgentID: dataCollectedEvent.AgentID}).First(&subscription)
 
-		if subscription.LastProjectedEventID >= dataCollectedEvent.ID {
-			log.Warnf("Projector: %s received an old event: %s %s %d. Skipping",
-				p.ID, dataCollectedEvent.DiscoveryType,
-				dataCollectedEvent.AgentID,
-				dataCollectedEvent.ID)
-
-			return nil
-		}
-
 		tx.Clauses(clause.OnConflict{
 			UpdateAll: true,
 		}).Create(&Subscription{

web/datapipeline/projector_test.go

@@ -114,29 +114,3 @@ func (suite *ProjectorTestSuite) TestProjector_Concurrency() {
 
 	suite.Equal(int64(2), subscription.LastProjectedEventID)
 }
-
-// TestProjector_SkipPastEvent tests that a projector does not project and update the subscription
-// if the event is older than the last projected one
-func (suite *ProjectorTestSuite) TestProjector_SkipPastEvent() {
-	projector := NewProjector("dummy_projector", suite.tx)
-	projected := false
-
-	projector.AddHandler("dummy_discovery_type", func(dataCollectedEvent *DataCollectedEvent, _ *gorm.DB) error {
-		projected = true
-		return nil
-	})
-
-	suite.tx.Create(&Subscription{
-		LastProjectedEventID: 123,
-		ProjectorID:          "dummy_projector",
-		AgentID:              "345",
-	})
-
-	projector.Project(&DataCollectedEvent{ID: 120, DiscoveryType: "dummy_discovery_type", AgentID: "345"})
-
-	var subscription Subscription
-	suite.tx.First(&subscription)
-
-	suite.False(projected)
-	suite.Equal(int64(123), subscription.LastProjectedEventID)
-}