// 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: beam.shims.go

package beam

import (
	"context"
	"reflect"

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

func init() {
	runtime.RegisterFunction(addFixedKeyFn)
	runtime.RegisterFunction(dropKeyFn)
	runtime.RegisterFunction(dropValueFn)
	runtime.RegisterFunction(explodeFn)
	runtime.RegisterFunction(jsonDec)
	runtime.RegisterFunction(jsonEnc)
	runtime.RegisterFunction(makePartitionFn)
	runtime.RegisterFunction(protoDec)
	runtime.RegisterFunction(protoEnc)
	runtime.RegisterFunction(schemaDec)
	runtime.RegisterFunction(schemaEnc)
	runtime.RegisterFunction(swapKVFn)
	runtime.RegisterType(reflect.TypeOf((*T)(nil)).Elem())
	schema.RegisterType(reflect.TypeOf((*T)(nil)).Elem())
	runtime.RegisterType(reflect.TypeOf((*X)(nil)).Elem())
	schema.RegisterType(reflect.TypeOf((*X)(nil)).Elem())
	runtime.RegisterType(reflect.TypeOf((*Y)(nil)).Elem())
	schema.RegisterType(reflect.TypeOf((*Y)(nil)).Elem())
	runtime.RegisterType(reflect.TypeOf((*reflect.Type)(nil)).Elem())
	schema.RegisterType(reflect.TypeOf((*reflect.Type)(nil)).Elem())
	runtime.RegisterType(reflect.TypeOf((*reflectx.Func)(nil)).Elem())
	schema.RegisterType(reflect.TypeOf((*reflectx.Func)(nil)).Elem())
	reflectx.RegisterFunc(reflect.TypeOf((*func(reflect.Type, []byte) (T, error))(nil)).Elem(), funcMakerReflect۰TypeSliceOfByteГTError)
	reflectx.RegisterFunc(reflect.TypeOf((*func(reflect.Type, T) ([]byte, error))(nil)).Elem(), funcMakerReflect۰TypeTГSliceOfByteError)
	reflectx.RegisterFunc(reflect.TypeOf((*func([]T, func(T)))(nil)).Elem(), funcMakerSliceOfTEmitTГ)
	reflectx.RegisterFunc(reflect.TypeOf((*func(string, reflect.Type, []byte) reflectx.Func)(nil)).Elem(), funcMakerStringReflect۰TypeSliceOfByteГReflectx۰Func)
	reflectx.RegisterFunc(reflect.TypeOf((*func(T) (int, T))(nil)).Elem(), funcMakerTГIntT)
	reflectx.RegisterFunc(reflect.TypeOf((*func(T) ([]byte, error))(nil)).Elem(), funcMakerTГSliceOfByteError)
	reflectx.RegisterFunc(reflect.TypeOf((*func(X, Y) X)(nil)).Elem(), funcMakerXYГX)
	reflectx.RegisterFunc(reflect.TypeOf((*func(X, Y) Y)(nil)).Elem(), funcMakerXYГY)
	reflectx.RegisterFunc(reflect.TypeOf((*func(X, Y) (Y, X))(nil)).Elem(), funcMakerXYГYX)
	exec.RegisterEmitter(reflect.TypeOf((*func(T))(nil)).Elem(), emitMakerT)
}

type callerReflect۰TypeSliceOfByteГTError struct {
	fn func(reflect.Type, []byte) (T, error)
}

func funcMakerReflect۰TypeSliceOfByteГTError(fn any) reflectx.Func {
	f := fn.(func(reflect.Type, []byte) (T, error))
	return &callerReflect۰TypeSliceOfByteГTError{fn: f}
}

func (c *callerReflect۰TypeSliceOfByteГTError) Name() string {
	return reflectx.FunctionName(c.fn)
}

func (c *callerReflect۰TypeSliceOfByteГTError) Type() reflect.Type {
	return reflect.TypeOf(c.fn)
}

func (c *callerReflect۰TypeSliceOfByteГTError) Call(args []any) []any {
	out0, out1 := c.fn(args[0].(reflect.Type), args[1].([]byte))
	return []any{out0, out1}
}

func (c *callerReflect۰TypeSliceOfByteГTError) Call2x2(arg0, arg1 any) (any, any) {
	return c.fn(arg0.(reflect.Type), arg1.([]byte))
}

type callerReflect۰TypeTГSliceOfByteError struct {
	fn func(reflect.Type, T) ([]byte, error)
}

func funcMakerReflect۰TypeTГSliceOfByteError(fn any) reflectx.Func {
	f := fn.(func(reflect.Type, T) ([]byte, error))
	return &callerReflect۰TypeTГSliceOfByteError{fn: f}
}

func (c *callerReflect۰TypeTГSliceOfByteError) Name() string {
	return reflectx.FunctionName(c.fn)
}

func (c *callerReflect۰TypeTГSliceOfByteError) Type() reflect.Type {
	return reflect.TypeOf(c.fn)
}

func (c *callerReflect۰TypeTГSliceOfByteError) Call(args []any) []any {
	out0, out1 := c.fn(args[0].(reflect.Type), args[1].(T))
	return []any{out0, out1}
}

func (c *callerReflect۰TypeTГSliceOfByteError) Call2x2(arg0, arg1 any) (any, any) {
	return c.fn(arg0.(reflect.Type), arg1.(T))
}

type callerSliceOfTEmitTГ struct {
	fn func([]T, func(T))
}

func funcMakerSliceOfTEmitTГ(fn any) reflectx.Func {
	f := fn.(func([]T, func(T)))
	return &callerSliceOfTEmitTГ{fn: f}
}

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

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

func (c *callerSliceOfTEmitTГ) Call(args []any) []any {
	c.fn(args[0].([]T), args[1].(func(T)))
	return []any{}
}

func (c *callerSliceOfTEmitTГ) Call2x0(arg0, arg1 any) {
	c.fn(arg0.([]T), arg1.(func(T)))
}

type callerStringReflect۰TypeSliceOfByteГReflectx۰Func struct {
	fn func(string, reflect.Type, []byte) reflectx.Func
}

func funcMakerStringReflect۰TypeSliceOfByteГReflectx۰Func(fn any) reflectx.Func {
	f := fn.(func(string, reflect.Type, []byte) reflectx.Func)
	return &callerStringReflect۰TypeSliceOfByteГReflectx۰Func{fn: f}
}

func (c *callerStringReflect۰TypeSliceOfByteГReflectx۰Func) Name() string {
	return reflectx.FunctionName(c.fn)
}

func (c *callerStringReflect۰TypeSliceOfByteГReflectx۰Func) Type() reflect.Type {
	return reflect.TypeOf(c.fn)
}

func (c *callerStringReflect۰TypeSliceOfByteГReflectx۰Func) Call(args []any) []any {
	out0 := c.fn(args[0].(string), args[1].(reflect.Type), args[2].([]byte))
	return []any{out0}
}

func (c *callerStringReflect۰TypeSliceOfByteГReflectx۰Func) Call3x1(arg0, arg1, arg2 any) any {
	return c.fn(arg0.(string), arg1.(reflect.Type), arg2.([]byte))
}

type callerTГIntT struct {
	fn func(T) (int, T)
}

func funcMakerTГIntT(fn any) reflectx.Func {
	f := fn.(func(T) (int, T))
	return &callerTГIntT{fn: f}
}

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

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

func (c *callerTГIntT) Call(args []any) []any {
	out0, out1 := c.fn(args[0].(T))
	return []any{out0, out1}
}

func (c *callerTГIntT) Call1x2(arg0 any) (any, any) {
	return c.fn(arg0.(T))
}

type callerTГSliceOfByteError struct {
	fn func(T) ([]byte, error)
}

func funcMakerTГSliceOfByteError(fn any) reflectx.Func {
	f := fn.(func(T) ([]byte, error))
	return &callerTГSliceOfByteError{fn: f}
}

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

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

func (c *callerTГSliceOfByteError) Call(args []any) []any {
	out0, out1 := c.fn(args[0].(T))
	return []any{out0, out1}
}

func (c *callerTГSliceOfByteError) Call1x2(arg0 any) (any, any) {
	return c.fn(arg0.(T))
}

type callerXYГX struct {
	fn func(X, Y) X
}

func funcMakerXYГX(fn any) reflectx.Func {
	f := fn.(func(X, Y) X)
	return &callerXYГX{fn: f}
}

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

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

func (c *callerXYГX) Call(args []any) []any {
	out0 := c.fn(args[0].(X), args[1].(Y))
	return []any{out0}
}

func (c *callerXYГX) Call2x1(arg0, arg1 any) any {
	return c.fn(arg0.(X), arg1.(Y))
}

type callerXYГY struct {
	fn func(X, Y) Y
}

func funcMakerXYГY(fn any) reflectx.Func {
	f := fn.(func(X, Y) Y)
	return &callerXYГY{fn: f}
}

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

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

func (c *callerXYГY) Call(args []any) []any {
	out0 := c.fn(args[0].(X), args[1].(Y))
	return []any{out0}
}

func (c *callerXYГY) Call2x1(arg0, arg1 any) any {
	return c.fn(arg0.(X), arg1.(Y))
}

type callerXYГYX struct {
	fn func(X, Y) (Y, X)
}

func funcMakerXYГYX(fn any) reflectx.Func {
	f := fn.(func(X, Y) (Y, X))
	return &callerXYГYX{fn: f}
}

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

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

func (c *callerXYГYX) Call(args []any) []any {
	out0, out1 := c.fn(args[0].(X), args[1].(Y))
	return []any{out0, out1}
}

func (c *callerXYГYX) Call2x2(arg0, arg1 any) (any, any) {
	return c.fn(arg0.(X), arg1.(Y))
}

type emitNative struct {
	n   exec.ElementProcessor
	fn  any
	est *sdf.WatermarkEstimator

	ctx   context.Context
	pn    typex.PaneInfo
	ws    []typex.Window
	et    typex.EventTime
	value exec.FullValue
}

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

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

func (e *emitNative) AttachEstimator(est *sdf.WatermarkEstimator) {
	e.est = est
}

func emitMakerT(n exec.ElementProcessor) exec.ReusableEmitter {
	ret := &emitNative{n: n}
	ret.fn = ret.invokeT
	return ret
}

func (e *emitNative) invokeT(val T) {
	e.value = exec.FullValue{Pane: e.pn, Windows: e.ws, Timestamp: e.et, Elm: val}
	if e.est != nil {
		(*e.est).(sdf.TimestampObservingEstimator).ObserveTimestamp(e.et.ToTime())
	}
	if err := e.n.ProcessElement(e.ctx, &e.value); err != nil {
		panic(err)
	}
}

// DO NOT MODIFY: GENERATED CODE
