pkg/flow/op.go

// Licensed to 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. Apache Software Foundation (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.

package flow

import (
	"context"
)

// UnaryOperation represents user-defined unary function (i.e. Map, Filter, etc).
type UnaryOperation[R any] interface {
	Apply(ctx context.Context, data interface{}) R
}

// UnaryFunc implements UnaryOperation as type func (context.Context, interface{}).
type UnaryFunc[R any] func(context.Context, interface{}) R

// Apply implements UnOperation.Apply method.
func (f UnaryFunc[R]) Apply(ctx context.Context, data interface{}) R {
	return f(ctx, data)
}

// FilterFunc transform a function to an UnaryOperation.
func FilterFunc(filter UnaryOperation[bool]) (UnaryOperation[any], error) {
	return UnaryFunc[any](func(ctx context.Context, payload interface{}) interface{} {
		predicate := filter.Apply(ctx, payload)
		if !predicate {
			return nil
		}
		return payload
	}), nil
}