func (s *Service) Init(ctx context.Context) error {
	err := fileutil.EnsureDirExists(s.Config.Dir)
	if err != nil {
		return errors2.Wrapf(err, "it seems like %s dir doesn't exist, and it is not possible to create it", s.Config.Dir)
	}
	s.psr = newPersister(s.Config.Dir)
	ppipes, err := s.psr.loadPipes()
	if err != nil {
		s.logger.Error("could not read information about pipes. Corrupted? ", err)
		return err
	}
	for _, st := range ppipes {
		stm, err := newPPipe(s, st)
		if err != nil {
			s.logger.Error("Could not create pipe ", st, ", err=", err)
			return err
		}
		s.logger.Debug("creating instance for ", st)
		s.ppipes[st.Name] = stm
	}
	s.logger.Info(len(ppipes), " pipe(s) were created.")
	go s.notificatior()
	go s.pipesCleaner(time.Minute, 10*time.Minute)
	err = s.ensurePipesAtStart()
	if err != nil {
		s.Shutdown()
	}
	return err
}
Init provides an implementaion of linker.Initializer interface
package pipe
import (
	"context"
	"fmt"
	"github.com/jrivets/log4g"
	"github.com/logrange/logrange/pkg/cursor"
	"github.com/logrange/logrange/pkg/model/tag"
	journal2 "github.com/logrange/logrange/pkg/partition"
	context2 "github.com/logrange/range/pkg/context"
	"github.com/logrange/range/pkg/utils/errors"
	"github.com/logrange/range/pkg/utils/fileutil"
	errors2 "github.com/pkg/errors"
	"sort"
	"sync"
	"time"
)
type (
	PipesConfig struct {
		Dir string
		EnsureAtStart []Pipe
	}
	Pipe struct {
		Name string
		TagsCond string
		FltCond string
	}
	PipeDesc struct {
		Pipe
		DestTags tag.Set
	}
	Service struct {
		CurProvider cursor.Provider   `inject:""`
		Journals    *journal2.Service `inject:""`
		Config      *PipesConfig      `inject:""`
		logger    log4g.Logger
		lock      sync.Mutex
		closedCh  chan struct{}
		closedCtx context.Context
		psr       *persister
		ppipes map[string]*ppipe
		weCache map[string][]*ppipe
		wwg sync.WaitGroup
	}
)
func NewService() *Service {
	s := new(Service)
	s.logger = log4g.GetLogger("pipe.Service")
	s.ppipes = make(map[string]*ppipe)
	s.closedCh = make(chan struct{})
	s.closedCtx = context2.WrapChannel(s.closedCh)
	s.weCache = make(map[string][]*ppipe)
	return s
}