package filter

import (
	"reflect"

	"github.com/childe/gohangout/topology"
	"k8s.io/klog/v2"
)

type FiltersFilter struct {
	config        map[any]any
	processorNode *topology.ProcessorNode
	filterBoxes   []*topology.FilterBox
}

func init() {
	Register("Filters", newFiltersFilter)
}

func newFiltersFilter(config map[any]any) topology.Filter {
	f := &FiltersFilter{
		config: config,
	}

	_config := make(map[string]any)
	for k, v := range config {
		_config[k.(string)] = v
	}

	f.filterBoxes = topology.BuildFilterBoxes(_config, BuildFilter)
	if len(f.filterBoxes) == 0 {
		klog.Fatal("no filters configured in Filters")
	}

	for _, b := range f.filterBoxes {
		f.processorNode = topology.AppendProcessorsToLink(f.processorNode, b)
	}

	return f
}

func (f *FiltersFilter) Filter(event map[string]any) (map[string]any, bool) {
	return f.processorNode.Process(event), true
}

func (f *FiltersFilter) SetBelongTo(next topology.Processor) {
	var b *topology.FilterBox = f.filterBoxes[len(f.filterBoxes)-1]
	v := reflect.ValueOf(b.Filter)
	fun := v.MethodByName("SetBelongTo")
	if fun.IsValid() {
		fun.Call([]reflect.Value{reflect.ValueOf(next)})
	}
}
