debug.shims.go

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

package debug

import (
	"context"
	"fmt"
	"io"
	"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(discardFn)
	runtime.RegisterType(reflect.TypeOf((*context.Context)(nil)).Elem())
	schema.RegisterType(reflect.TypeOf((*context.Context)(nil)).Elem())
	runtime.RegisterType(reflect.TypeOf((*headFn)(nil)).Elem())
	schema.RegisterType(reflect.TypeOf((*headFn)(nil)).Elem())
	runtime.RegisterType(reflect.TypeOf((*headKVFn)(nil)).Elem())
	schema.RegisterType(reflect.TypeOf((*headKVFn)(nil)).Elem())
	runtime.RegisterType(reflect.TypeOf((*printFn)(nil)).Elem())
	schema.RegisterType(reflect.TypeOf((*printFn)(nil)).Elem())
	runtime.RegisterType(reflect.TypeOf((*printGBKFn)(nil)).Elem())
	schema.RegisterType(reflect.TypeOf((*printGBKFn)(nil)).Elem())
	runtime.RegisterType(reflect.TypeOf((*printKVFn)(nil)).Elem())
	schema.RegisterType(reflect.TypeOf((*printKVFn)(nil)).Elem())
	reflectx.RegisterStructWrapper(reflect.TypeOf((*headFn)(nil)).Elem(), wrapMakerHeadFn)
	reflectx.RegisterStructWrapper(reflect.TypeOf((*headKVFn)(nil)).Elem(), wrapMakerHeadKVFn)
	reflectx.RegisterStructWrapper(reflect.TypeOf((*printFn)(nil)).Elem(), wrapMakerPrintFn)
	reflectx.RegisterStructWrapper(reflect.TypeOf((*printGBKFn)(nil)).Elem(), wrapMakerPrintGBKFn)
	reflectx.RegisterStructWrapper(reflect.TypeOf((*printKVFn)(nil)).Elem(), wrapMakerPrintKVFn)
	reflectx.RegisterFunc(reflect.TypeOf((*func(context.Context, typex.T) typex.T)(nil)).Elem(), funcMakerContext۰ContextTypex۰TГTypex۰T)
	reflectx.RegisterFunc(reflect.TypeOf((*func(context.Context, typex.X, func(*typex.Y) bool) typex.X)(nil)).Elem(), funcMakerContext۰ContextTypex۰XIterTypex۰YГTypex۰X)
	reflectx.RegisterFunc(reflect.TypeOf((*func(context.Context, typex.X, typex.Y) (typex.X, typex.Y))(nil)).Elem(), funcMakerContext۰ContextTypex۰XTypex۰YГTypex۰XTypex۰Y)
	reflectx.RegisterFunc(reflect.TypeOf((*func([]byte, func(*typex.T) bool, func(typex.T)))(nil)).Elem(), funcMakerSliceOfByteIterTypex۰TEmitTypex۰TГ)
	reflectx.RegisterFunc(reflect.TypeOf((*func([]byte, func(*typex.X, *typex.Y) bool, func(typex.X, typex.Y)))(nil)).Elem(), funcMakerSliceOfByteIterTypex۰XTypex۰YEmitTypex۰XTypex۰YГ)
	reflectx.RegisterFunc(reflect.TypeOf((*func(typex.T))(nil)).Elem(), funcMakerTypex۰TГ)
	exec.RegisterEmitter(reflect.TypeOf((*func(typex.T))(nil)).Elem(), emitMakerTypex۰T)
	exec.RegisterEmitter(reflect.TypeOf((*func(typex.X, typex.Y))(nil)).Elem(), emitMakerTypex۰XTypex۰Y)
	exec.RegisterInput(reflect.TypeOf((*func(*typex.T) bool)(nil)).Elem(), iterMakerTypex۰T)
	exec.RegisterInput(reflect.TypeOf((*func(*typex.X, *typex.Y) bool)(nil)).Elem(), iterMakerTypex۰XTypex۰Y)
	exec.RegisterInput(reflect.TypeOf((*func(*typex.Y) bool)(nil)).Elem(), iterMakerTypex۰Y)
}

func wrapMakerHeadFn(fn interface{}) map[string]reflectx.Func {
	dfn := fn.(*headFn)
	return map[string]reflectx.Func{
		"ProcessElement": reflectx.MakeFunc(func(a0 []byte, a1 func(*typex.T) bool, a2 func(typex.T)) { dfn.ProcessElement(a0, a1, a2) }),
	}
}

func wrapMakerHeadKVFn(fn interface{}) map[string]reflectx.Func {
	dfn := fn.(*headKVFn)
	return map[string]reflectx.Func{
		"ProcessElement": reflectx.MakeFunc(func(a0 []byte, a1 func(*typex.X, *typex.Y) bool, a2 func(typex.X, typex.Y)) {
			dfn.ProcessElement(a0, a1, a2)
		}),
	}
}

func wrapMakerPrintFn(fn interface{}) map[string]reflectx.Func {
	dfn := fn.(*printFn)
	return map[string]reflectx.Func{
		"ProcessElement": reflectx.MakeFunc(func(a0 context.Context, a1 typex.T) typex.T { return dfn.ProcessElement(a0, a1) }),
	}
}

func wrapMakerPrintGBKFn(fn interface{}) map[string]reflectx.Func {
	dfn := fn.(*printGBKFn)
	return map[string]reflectx.Func{
		"ProcessElement": reflectx.MakeFunc(func(a0 context.Context, a1 typex.X, a2 func(*typex.Y) bool) typex.X {
			return dfn.ProcessElement(a0, a1, a2)
		}),
	}
}

func wrapMakerPrintKVFn(fn interface{}) map[string]reflectx.Func {
	dfn := fn.(*printKVFn)
	return map[string]reflectx.Func{
		"ProcessElement": reflectx.MakeFunc(func(a0 context.Context, a1 typex.X, a2 typex.Y) (typex.X, typex.Y) {
			return dfn.ProcessElement(a0, a1, a2)
		}),
	}
}

type callerContext۰ContextTypex۰TГTypex۰T struct {
	fn func(context.Context, typex.T) typex.T
}

func funcMakerContext۰ContextTypex۰TГTypex۰T(fn interface{}) reflectx.Func {
	f := fn.(func(context.Context, typex.T) typex.T)
	return &callerContext۰ContextTypex۰TГTypex۰T{fn: f}
}

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

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

func (c *callerContext۰ContextTypex۰TГTypex۰T) Call(args []interface{}) []interface{} {
	out0 := c.fn(args[0].(context.Context), args[1].(typex.T))
	return []interface{}{out0}
}

func (c *callerContext۰ContextTypex۰TГTypex۰T) Call2x1(arg0, arg1 interface{}) interface{} {
	return c.fn(arg0.(context.Context), arg1.(typex.T))
}

type callerContext۰ContextTypex۰XIterTypex۰YГTypex۰X struct {
	fn func(context.Context, typex.X, func(*typex.Y) bool) typex.X
}

func funcMakerContext۰ContextTypex۰XIterTypex۰YГTypex۰X(fn interface{}) reflectx.Func {
	f := fn.(func(context.Context, typex.X, func(*typex.Y) bool) typex.X)
	return &callerContext۰ContextTypex۰XIterTypex۰YГTypex۰X{fn: f}
}

func (c *callerContext۰ContextTypex۰XIterTypex۰YГTypex۰X) Name() string {
	return reflectx.FunctionName(c.fn)
}

func (c *callerContext۰ContextTypex۰XIterTypex۰YГTypex۰X) Type() reflect.Type {
	return reflect.TypeOf(c.fn)
}

func (c *callerContext۰ContextTypex۰XIterTypex۰YГTypex۰X) Call(args []interface{}) []interface{} {
	out0 := c.fn(args[0].(context.Context), args[1].(typex.X), args[2].(func(*typex.Y) bool))
	return []interface{}{out0}
}

func (c *callerContext۰ContextTypex۰XIterTypex۰YГTypex۰X) Call3x1(arg0, arg1, arg2 interface{}) interface{} {
	return c.fn(arg0.(context.Context), arg1.(typex.X), arg2.(func(*typex.Y) bool))
}

type callerContext۰ContextTypex۰XTypex۰YГTypex۰XTypex۰Y struct {
	fn func(context.Context, typex.X, typex.Y) (typex.X, typex.Y)
}

func funcMakerContext۰ContextTypex۰XTypex۰YГTypex۰XTypex۰Y(fn interface{}) reflectx.Func {
	f := fn.(func(context.Context, typex.X, typex.Y) (typex.X, typex.Y))
	return &callerContext۰ContextTypex۰XTypex۰YГTypex۰XTypex۰Y{fn: f}
}

func (c *callerContext۰ContextTypex۰XTypex۰YГTypex۰XTypex۰Y) Name() string {
	return reflectx.FunctionName(c.fn)
}

func (c *callerContext۰ContextTypex۰XTypex۰YГTypex۰XTypex۰Y) Type() reflect.Type {
	return reflect.TypeOf(c.fn)
}

func (c *callerContext۰ContextTypex۰XTypex۰YГTypex۰XTypex۰Y) Call(args []interface{}) []interface{} {
	out0, out1 := c.fn(args[0].(context.Context), args[1].(typex.X), args[2].(typex.Y))
	return []interface{}{out0, out1}
}

func (c *callerContext۰ContextTypex۰XTypex۰YГTypex۰XTypex۰Y) Call3x2(arg0, arg1, arg2 interface{}) (interface{}, interface{}) {
	return c.fn(arg0.(context.Context), arg1.(typex.X), arg2.(typex.Y))
}

type callerSliceOfByteIterTypex۰TEmitTypex۰TГ struct {
	fn func([]byte, func(*typex.T) bool, func(typex.T))
}

func funcMakerSliceOfByteIterTypex۰TEmitTypex۰TГ(fn interface{}) reflectx.Func {
	f := fn.(func([]byte, func(*typex.T) bool, func(typex.T)))
	return &callerSliceOfByteIterTypex۰TEmitTypex۰TГ{fn: f}
}

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

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

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

func (c *callerSliceOfByteIterTypex۰TEmitTypex۰TГ) Call3x0(arg0, arg1, arg2 interface{}) {
	c.fn(arg0.([]byte), arg1.(func(*typex.T) bool), arg2.(func(typex.T)))
}

type callerSliceOfByteIterTypex۰XTypex۰YEmitTypex۰XTypex۰YГ struct {
	fn func([]byte, func(*typex.X, *typex.Y) bool, func(typex.X, typex.Y))
}

func funcMakerSliceOfByteIterTypex۰XTypex۰YEmitTypex۰XTypex۰YГ(fn interface{}) reflectx.Func {
	f := fn.(func([]byte, func(*typex.X, *typex.Y) bool, func(typex.X, typex.Y)))
	return &callerSliceOfByteIterTypex۰XTypex۰YEmitTypex۰XTypex۰YГ{fn: f}
}

func (c *callerSliceOfByteIterTypex۰XTypex۰YEmitTypex۰XTypex۰YГ) Name() string {
	return reflectx.FunctionName(c.fn)
}

func (c *callerSliceOfByteIterTypex۰XTypex۰YEmitTypex۰XTypex۰YГ) Type() reflect.Type {
	return reflect.TypeOf(c.fn)
}

func (c *callerSliceOfByteIterTypex۰XTypex۰YEmitTypex۰XTypex۰YГ) Call(args []interface{}) []interface{} {
	c.fn(args[0].([]byte), args[1].(func(*typex.X, *typex.Y) bool), args[2].(func(typex.X, typex.Y)))
	return []interface{}{}
}

func (c *callerSliceOfByteIterTypex۰XTypex۰YEmitTypex۰XTypex۰YГ) Call3x0(arg0, arg1, arg2 interface{}) {
	c.fn(arg0.([]byte), arg1.(func(*typex.X, *typex.Y) bool), arg2.(func(typex.X, typex.Y)))
}

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

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

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

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

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

func (c *callerTypex۰TГ) Call1x0(arg0 interface{}) {
	c.fn(arg0.(typex.T))
}

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

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

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 (e *emitNative) AttachEstimator(est *sdf.WatermarkEstimator) {
	e.est = est
}

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) {
	e.value = exec.FullValue{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)
	}
}

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

func (e *emitNative) invokeTypex۰XTypex۰Y(key typex.X, val typex.Y) {
	e.value = exec.FullValue{Windows: e.ws, Timestamp: e.et, Elm: key, Elm2: 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)
	}
}

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 (v *iterNative) Reset() error {
	if err := v.cur.Close(); err != nil {
		return err
	}
	v.cur = nil
	return nil
}

func iterMakerTypex۰T(s exec.ReStream) exec.ReusableInput {
	ret := &iterNative{s: s}
	ret.fn = ret.readTypex۰T
	return ret
}

func (v *iterNative) readTypex۰T(value *typex.T) 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.(typex.T)
	return true
}

func iterMakerTypex۰XTypex۰Y(s exec.ReStream) exec.ReusableInput {
	ret := &iterNative{s: s}
	ret.fn = ret.readTypex۰XTypex۰Y
	return ret
}

func (v *iterNative) readTypex۰XTypex۰Y(key *typex.X, value *typex.Y) bool {
	elm, err := v.cur.Read()
	if err != nil {
		if err == io.EOF {
			return false
		}
		panic(fmt.Sprintf("broken stream: %v", err))
	}
	*key = elm.Elm.(typex.X)
	*value = elm.Elm2.(typex.Y)
	return true
}

func iterMakerTypex۰Y(s exec.ReStream) exec.ReusableInput {
	ret := &iterNative{s: s}
	ret.fn = ret.readTypex۰Y
	return ret
}

func (v *iterNative) readTypex۰Y(value *typex.Y) 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.(typex.Y)
	return true
}

// DO NOT MODIFY: GENERATED CODE