From 88e999a0bbb0f34ed456ba565f67dce45590a9f5 Mon Sep 17 00:00:00 2001 From: Paul Buetow Date: Fri, 10 Dec 2021 09:53:48 +0000 Subject: Refactor --- internal/mapr/server/aggregate.go | 50 ++++++++++++++++++++++++--------------- 1 file changed, 31 insertions(+), 19 deletions(-) diff --git a/internal/mapr/server/aggregate.go b/internal/mapr/server/aggregate.go index 9e3f68e..ed32f8f 100644 --- a/internal/mapr/server/aggregate.go +++ b/internal/mapr/server/aggregate.go @@ -165,30 +165,14 @@ func (a *Aggregate) fieldsFromLines(ctx context.Context) <-chan map[string]strin line, ok, noMoreChannels := a.nextLine() if !ok { if noMoreChannels { - break + return } time.Sleep(time.Millisecond * 100) continue } - maprLine := strings.TrimSpace(line.Content.String()) - line.Recycle() // after this, don't use line object anymore!!! - fields, err := a.parser.MakeFields(maprLine) - - if err != nil { - // Should fields be ignored anyway? - if err != logformat.ErrIgnoreFields { - dlog.Server.Error(fields, err) - } - continue - } - if !a.query.WhereClause(fields) { - continue - } - - select { - case fieldsCh <- fields: - case <-ctx.Done(): + if err := a.fieldFromLine(ctx, line, fieldsCh); err != nil { + dlog.Server.Error(err) } } }() @@ -196,6 +180,34 @@ func (a *Aggregate) fieldsFromLines(ctx context.Context) <-chan map[string]strin return fieldsCh } +func (a *Aggregate) fieldFromLine(ctx context.Context, line *line.Line, + fieldsCh chan<- map[string]string) error { + + maprLine := strings.TrimSpace(line.Content.String()) + + // after recycling it, don't use line object anymore!!! + line.Recycle() + fields, err := a.parser.MakeFields(maprLine) + + if err != nil { + // Should fields be ignored anyway? + if err != logformat.ErrIgnoreFields { + return err + } + return nil + } + if !a.query.WhereClause(fields) { + return nil + } + + select { + case fieldsCh <- fields: + case <-ctx.Done(): + } + + return nil +} + func (a *Aggregate) setAdditionalFields(ctx context.Context, fieldsCh <-chan map[string]string) <-chan map[string]string { -- cgit v1.2.3