// Licensed to the Apache Software Foundation (ASF) under one or more
// contributor license agreements.  See the NOTICE file distributed with
// this work for additional information regarding copyright ownership.
// The ASF licenses this file to You under the Apache License, Version 2.0
// (the "License"); you may not use this file except in compliance with
// the License.  You may obtain a copy of the License at
//
//    http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

// Code generated by starcgen. DO NOT EDIT.
// File: filter.shims.go

package filter

import (
	"context"
	"fmt"
	"io"
	"reflect"

	// Library imports
	"github.com/apache/beam/sdks/go/pkg/beam/core/runtime"
	"github.com/apache/beam/sdks/go/pkg/beam/core/runtime/exec"
	"github.com/apache/beam/sdks/go/pkg/beam/core/typex"
	"github.com/apache/beam/sdks/go/pkg/beam/core/util/reflectx"
)

func init() {
	runtime.RegisterFunction(keyFn)
	runtime.RegisterFunction(mapFn)
	runtime.RegisterType(reflect.TypeOf((*filterFn)(nil)).Elem())
	runtime.RegisterType(reflect.TypeOf((*typex.T)(nil)).Elem())
	reflectx.RegisterStructWrapper(reflect.TypeOf((*filterFn)(nil)).Elem(), wrapMakerFilterFn)
	reflectx.RegisterFunc(reflect.TypeOf((*func(typex.T,func(typex.T)) ())(nil)).Elem(), funcMakerTypex۰TEmitTypex۰TГ)
	reflectx.RegisterFunc(reflect.TypeOf((*func(typex.T,func(*int) bool) (typex.T))(nil)).Elem(), funcMakerTypex۰TIterIntГTypex۰T)
	reflectx.RegisterFunc(reflect.TypeOf((*func(typex.T) (typex.T,int))(nil)).Elem(), funcMakerTypex۰TГTypex۰TInt)
	reflectx.RegisterFunc(reflect.TypeOf((*func() ())(nil)).Elem(), funcMakerГ)
	exec.RegisterEmitter(reflect.TypeOf((*func(typex.T) ())(nil)).Elem(), emitMakerTypex۰T)
	exec.RegisterInput(reflect.TypeOf((*func(*int) (bool))(nil)).Elem(), iterMakerInt)
}

func wrapMakerFilterFn(fn interface{}) map[string]reflectx.Func {
	dfn := fn.(*filterFn)
	return map[string]reflectx.Func{
		"ProcessElement": reflectx.MakeFunc(func(a0 typex.T, a1 func(typex.T)) { dfn.ProcessElement(a0, a1) }),
		"Setup": reflectx.MakeFunc(func() { dfn.Setup() }),
	}
}

type callerTypex۰TEmitTypex۰TГ struct {
	fn func(typex.T,func(typex.T)) ()
}

func funcMakerTypex۰TEmitTypex۰TГ(fn interface{}) reflectx.Func {
	f := fn.(func(typex.T,func(typex.T)) ())
	return &callerTypex۰TEmitTypex۰TГ{fn: f}
}

func (c *callerTypex۰TEmitTypex۰TГ) Name() string {
	return reflectx.FunctionName(c.fn)
}

func (c *callerTypex۰TEmitTypex۰TГ) Type() reflect.Type {
	return reflect.TypeOf(c.fn)
}

func (c *callerTypex۰TEmitTypex۰TГ) Call(args []interface{}) []interface{} {
	c.fn(args[0].(typex.T), args[1].(func(typex.T)))
	return []interface{}{}
}

func (c *callerTypex۰TEmitTypex۰TГ) Call2x0(arg0, arg1 interface{}) () {
	c.fn(arg0.(typex.T), arg1.(func(typex.T)))
}

type callerTypex۰TIterIntГTypex۰T struct {
	fn func(typex.T,func(*int) bool) (typex.T)
}

func funcMakerTypex۰TIterIntГTypex۰T(fn interface{}) reflectx.Func {
	f := fn.(func(typex.T,func(*int) bool) (typex.T))
	return &callerTypex۰TIterIntГTypex۰T{fn: f}
}

func (c *callerTypex۰TIterIntГTypex۰T) Name() string {
	return reflectx.FunctionName(c.fn)
}

func (c *callerTypex۰TIterIntГTypex۰T) Type() reflect.Type {
	return reflect.TypeOf(c.fn)
}

func (c *callerTypex۰TIterIntГTypex۰T) Call(args []interface{}) []interface{} {
	out0 := c.fn(args[0].(typex.T), args[1].(func(*int) bool))
	return []interface{}{out0}
}

func (c *callerTypex۰TIterIntГTypex۰T) Call2x1(arg0, arg1 interface{}) (interface{}) {
	return c.fn(arg0.(typex.T), arg1.(func(*int) bool))
}

type callerTypex۰TГTypex۰TInt struct {
	fn func(typex.T) (typex.T,int)
}

func funcMakerTypex۰TГTypex۰TInt(fn interface{}) reflectx.Func {
	f := fn.(func(typex.T) (typex.T,int))
	return &callerTypex۰TГTypex۰TInt{fn: f}
}

func (c *callerTypex۰TГTypex۰TInt) Name() string {
	return reflectx.FunctionName(c.fn)
}

func (c *callerTypex۰TГTypex۰TInt) Type() reflect.Type {
	return reflect.TypeOf(c.fn)
}

func (c *callerTypex۰TГTypex۰TInt) Call(args []interface{}) []interface{} {
	out0, out1 := c.fn(args[0].(typex.T))
	return []interface{}{out0, out1}
}

func (c *callerTypex۰TГTypex۰TInt) Call1x2(arg0 interface{}) (interface{}, interface{}) {
	return c.fn(arg0.(typex.T))
}

type callerГ struct {
	fn func() ()
}

func funcMakerГ(fn interface{}) reflectx.Func {
	f := fn.(func() ())
	return &callerГ{fn: f}
}

func (c *callerГ) Name() string {
	return reflectx.FunctionName(c.fn)
}

func (c *callerГ) Type() reflect.Type {
	return reflect.TypeOf(c.fn)
}

func (c *callerГ) Call(args []interface{}) []interface{} {
	c.fn()
	return []interface{}{}
}

func (c *callerГ) Call0x0() () {
	c.fn()
}

type emitNative struct {
	n     exec.ElementProcessor
	fn    interface{}

	ctx context.Context
	ws  []typex.Window
	et  typex.EventTime
}

func (e *emitNative) Init(ctx context.Context, ws []typex.Window, et typex.EventTime) error {
	e.ctx = ctx
	e.ws = ws
	e.et = et
	return nil
}

func (e *emitNative) Value() interface{} {
	return e.fn
}

func emitMakerTypex۰T(n exec.ElementProcessor) exec.ReusableEmitter {
	ret := &emitNative{n: n}
	ret.fn = ret.invokeTypex۰T
	return ret
}

func (e *emitNative) invokeTypex۰T(val typex.T) {
	value := exec.FullValue{Windows: e.ws, Timestamp: e.et, Elm: val}
	if err := e.n.ProcessElement(e.ctx, value); err != nil {
		panic(err)
	}
}

type iterNative struct {
	s     exec.ReStream
	fn    interface{}

	// cur is the "current" stream, if any.
	cur exec.Stream
}

func (v *iterNative) Init() error {
	cur, err := v.s.Open()
	if err != nil {
		return err
	}
	v.cur = cur
	return nil
}

func (v *iterNative) Value() interface{} {
	return v.fn
}

func convToString(v interface{}) string {
	switch v.(type) {
	case []byte:
		return string(v.([]byte))
	default:
		return v.(string)
	}
}

func (v *iterNative) Reset() error {
	if err := v.cur.Close(); err != nil {
		return err
	}
	v.cur = nil
	return nil
}

func iterMakerInt(s exec.ReStream) exec.ReusableInput {
	ret := &iterNative{s: s}
	ret.fn = ret.readInt
	return ret
}

func (v *iterNative) readInt(value *int) bool {
	elm, err := v.cur.Read()
	if err != nil {
		if err == io.EOF {
			return false
		}
		panic(fmt.Sprintf("broken stream: %v", err))
	}
	*value = elm.Elm.(int)
	return true
}


// DO NOT MODIFY: GENERATED CODE
