pipeline

The example uses dummy data and simulates a specific (random) scenario were we need to get a resource that is created from two different data sources (and we can obtain them concurrently to improve the performance) and finally reduce it to a single output

Note: we use several UnitStep to showcase as it allows us to easily run dummy code, but it could use any type of step you want as long as it implements pipeline.Step[I, O]

type DriverID int
type Driver struct {
	Person  any
	Vehicle any
}

gp := pipeline.NewUnitStep(
	"get_person",
	func(ctx context.Context, i DriverID) (Driver, error) {
		// do something with input
		return Driver{
			Person: true,
		}, nil
	},
)
gv := pipeline.NewUnitStep(
	"get_vehicle",
	func(ctx context.Context, i DriverID) (Driver, error) {
		// do something with input
		return Driver{
			Vehicle: true,
		}, nil
	},
)
reduce := func(ctx context.Context, a, b Driver) (Driver, error) {
	if b.Person != nil {
		a.Person = b.Person
	}
	if b.Vehicle != nil {
		a.Vehicle = b.Vehicle
	}
	return a, nil
}
ctx := context.Background()
in := DriverID(1)

pipe := pipeline.NewConcurrentStep(
	[]pipeline.Step[DriverID, Driver]{gp, gv},
	reduce,
)

out, err := pipe.Run(ctx, in)

fmt.Println(out, err)

Output:

{true true} <nil>

This example shows a same step that is run many times concurrently.

The example uses dummy data to better showcase the immutability of the graph and step since we can produce different values and later reduce them without needing to take into account goroutines, mutexes, waitgroups to synchronize data at the end

Note: we use several UnitStep to showcase as it allows us to easily run dummy code, but it could use any type of step you want as long as it implements pipeline.Step[I, O]

step := pipeline.NewUnitStep(
	"half_increase",
	func(ctx context.Context, i int) (float32, error) {
		return float32(i) * 1.5, nil
	},
)
reduce := func(ctx context.Context, a, b float32) (float32, error) {
	return a + b, nil
}
ctx := context.Background()
in := 1

pipe := pipeline.NewConcurrentStep(
	[]pipeline.Step[int, float32]{
		step, step, step, step, step, step,
		step, step, step, step, step, step,
	},
	reduce,
)

out, err := pipe.Run(ctx, in)

fmt.Println(out, err)

Output:

18 <nil>

The following example evaluates a dummy condition and depending on the result it branches to one step or another.

This example uses dummy data to showcase as simple as possible this scenario.

Note: we use several UnitStep to showcase as it allows us to easily run dummy code, but it could use any type of step you want as long as it implements pipeline.Step[I, O]

type User any
type Data any
stmt := pipeline.NewStatement(
	"check_something",
	func(ctx context.Context, in User) bool {
		// check and return were to branch
		return true
	},
)
tf := pipeline.NewUnitStep(
	"true_case",
	func(ctx context.Context, in User) (Data, error) {
		// do something with input
		return Data(true), nil
	},
)
ff := pipeline.NewUnitStep(
	"false_step",
	func(ctx context.Context, u User) (Data, error) {
		// do something with input
		return Data(false), nil
	},
)
ctx := context.Background()
in := User(nil)

pipe := pipeline.NewConditionalStep[User, Data](stmt, tf, ff)

out, err := pipe.Run(ctx, in)

fmt.Println(out, err)

Output:

true <nil>

The following example shows an optional step were it will be run if the evaluated statement yields true, otherwise it will return the same input it was provided this type of constructed step doesn't allow output mutation as we don't know how to default the mutated output if the step is skipped

This example uses dummy data to showcase as simple as possible this scenario.

Note: we use several UnitStep to showcase as it allows us to easily run dummy code, but it could use any type of step you want as long as it implements pipeline.Step[I, O]

type User any
stmt := pipeline.NewStatement(
	"check_something",
	func(ctx context.Context, in User) bool {
		// check and return if we should run optional or not
		return in == User(1)
	},
)
of := pipeline.NewUnitStep(
	"optional_case",
	func(ctx context.Context, in User) (User, error) {
		// do something with input
		return User(true), nil
	},
)
ctx := context.Background()

pipe := pipeline.NewOptionalStep[User](stmt, of)

// Skips optional step (returns same input)
out, err := pipe.Run(ctx, User(12))
fmt.Println(out, err)

// Runs optional step (returns step output)
out, err = pipe.Run(ctx, User(1))
fmt.Println(out, err)

Output:

12 <nil>
true <nil>

The following example allows us to define an optional step with different output from its input, and in case the optional step is skipped it will call a provided default function for it to return the default output you want for that case

This example uses dummy data to showcase as simple as possible this scenario.

Note: we use several UnitStep to showcase as it allows us to easily run dummy code, but it could use any type of step you want as long as it implements pipeline.Step[I, O]

type User any
type Data any
stmt := pipeline.NewStatement(
	"check_something",
	func(ctx context.Context, in User) bool {
		// check and return if we should run optional or not
		return in == User(1)
	},
)
of := pipeline.NewUnitStep(
	"optional_case",
	func(ctx context.Context, in User) (Data, error) {
		// do something with input
		return Data(true), nil
	},
)
def := func(ctx context.Context, in User) (Data, error) {
	// create default value of type Data (gets run if we
	// skip the step as we don't know how to default Data type)
	return Data(false), nil
}
ctx := context.Background()

pipe := pipeline.NewOptionalStepWithDefault[User, Data](stmt, of, def)

// Skips optional step (returns default)
out, err := pipe.Run(ctx, User(12))
fmt.Println(out, err)

// Runs optional step (returns step output)
out, err = pipe.Run(ctx, User(1))
fmt.Println(out, err)

Output:

false <nil>
true <nil>

The following example shows a sequence between two steps were each of them is run sequentially with the output of the previous one as input

This example uses dummy data to showcase as simple as possible this scenario.

Note: we use several UnitStep to showcase as it allows us to easily run dummy code, but it could use any type of step you want as long as it implements pipeline.Step[I, O]

type DriverID int
type Driver any
type NotificationReceipt any
gd := pipeline.NewUnitStep(
	"get_driver",
	func(ctx context.Context, id DriverID) (Driver, error) {
		// do something with input
		return Driver(id), nil
	},
)
sn := pipeline.NewUnitStep(
	"send_notification",
	func(ctx context.Context, d Driver) (NotificationReceipt, error) {
		// do something with input
		return NotificationReceipt(25), nil
	},
)

pipe := pipeline.NewSequentialStep[DriverID, Driver, NotificationReceipt](gd, sn)

out, err := pipe.Run(context.Background(), DriverID(1234))

fmt.Println(out, err)

Output:

25 <nil>

This examples shows a simple statement that lets us evaluate it with a given input to yield a boolean result

This example uses dummy data to showcase as simple as possible this scenario.

stmt := pipeline.NewStatement(
	"is_number_odd",
	func(ctx context.Context, i int) bool {
		return i%2 != 0
	},
)

out := stmt.Evaluate(context.Background(), 25)

fmt.Println(out)

Output:

true

The following example shows a simple unit step that runs a unit of work with a given input and yields a result of a different type or an error depending on the execution

This example uses dummy data to showcase as simple as possible this scenario.

Note: we use several UnitStep to showcase as it allows us to easily run dummy code, but it could use any type of step you want as long as it implements pipeline.Step[I, O]

type (
	InData  any
	OutData any
)
step := pipeline.NewUnitStep(
	"do_something",
	func(ctx context.Context, in InData) (OutData, error) {
		// do something with input
		return OutData(in), nil
	},
)

out, err := step.Run(context.Background(), InData("example"))

fmt.Println(out, err)

Output:

example <nil>