func (pp *ppipe) cleanPartitionsUnsafe() {
	if pp.deleted {
		return
	}
	removed := 0
	for src := range pp.partitions {
		_, err := pp.svc.Journals.TIndex.GetJournalTags(src, false)
		if err == errors2.NotFound {
			delete(pp.partitions, src)
			removed++
		}
	}
	if removed > 0 {
		pp.logger.Debug("cleanPartitions(): ", removed, " partitions were removed from the pipe settings ")
		pp.svc.psr.savePipeInfo(pp.cfg.Name, pp.partitions)
	}
}
cleanPartitionsUnsafe walks through partitions and removes all that doesn't exist anymore
package pipe
import (
	"context"
	"github.com/jrivets/log4g"
	"github.com/logrange/logrange/pkg/cursor"
	"github.com/logrange/logrange/pkg/lql"
	"github.com/logrange/logrange/pkg/model/tag"
	journal2 "github.com/logrange/logrange/pkg/partition"
	"github.com/logrange/range/pkg/records/journal"
	errors2 "github.com/logrange/range/pkg/utils/errors"
	"github.com/pkg/errors"
	"strconv"
	"strings"
	"sync"
)
type (
	ppipe struct {
		svc  *Service
		cfg  Pipe
		tags tag.Set
		srcF lql.TagsExpFunc
		fltF lql.WhereExpFunc
		lock    sync.Mutex
		logger  log4g.Logger
		clsCtx  context.Context
		cancelF context.CancelFunc
		deleted bool
		partitions map[string]*ppDesc
	}
	ppDesc struct {
		Tags string
		Pos journal.Pos
		LastKnwnPos journal.Pos
		wCharged bool
		curId uint64
	}
)
func newPPipe(svc *Service, p Pipe) (*ppipe, error) {
	srcF, err := lql.BuildTagsExpFunc(p.TagsCond)
	if err != nil {
		return nil, errors.Wrapf(err, "could not parse source condition %s", p.TagsCond)
	}
	fltF, err := lql.BuildWhereExpFunc(p.FltCond)
	if err != nil {
		return nil, errors.Wrapf(err, "could not parse filter condition %s", p.FltCond)
	}
	pp := new(ppipe)
	pp.svc = svc
	pp.srcF = srcF
	pp.fltF = fltF
	pp.cfg = p
	pp.clsCtx, pp.cancelF = context.WithCancel(svc.closedCtx)
	pp.tags = streamTags(p.Name)
	pp.partitions = make(map[string]*ppDesc)
	pp.logger = log4g.GetLogger("pipe.ppipe").WithId("{" + p.Name + "}").(log4g.Logger)
	svc.psr.loadPipeInfo(p.Name, pp.partitions)
	return pp, nil
}
func (pp *ppipe) onWriteEvent(we *journal2.WriteEvent) {
	pp.lock.Lock()
	pd, ok := pp.partitions[we.Src]
	if !ok {
		pp.logger.Debug("onWriteEvent(): first notification about a change ", we)
		pd = new(ppDesc)
		pd.Pos = we.StartPos
		pd.Tags = we.Tags.Line().String()
		pp.partitions[we.Src] = pd
	}
	pd.LastKnwnPos = we.EndPos
	pp.startWorker(we.Src, we.Tags.Line(), pd)
	pp.lock.Unlock()
}
func (pp *ppipe) startWorker(src string, srcTags tag.Line, pd *ppDesc) {
	pp.svc.wwg.Add(1)
	if pp.svc.closedCtx.Err() == nil && !pd.wCharged && pd.Pos.Less(pd.LastKnwnPos) {
		pd.wCharged = true
		w := newWorker(pp, src, srcTags, pp.tags.Line())
		go w.run(pp.clsCtx)
	} else {
		pp.svc.wwg.Done()
	}
}
func (pp *ppipe) getState(src string) (cursor.State, error) {
	pp.lock.Lock()
	sd, ok := pp.partitions[src]
	if !ok {
		pp.lock.Unlock()
		return cursor.State{}, errors2.NotFound
	}
	res := cursor.State{Id: sd.curId, Src: src, Pos: src + "=" + sd.Pos.String()}
	pp.lock.Unlock()
	return res, nil
}
func (pp *ppipe) cleanPartitions() {
	pp.lock.Lock()
	defer pp.lock.Unlock()
	pp.cleanPartitionsUnsafe()
}