addPartial appends the event that matched the expression at the sequence index. If the event arrived out of order, then the isOOO parameter is equal to false.
(seqID int, e *event.Event, isOOO bool)
| 313 | // sequence index. If the event arrived out of order, then the isOOO |
| 314 | // parameter is equal to false. |
| 315 | func (s *sequenceState) addPartial(seqID int, e *event.Event, isOOO bool) { |
| 316 | s.mu.Lock() |
| 317 | defer s.mu.Unlock() |
| 318 | if len(s.partials[seqID]) > maxOutstandingPartials { |
| 319 | partialBreaches.Add(s.name, 1) |
| 320 | if !s.isPartialsBreached.Load() { |
| 321 | log.Warnf("max partials encountered in sequence %s slot [%d]. "+ |
| 322 | "Dropping incoming partial: %s", s.name, seqID, e) |
| 323 | } |
| 324 | s.isPartialsBreached.Store(true) |
| 325 | return |
| 326 | } |
| 327 | key := e.PartialKey() |
| 328 | if key != 0 { |
| 329 | for _, p := range s.partials[seqID] { |
| 330 | if key == p.PartialKey() { |
| 331 | log.Debugf("event %s for tuple %d already in sequence state", e, key) |
| 332 | return |
| 333 | } |
| 334 | } |
| 335 | } |
| 336 | if isOOO { |
| 337 | e.AddMeta(event.RuleSequenceOOOKey, true) |
| 338 | } |
| 339 | log.Debugf("adding partial to sequence [%s] slot [%d] for expression %q, ooo: %t: %s", s.name, seqID, s.expr(seqID), isOOO, e) |
| 340 | partialsPerSequence.Add(s.name, 1) |
| 341 | s.partials[seqID] = append(s.partials[seqID], e) |
| 342 | sort.Slice(s.partials[seqID], func(n, m int) bool { return s.partials[seqID][n].Timestamp.Before(s.partials[seqID][m].Timestamp) }) |
| 343 | } |
| 344 | |
| 345 | // gc prunes the sequence partial if it remained |
| 346 | // more time than specified by max span or if max |