func (this *SkyOutput) Run(r engine.OutputRunner, h engine.PluginHelper) error {
	var (
		ok      = true
		pack    *engine.PipelinePack
		inChan  = r.InChan()
		globals = engine.Globals()
		project = h.Project(this.project)
	)

LOOP:
	for ok {
		select {
		case pack, ok = <-inChan:
			if !ok {
				break LOOP
			}

			if globals.Debug {
				globals.Println(*pack)
			}

			this.feedSky(project, pack)
			pack.Recycle()
		}
	}

	return nil
}

func (this *EsOutput) feedEs(project *engine.ConfProject, pack *engine.PipelinePack) {
	if pack.EsType == "" || pack.EsIndex == "" {
		if project.ShowError {
			project.Printf("Empty ES meta: %s plugins:%v",
				*pack, pack.PluginNames())
		}

		this.counters.Inc("_error_", 1)

		return
	}

	this.counters.Inc(pack.EsIndex+":"+pack.EsType, 1)
	this.totalN += 1

	if this.dryRun {
		return
	}

	date := time.Unix(int64(pack.Message.Timestamp), 0)
	data, err := pack.Message.MarshalPayload()
	if err != nil {
		project.Println(err, *pack)
		return
	}
	id, _ := uuid.UUID()
	this.indexer.Index(pack.EsIndex, pack.EsType, id, "", &date, data) // ttl empty
}

func (this *DebugOutput) Run(r engine.OutputRunner, h engine.PluginHelper) error {
	var (
		globals = engine.Globals()
		pack    *engine.PipelinePack
		ok      = true
		inChan  = r.InChan()
	)

LOOP:
	for ok {
		select {
		case pack, ok = <-inChan:
			if !ok {
				break LOOP
			}

			if !this.blackhole {
				globals.Println(*pack)
			}

			pack.Recycle()
		}
	}

	return nil
}

func (this *EsFilter) Run(r engine.FilterRunner, h engine.PluginHelper) error {
	var (
		globals = engine.Globals()
		pack    *engine.PipelinePack
		ok      = true
		count   = 0
		inChan  = r.InChan()
	)

LOOP:
	for ok {
		select {
		case pack, ok = <-inChan:
			if !ok {
				break LOOP
			}

			if globals.Debug {
				globals.Println(*pack)
			}

			if this.handlePack(pack, h.Project(pack.Project)) {
				count += 1
				r.Inject(pack)
			} else {
				pack.Recycle()
			}
		}
	}

	globals.Printf("[%s]Total filtered: %d", r.Name(), count)

	return nil
}

func (this *CardinalityOutput) Run(r engine.OutputRunner, h engine.PluginHelper) error {
	var (
		pack   *engine.PipelinePack
		ok     = true
		inChan = r.InChan()
	)

	h.RegisterHttpApi("/card/{key}", func(w http.ResponseWriter,
		req *http.Request, params map[string]interface{}) (interface{}, error) {
		return this.handleHttpRequest(w, req, params)
	}).Methods("GET", "PUT")

LOOP:
	for ok {
		select {
		case pack, ok = <-inChan:
			if !ok {
				break LOOP
			}

			if pack.CardinalityKey != "" && pack.CardinalityData != nil {
				this.counters.Add(pack.CardinalityKey, pack.CardinalityData)
			}

			pack.Recycle()
		}
	}

	// before we quit, dump counters
	if this.checkpoint != "" {
		this.counters.Dump(this.checkpoint)
	}

	return nil
}

func (this *AlarmOutput) Run(r engine.OutputRunner, h engine.PluginHelper) error {
	var (
		pack       *engine.PipelinePack
		reloadChan = make(chan interface{})
		ok         = true
		inChan     = r.InChan()
	)

	for name, project := range this.projects {
		go this.runSendAlarmsWatchdog(h.Project(name), project)
	}

	// start all the workers
	goAhead := make(chan bool)
	for _, project := range this.projects {
		for _, worker := range project.workers {
			go worker.run(h, goAhead)
			<-goAhead // in case of race condition with worker.inject
		}
	}

	observer.Subscribe(engine.RELOAD, reloadChan)

LOOP:
	for ok {
		select {
		case <-reloadChan:
			// TODO

		case pack, ok = <-inChan:
			if !ok {
				break LOOP
			}

			this.handlePack(pack, h)
			pack.Recycle()
		}
	}

	close(this.stopChan)

	// all the workers cleanup
	for _, project := range this.projects {
		for _, worker := range project.workers {
			worker.cleanup()
		}
	}

	for _, project := range this.projects {
		close(project.emailChan)
	}

	return nil
}

func (this *EsBufferFilter) Run(r engine.FilterRunner, h engine.PluginHelper) error {
	var (
		pack    *engine.PipelinePack
		ok      = true
		globals = engine.Globals()
		inChan  = r.InChan()
	)

	for _, worker := range this.wokers {
		go worker.run(r, h)
	}

LOOP:
	for ok {
		select {
		case pack, ok = <-inChan:
			if !ok {
				break LOOP
			}

			if globals.Debug {
				globals.Println(*pack)
			}

			this.handlePack(pack)
			pack.Recycle()
		}
	}

	total := 0
	for _, worker := range this.wokers {
		total += worker.summary.N
		worker.flush(r, h)
	}

	// all workers will get notified and stop running
	close(this.stopChan)

	globals.Printf("[%s]Total filtered: %d", r.Name(), total)

	return nil
}

func (this *NetSenderOutput) Run(r engine.OutputRunner, h engine.PluginHelper) error {
	var (
		pack   *engine.PipelinePack
		ok     = true
		inChan = r.InChan()
	)

LOOP:
	for ok {
		select {
		case pack, ok = <-inChan:
			if !ok {
				break LOOP
			}

			this.target(pack).send(pack)
			pack.Recycle()
		}
	}

	return nil
}

func (this *CardinalityFilter) Run(r engine.FilterRunner,
	h engine.PluginHelper) error {
	var (
		pack   *engine.PipelinePack
		ok     = true
		inChan = r.InChan()
	)

LOOP:
	for ok {
		select {
		case pack, ok = <-inChan:
			if !ok {
				break LOOP
			}

			this.handlePack(r, h, pack)
			pack.Recycle()
		}
	}

	return nil
}

func (this *SelfSysInput) Run(r engine.InputRunner, h engine.PluginHelper) error {
	var (
		globals    = engine.Globals()
		stats      = newSysStat()
		inChan     = r.InChan()
		pack       *engine.PipelinePack
		jsonString string
		err        error
		stopped    = false
	)

	for !stopped {
		select {
		case <-this.stopChan:
			stopped = true

		case <-r.Ticker():
			// same effect as sleep
		}

		if stopped {
			break
		}

		stats.gatherStats()
		jsonString, err = stats.jsonString()
		if err != nil {
			globals.Println(err)
			continue
		}

		pack = <-inChan
		if err = pack.Message.FromLine(fmt.Sprintf("als,%d,%s",
			time.Now().Unix(), jsonString)); err != nil {
			globals.Printf("invalid sys stat: %s\n", jsonString)

			pack.Recycle()
			continue
		}

		pack.Project = "als"
		pack.Ident = this.ident
		pack.EsIndex = "fun_als"
		pack.EsType = "sys"
		r.Inject(pack)
	}

	return nil
}

func (this *ArchiveInput) doRunSingleLogfile(path string) {
	reader := als.NewAlsReader(path)
	if e := reader.Open(); e != nil {
		panic(e)
	}

	defer func() {
		reader.Close()
		this.workersWg.Done()
		atomic.AddInt32(&this.leftN, -1)

		<-this.workerNChan // release the lock
	}()

	var (
		line    []byte
		lineN   int
		inChan  = this.runner.InChan()
		err     error
		project = this.h.Project(this.project)
		pack    *engine.PipelinePack
		globals = engine.Globals()
	)

	for !this.stopping {
		line, err = reader.ReadLine()
		switch err {
		case nil:
			lineN += 1
			atomic.AddInt64(&this.lineN, 1)
			if globals.Verbose && lineN == 1 {
				project.Printf("[%s]started\n", path)
			}

			pack = <-inChan
			if err = pack.Message.FromLine(string(line)); err != nil {
				if project.ShowError && err != als.ErrEmptyLine {
					project.Printf("[%s]%v: %s", path, err, string(line))
				}

				pack.Recycle()
				continue
			}

			pack.Ident = this.ident
			pack.Project = this.project
			pack.Logfile.SetPath(path)
			if globals.Debug {
				globals.Println(*pack)
			}
			this.runner.Inject(pack)

		case io.EOF:
			if globals.Verbose {
				project.Printf("[%s]done, lines: %d\n", path, lineN)
			}

			this.chkpnt.Put(path)
			this.chkpnt.Dump()

			return

		default:
			// unknown error
			panic(err)
		}
	}

}

func (this *EsOutput) Run(r engine.OutputRunner, h engine.PluginHelper) error {
	var (
		pack         *engine.PipelinePack
		reloadChan   = make(chan interface{})
		ok           = true
		globals      = engine.Globals()
		inChan       = r.InChan()
		reportTicker = time.NewTicker(this.reportInterval)
	)

	this.indexer = core.NewBulkIndexer(this.bulkMaxConn)
	this.indexer.BulkMaxDocs = this.bulkMaxDocs
	this.indexer.BulkMaxBuffer = this.bulkMaxBuffer

	// start the bulk indexer
	this.indexer.Run(this.stopChan)

	defer reportTicker.Stop()

	observer.Subscribe(engine.RELOAD, reloadChan)

LOOP:
	for ok {
		select {
		case <-this.stopChan:
			ok = false

		case <-reportTicker.C:
			this.showPeriodicalStats()

		case <-reloadChan:
			// TODO

		case <-time.After(this.flushInterval):
			this.indexer.Flush()

		case pack, ok = <-inChan:
			if !ok {
				break LOOP
			}

			if globals.Debug {
				globals.Println(*pack)
			}

			this.feedEs(h.Project(pack.Project), pack)
			pack.Recycle()
		}
	}

	engine.Globals().Printf("[%s]Total output to ES: %d", r.Name(), this.totalN)

	// before shutdown, flush again
	if globals.Verbose {
		engine.Globals().Println("Waiting for ES flush...")
	}
	this.indexer.Flush()
	if globals.Verbose {
		engine.Globals().Println("ES flushed")
	}

	// let indexer stop
	this.stopChan <- true

	return nil
}

func (this *EsFilter) handlePack(pack *engine.PipelinePack, project *engine.ConfProject) bool {
	if pack.EsType == "" {
		pack.EsType = pack.Logfile.CamelCaseName()
	}
	if pack.EsIndex == "" {
		pack.EsIndex = indexName(project, this.indexPattern,
			time.Unix(int64(pack.Message.Timestamp), 0))
	}

	// each ES item has area and ts fields
	pack.Ident = this.ident
	pack.Message.SetField("_area", pack.Message.Area)
	pack.Message.SetField("_t", pack.Message.Timestamp)

	for _, conv := range this.converters {
		for _, key := range conv.keys {
			if conv.normalizers != nil {
				for _, norm := range conv.normalizers {
					val, err := pack.Message.FieldValue(key, als.KEY_TYPE_STRING)
					if err != nil {
						// no such field
						break
					}

					normed := normalizers[norm].ReplaceAll([]byte(val.(string)),
						[]byte("?"))
					val = string(normed)
					pack.Message.SetField(key+"_norm", val)
				}

				continue
			}

			switch conv.typ {
			case "money":
				amount, err := pack.Message.FieldValue(key, als.KEY_TYPE_MONEY)
				if err != nil {
					// has no such field
					continue
				}

				currency, err := pack.Message.FieldValue(conv.currency, als.KEY_TYPE_STRING)
				if err != nil {
					// has money field, but no currency field?
					return false
				}

				pack.Message.SetField("_usd",
					als.MoneyInUsdCents(currency.(string), amount.(int)))

			case "ip":
				ip, err := pack.Message.FieldValue(key, als.KEY_TYPE_IP)
				if err != nil {
					continue
				}

				pack.Message.SetField("_cntry", als.IpToCountry(ip.(string)))

			case "range":
				if len(conv.rang) < 2 {
					continue
				}

				val, err := pack.Message.FieldValue(key, als.KEY_TYPE_INT)
				if err != nil {
					continue
				}

				pack.Message.SetField(key+"_rg", als.GroupInt(val.(int), conv.rang))

			case "del":
				pack.Message.DelField(key)
			}
		}

	}

	return true
}