Installation

  go get github.com/whitaker-io/machine/v3

// Monad is a function that is applied to payload and used for transformationstypeMonad[Tany]func(dT)T// Filter is a function that can be used to filter the payload.typeFilter[Tany]func(dT)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.funcNew[Tany](namestring,inputchanT,options...Option) (startFnfunc(context.Context),xMachine[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.funcTransform[T,Uany](mMachine[T],fnfunc(dT)U) (Machine[U],error)// Machine is the interface provided for creating a data processing stream.typeMachine[Tany]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(aMonad[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(xMonad[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(xMonad[Monad[T]],indexfunc(T)string)Machine[T]// Or runs all of the functions until one succeeds or sends the payload to the right branchOr(x...Filter[T]) (Machine[T],Machine[T])// And runs all of the functions and if one doesnt succeed sends the payload to the right branchAnd(x...Filter[T]) (Machine[T],Machine[T])// Filter splits the data into multiple stream branchesIf(fFilter[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,bT)) (Machine[T],Machine[T])// While creates a loop in the stream based on the filterWhile(xFilter[T]) (loop,outMachine[T])// Drop terminates the data from further processing without passing it onDrop()// Distribute is a function used for fanoutDistribute(Edge[T])Machine[T]// Output provided channelOutput()chanT}

// Edge is an interface that is used for transferring data between verticestypeEdge[Tany]interface {Output()chanTSend(payloadT)}

// Option is used to configure the machinetypeOptioninterface// 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.varOptionFIF0Option// 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.funcOptionBufferSize(sizeint)Option// OptionAttributes apply the slog.Attr's to the machine metrics and spans// Do not override the "name", "type", "duration", "error", or "value" attributesfuncOptionAttributes(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.funcOptionFlush(gracePeriod time.Duration,flushFNfunc(vertexNamestring,payloadany))Option

// import "github.com/whitaker-io/machine/telemetry"// Make your slog handler however you pleaseyourSlogHandler:=slog.Default().Handler()// wrap your handler and provide your tracer and metertelemetryHandler:=telemetry.New(yourSlogHandler,meterProvider.Meter("your_meter"),// Your otel metric.MetertracerProvider.Tracer("your_tracer"),// Your otel trace.Tracerfalse,// Log Metrics and Traces to logs as well (useful for debugging))slog.SetDefault(slog.New(telemetryHandler))

// import "github.com/whitaker-io/machine/edge/pubsub"funcNew[Tany](ctx context.Context,subscription*pubsub.Subscription,publisher*pubsub.Topic,tofunc(T)*pubsub.Message,fromfunc(context.Context,*pubsub.Message)T,) machine.Edge[T]// import "github.com/whitaker-io/machine/edge/http"funcNew[Tany](c http.Client,fnfunc(context.Context,T)*http.Request) machine.Edge[T]

🤝 Contributing

Author

Show your support

License

The MIT License (MIT)Copyright (c) 2020 Jonathan Whitaker