diff --git a/stage.go b/stage.go index 385453b..f166e32 100644 --- a/stage.go +++ b/stage.go @@ -86,13 +86,15 @@ func (s *stage) chain(p Plugin) { wg.Add(1) go func(f *file) { defer wg.Done() - f.rewind() - if err := proc.Process(s, f); err != nil { + keep, err := proc.Process(s, f) + if err != nil { s.gs.fault(s, "Processing", f, err) + } else if keep { + dispatch(f) + } else { + atomic.AddInt64(&s.gs.active, -1) } - - dispatch(f) }(f) } } diff --git a/types.go b/types.go index d0f6010..7fa045e 100644 --- a/types.go +++ b/types.go @@ -96,5 +96,5 @@ type Finalizer interface { } type Processor interface { - Process(ctx Context, f File) error + Process(ctx Context, f File) (bool, error) }