func (f *Filterer) HandleFrame(p FrameHandlerPayload) {
	f.cl.Do(func() {
		f.statIncomingRate.Add(1)
		fm := f.p.get()
		if ret := avutil.AvFrameRef(fm, p.Frame); ret < 0 {
			emitAvError(f, f.eh, ret, "avutil.AvFrameRef failed")
			return
		}
		f.c.Add(func() {
			f.cl.Do(func() {
				defer f.HandlePause()
				defer f.p.put(fm)
				f.statProcessedRate.Add(1)
				bufferSrcCtxs, ok := f.bufferSrcCtxs[p.Node]
				if !ok {
					return
				}
				for _, bufferSrcCtx := range bufferSrcCtxs {
					if ret := f.g.AvBuffersrcAddFrameFlags(bufferSrcCtx, fm, avfilter.AV_BUFFERSRC_FLAG_KEEP_REF); ret < 0 {
						emitAvError(f, f.eh, ret, "f.g.AvBuffersrcAddFrameFlags failed")
						return
					}
				}
				for {
					if stop := f.pullFilteredFrame(p.Descriptor); stop {
						return
					}
				}
			})
		})
	})
}
HandleFrame implements the FrameHandler interface
package astilibav
import (
	"context"
	"errors"
	"fmt"
	"strconv"
	"strings"
	"sync/atomic"
	"time"
	"github.com/asticode/go-astiencoder"
	"github.com/asticode/go-astikit"
	"github.com/asticode/goav/avcodec"
	"github.com/asticode/goav/avfilter"
	"github.com/asticode/goav/avutil"
)
var countFilterer uint64
type Filterer struct {
	*astiencoder.BaseNode
	bufferSinkCtx     *avfilter.Context
	bufferSrcCtxs     map[astiencoder.Node][]*avfilter.Context
	c                 *astikit.Chan
	cl                *astikit.Closer
	d                 *frameDispatcher
	eh                *astiencoder.EventHandler
	emulatePeriod     time.Duration
	g                 *avfilter.Graph
	outputCtx         Context
	p                 *framePool
	restamper         FrameRestamper
	statIncomingRate  *astikit.CounterRateStat
	statProcessedRate *astikit.CounterRateStat
}
type FiltererOptions struct {
	Content     string
	EmulateRate avutil.Rational
	Inputs      map[string]astiencoder.Node
	Node        astiencoder.NodeOptions
	OutputCtx   Context
	Restamper   FrameRestamper
}
func NewFilterer(o FiltererOptions, eh *astiencoder.EventHandler, c *astikit.Closer, s *astiencoder.Stater) (f *Filterer, err error) {
	count := atomic.AddUint64(&countFilterer, uint64(1))
	o.Node.Metadata = o.Node.Metadata.Extend(fmt.Sprintf("filterer_%d", count), fmt.Sprintf("Filterer #%d", count), "Filters", "filterer")
	f = &Filterer{
		bufferSrcCtxs:     make(map[astiencoder.Node][]*avfilter.Context),
		c:                 astikit.NewChan(astikit.ChanOptions{ProcessAll: true}),
		cl:                c.NewChild(),
		eh:                eh,
		outputCtx:         o.OutputCtx,
		restamper:         o.Restamper,
		statIncomingRate:  astikit.NewCounterRateStat(),
		statProcessedRate: astikit.NewCounterRateStat(),
	}
	f.BaseNode = astiencoder.NewBaseNode(o.Node, eh, s, f, astiencoder.EventTypeToNodeEventName)
	f.p = newFramePool(f.cl)
	f.d = newFrameDispatcher(f, eh, f.p)
	f.addStats()
	if len(o.Inputs) == 0 {
		if o.EmulateRate.Num() <= 0 || o.EmulateRate.Den() <= 0 {
			err = errors.New("astilibav: no inputs but no emulate rate either")
			return
		}
		f.emulatePeriod = time.Duration(o.EmulateRate.Den() * 1e9 / o.EmulateRate.Num())
	}
	f.g = avfilter.AvfilterGraphAlloc()
	f.cl.Add(func() error {
		f.g.AvfilterGraphFree()
		return nil
	})
	var bufferFunc func() *avfilter.Filter
	var bufferSink *avfilter.Filter
	switch o.OutputCtx.CodecType {
	case avcodec.AVMEDIA_TYPE_AUDIO:
		bufferFunc = func() *avfilter.Filter { return avfilter.AvfilterGetByName("abuffer") }
		bufferSink = avfilter.AvfilterGetByName("abuffersink")
	case avcodec.AVMEDIA_TYPE_VIDEO:
		bufferFunc = func() *avfilter.Filter { return avfilter.AvfilterGetByName("buffer") }
		bufferSink = avfilter.AvfilterGetByName("buffersink")
	default:
		err = fmt.Errorf("astilibav: codec type %v is not handled by filterer", o.OutputCtx.CodecType)
		return
	}
	var bufferSinkCtx *avfilter.Context
	if ret := avfilter.AvfilterGraphCreateFilter(&bufferSinkCtx, bufferSink, "out", "", nil, f.g); ret < 0 {
		err = fmt.Errorf("astilibav: avfilter.AvfilterGraphCreateFilter on empty args failed: %w", NewAvError(ret))
		return
	}
	f.bufferSinkCtx = bufferSinkCtx
	inputs := avfilter.AvfilterInoutAlloc()
	inputs.SetName("out")
	inputs.SetFilterCtx(f.bufferSinkCtx)
	inputs.SetPadIdx(0)
	inputs.SetNext(nil)
	var previousOutput *avfilter.Input
	for n, i := range o.Inputs {
		v, ok := i.(OutputContexter)
		if !ok {
			err = fmt.Errorf("astilibav: input %s is not an OutputContexter", n)
			return
		}
		ctx := v.OutputCtx()
		bufferSrc := bufferFunc()
		var args []string
		switch ctx.CodecType {
		case avcodec.AVMEDIA_TYPE_AUDIO:
			args = []string{
				"channel_layout=" + avutil.AvGetChannelLayoutString(ctx.Channels, ctx.ChannelLayout),
				"sample_fmt=" + avutil.AvGetSampleFmtName(int(ctx.SampleFmt)),
				"sample_rate=" + strconv.Itoa(ctx.SampleRate),
				"time_base=" + ctx.TimeBase.String(),
			}
		case avcodec.AVMEDIA_TYPE_VIDEO:
			args = []string{
				"pix_fmt=" + strconv.Itoa(int(ctx.PixelFormat)),
				"pixel_aspect=" + ctx.SampleAspectRatio.String(),
				"time_base=" + ctx.TimeBase.String(),
				"video_size=" + strconv.Itoa(ctx.Width) + "x" + strconv.Itoa(ctx.Height),
			}
			if ctx.FrameRate.Num() > 0 {
				args = append(args, "frame_rate="+ctx.FrameRate.String())
			}
		default:
			err = fmt.Errorf("astilibav: codec type %v is not handled by filterer", ctx.CodecType)
			return
		}
		var bufferSrcCtx *avfilter.Context
		if ret := avfilter.AvfilterGraphCreateFilter(&bufferSrcCtx, bufferSrc, "in", strings.Join(args, ":"), nil, f.g); ret < 0 {
			err = fmt.Errorf("astilibav: avfilter.AvfilterGraphCreateFilter on args %s failed: %w", args, NewAvError(ret))
			return
		}
		outputs := avfilter.AvfilterInoutAlloc()
		outputs.SetName(n)
		outputs.SetFilterCtx(bufferSrcCtx)
		outputs.SetPadIdx(0)
		outputs.SetNext(previousOutput)
		f.bufferSrcCtxs[i] = append(f.bufferSrcCtxs[i], bufferSrcCtx)
		previousOutput = outputs
	}
	if ret := f.g.AvfilterGraphParsePtr(o.Content, &inputs, &previousOutput, nil); ret < 0 {
		err = fmt.Errorf("astilibav: g.AvfilterGraphParsePtr on content %s failed: %w", o.Content, NewAvError(ret))
		return
	}
	if ret := f.g.AvfilterGraphConfig(nil); ret < 0 {
		err = fmt.Errorf("astilibav: g.AvfilterGraphConfig failed: %w", NewAvError(ret))
		return
	}
	return
}
func (f *Filterer) Close() error {
	return f.cl.Close()
}
func (f *Filterer) addStats() {
	ss := f.c.Stats()
	ss = append(ss, f.d.stats()...)
	ss = append(ss,
		astikit.StatOptions{
			Handler: f.statIncomingRate,
			Metadata: &astikit.StatMetadata{
				Description: "Number of frames coming in per second",
				Label:       "Incoming rate",
				Name:        StatNameIncomingRate,
				Unit:        "fps",
			},
		},
		astikit.StatOptions{
			Handler: f.statProcessedRate,
			Metadata: &astikit.StatMetadata{
				Description: "Number of frames processed per second",
				Label:       "Processed rate",
				Name:        StatNameProcessedRate,
				Unit:        "fps",
			},
		},
	)
	f.BaseNode.AddStats(ss...)
}
func (f *Filterer) OutputCtx() Context {
	return f.outputCtx
}
func (f *Filterer) Connect(h FrameHandler) {
	f.d.addHandler(h)
	astiencoder.ConnectNodes(f, h)
}
func (f *Filterer) Disconnect(h FrameHandler) {
	f.d.delHandler(h)
	astiencoder.DisconnectNodes(f, h)
}
func (f *Filterer) Start(ctx context.Context, t astiencoder.CreateTaskFunc) {
	f.BaseNode.Start(ctx, t, func(t *astikit.Task) {
		if len(f.bufferSrcCtxs) == 0 {
			nextAt := time.Now()
			desc := newFiltererDescriptor(f.bufferSinkCtx, nil)
			for {
				if stop := f.tickFunc(&nextAt, desc); stop {
					break
				}
			}
			return
		}
		defer f.c.Stop()
		f.c.Start(f.Context())
	})
}
func (f *Filterer) tickFunc(nextAt *time.Time, desc Descriptor) (stop bool) {
	*nextAt = nextAt.Add(f.emulatePeriod)
	if delta := time.Until(*nextAt); delta > 0 {
		astikit.Sleep(f.Context(), delta)
	}
	if f.Context().Err() != nil {
		stop = true
		return
	}
	f.pullFilteredFrame(desc)
	return
}