whitaker-io / machine Goto Github PK

// Monad is a function that is applied to payload and used for transformations
type Monad[T any] func(d T) T

// Filter is a function that can be used to filter the payload.
type Filter[T any] func(d T) bool

// New is a function for creating a new Machine.
//
// name string
// input chan T
// option ...Option
//
// Call the startFn returned by New to start the Machine once built.
func New[T any](name string, input chan T, options ...Option) (startFn func(context.Context), x Machine[T])

// Transform is a function for converting the type of the Machine. Cannot be used inside a loop
// until I figure out how to do it without some kind of run time error or overly complex
// tracking method that isn't type safe. I really wish method level generics were a thing.
func Transform[T, U any](m Machine[T], fn func(d T) U) (Machine[U], error)

// Machine is the interface provided for creating a data processing stream.
type Machine[T any] interface {
	// Name returns the name of the Machine path. Useful for debugging or reasoning about the path.
	Name() string
	
	// Then apply a mutation to each individual element of the payload.
	Then(a Monad[T]) Machine[T]

	// Recurse applies a recursive function to the payload through a Y Combinator.
	// f is a function used by the Y Combinator to perform a recursion
	// on the payload.
	// Example:
	//
	//	func(f Monad[int]) Monad[int] {
	//		 return func(x int) int {
	//			 if x <= 0 {
	//				 return 1
	//			 } else {
	//				 return x * f(x-1)
	//			 }
	//		 }
	//	}
	Recurse(x Monad[Monad[T]]) Machine[T]

	// Memoize applies a recursive function to the payload through a Y Combinator
	// and memoizes the results based on the index func.
	// f is a function used by the Y Combinator to perform a recursion
	// on the payload.
	// Example:
	//
	//	func(f Monad[int]) Monad[int] {
	//		 return func(x int) int {
	//			 if x <= 0 {
	//				 return 1
	//			 } else {
	//				 return x * f(x-1)
	//			 }
	//		 }
	//	}
	Memoize(x Monad[Monad[T]], index func(T) string) Machine[T]

	// Or runs all of the functions until one succeeds or sends the payload to the right branch
	Or(x ...Filter[T]) (Machine[T], Machine[T])

	// And runs all of the functions and if one doesnt succeed sends the payload to the right branch
	And(x ...Filter[T]) (Machine[T], Machine[T])

	// Filter splits the data into multiple stream branches
	If(f Filter[T]) (Machine[T], Machine[T])

	// Select applies a series of Filters to the payload and returns a list of Builders
	// the last one being for any unmatched payloads.
	Select(fns ...Filter[T]) []Machine[T]

	// Tee duplicates the data into multiple stream branches.
	Tee(func(T) (a, b T)) (Machine[T], Machine[T])

	// While creates a loop in the stream based on the filter
	While(x Filter[T]) (loop, out Machine[T])

	// Drop terminates the data from further processing without passing it on
	Drop()

	// Distribute is a function used for fanout
	Distribute(Edge[T]) Machine[T]
	
	// Output provided channel
	Output() chan T
}

// Edge is an interface that is used for transferring data between vertices
type Edge[T any] interface {
	Output() chan T
	Send(payload T)
}

// Option is used to configure the machine
type Option interface

// OptionFIF0 controls the processing order of the payloads
// If set to true the system will wait for one payload
// to be processed before starting the next.
var OptionFIF0 Option

// OptionBufferSize sets the buffer size on the edge channels between the
// vertices, this setting can be useful when processing large amounts
// of data with FIFO turned on.
func OptionBufferSize(size int) Option

// OptionAttributes apply the slog.Attr's to the machine metrics and spans
// Do not override the "name", "type", "duration", "error", or "value" attributes
func OptionAttributes(attributes ...slog.Attr) Option

// OptionFlush attempts to send all data to the flushFN before exiting after the gracePeriod has expired
// Im looking for a good way to make this type specific, but want to avoid having to add separate option
// settings for the Transform function.
func OptionFlush(gracePeriod time.Duration, flushFN func(vertexName string, payload any)) Option

// import "github.com/whitaker-io/machine/telemetry"

// Make your slog handler however you please
yourSlogHandler := slog.Default().Handler()

// wrap your handler and provide your tracer and meter
telemetryHandler := telemetry.New(
	yourSlogHandler,
	meterProvider.Meter("your_meter"), // Your otel metric.Meter
	tracerProvider.Tracer("your_tracer"), // Your otel trace.Tracer
	false, // Log Metrics and Traces to logs as well (useful for debugging)
)

slog.SetDefault(slog.New(telemetryHandler))

// import "github.com/whitaker-io/machine/edge/pubsub"

func New[T any](
	ctx context.Context,
	subscription *pubsub.Subscription,
	publisher *pubsub.Topic,
	to func(T) *pubsub.Message,
	from func(context.Context, *pubsub.Message) T,
) machine.Edge[T]

// import "github.com/whitaker-io/machine/edge/http"

func New[T any](c http.Client, fn func(context.Context, T) *http.Request) machine.Edge[T]