component

README

go-component

This package provides simple yet powerful abstractions for creating synchronous and asynchronous components that can be plugged and played to realize a truly loosely-coupled architecture. You can read the code & run main.go in the sample package for a demonstration.

The main properties of the proposed model are:

• Structural - applications are described as components, their inputs and their outputs encapsulated as futures.
• Asynchronous/synchronous - you can decide the type of a component by implementing the corresponding interface. The order of execution for synchronous components can be determined when building an execution flow.
• Isolated - each component describes what it needs for its logic via an Input interface, which can then be provided by 1 or more components via their outputs wrapped as futures. State is not shared across components.
• Concurrent - execution is greedy, all asynchronous logic will get started in a goroutine immediately when an execution flow gets triggered. Synchronous logic will still get executed in the desired sequence. Components will automatically block & wait for each other when they access methods of futures that have not resolved yet.
• Fail fast - each component must handle its own errors (e.g. by using some default values as output or by logging the error & then ignoring it to return early in case some logic can be bypassed). If an error is returned by any components, the entire execution flow will stop immediately and this error will be used as the final result of this execution.

Documentation

Index

• Variables
• type AsyncComponent
• type ExecutionFlow
  • func (f ExecutionFlow) Cancel(firstLayerIdx int, err error)
• type ExecutionFlowBuilder
  • func NewExecutionFlowBuilder() *ExecutionFlowBuilder
  • func (b *ExecutionFlowBuilder) Append(executors ...IExecutor) *ExecutionFlowBuilder
  • func (b *ExecutionFlowBuilder) Get() ExecutionFlow
  • func (b *ExecutionFlowBuilder) NextLayer() *ExecutionFlowBuilder
• type Executor
  • func CreateAsyncExecutor[T any](c AsyncComponent[T]) Executor[T]
  • func CreateSyncExecutor[T any](c SyncComponent[T]) Executor[T]
  • func CreateSyncOrchestratingExecutor(doFn func(ctx context.Context) error) Executor[any]
  • func CreateSyncOrchestratingExecutorWithResult[T any](doFn func(ctx context.Context) (T, error)) (Executor[T], async.Task[T])
  • func (e Executor[T]) GetExecutingTask() async.Task[T]
  • func (e Executor[T]) InvokeExecutingTask(ctx context.Context) error
• type ExecutorWithLoading
  • func CreateSyncExecutorWithLoading[V any, T any](c SyncComponentWithLoading[V, T]) ExecutorWithLoading[V, T]
  • func (e ExecutorWithLoading[V, T]) GetExecutingTask() async.Task[T]
  • func (e ExecutorWithLoading[V, T]) InvokeExecutingTask(ctx context.Context) error
• type IExecutor
• type LoadData
• type MockAsyncComponent
  • func NewMockAsyncComponent[T interface{}](t interface{ ... }) *MockAsyncComponent[T]
  • func (_m *MockAsyncComponent[T]) Execute(ctx context.Context) (T, error)
• type MockIExecutor
  • func NewMockIExecutor(t interface{ ... }) *MockIExecutor
  • func (_m *MockIExecutor) InvokeExecutingTask(ctx context.Context) error
• type MockSyncComponent
  • func NewMockSyncComponent[T interface{}](t interface{ ... }) *MockSyncComponent[T]
  • func (_m *MockSyncComponent[T]) ExecuteSync(ctx context.Context) (T, error)
• type MockSyncComponentWithLoading
  • func NewMockSyncComponentWithLoading[V interface{}, T interface{}](t interface{ ... }) *MockSyncComponentWithLoading[V, T]
  • func (_m *MockSyncComponentWithLoading[V, T]) ExecuteSync(ctx context.Context, data LoadData[V]) (T, error)
  • func (_m *MockSyncComponentWithLoading[V, T]) Load(ctx context.Context) (V, error)
• type SyncComponent
• type SyncComponentWithLoading

Variables

var ForkJoinFailingFast = func(ctx context.Context, flow ExecutionFlow) error {
	if len(flow.Executors) == 0 {
		return nil
	}

	if len(flow.Executors) == 1 {
		return doForkJoinFailingFast(ctx, flow, 0)
	}

	errChan := make(chan error, len(flow.Executors))

	for i := 0; i < len(flow.Executors); i++ {
		go func(idx int) {
			errChan <- doForkJoinFailingFast(ctx, flow, idx)
		}(i)
	}

	done := 0

	for {
		select {
		case <-ctx.Done():
			return ctx.Err()
		case err := <-errChan:

			if err != nil {
				return err
			}

			done = done + 1
			if done == len(flow.Executors) {
				return nil
			}
		}
	}
}

ForkJoinFailingFast invokes the executors in the given ExecutionFlow based on its type. If an executor comes from an async component, it will be executed asynchronously. If an executor comes from a sync component, its loading task will be executed asynchronously while its executing task will be executed synchronously based on the order of the given tasks.

If any of the executing tasks of async or sync components returns an error, the function will stop immediately and return this error to the caller.

Types

type AsyncComponent

type AsyncComponent[T any] interface {
	Execute(ctx context.Context) (T, error)
}

AsyncComponent represents those components that can be executed concurrently together with other AsyncComponent. They can also get executed sequentially with other SyncComponent if engineers choose to do so.

type ExecutionFlow

type ExecutionFlow struct {
	Executors [][]IExecutor
}

ExecutionFlow ...

func (ExecutionFlow) Cancel

func (f ExecutionFlow) Cancel(firstLayerIdx int, err error)

Cancel cancels all executors from the given layer down.

type ExecutionFlowBuilder

type ExecutionFlowBuilder struct {
	// contains filtered or unexported fields
}

ExecutionFlowBuilder ...

func NewExecutionFlowBuilder

func NewExecutionFlowBuilder() *ExecutionFlowBuilder

NewExecutionFlowBuilder ...

func (*ExecutionFlowBuilder) Append

func (b *ExecutionFlowBuilder) Append(executors ...IExecutor) *ExecutionFlowBuilder

Append appends the given executors to the end of the current executor layer.

func (*ExecutionFlowBuilder) Get

func (b *ExecutionFlowBuilder) Get() ExecutionFlow

Get returns the current flow.

func (*ExecutionFlowBuilder) NextLayer

func (b *ExecutionFlowBuilder) NextLayer() *ExecutionFlowBuilder

NextLayer moves the builder to the next layer of executors, effectively finalize the current layer.

type Executor

type Executor[T any] struct {
	// contains filtered or unexported fields
}

Executor encapsulates the tasks that need to be executed to carry out the business logic of a component without loading logic.

func CreateAsyncExecutor

func CreateAsyncExecutor[T any](c AsyncComponent[T]) Executor[T]

CreateAsyncExecutor returns an Executor encapsulating the executing task that would be handled by the given AsyncComponent.

func CreateSyncExecutor

func CreateSyncExecutor[T any](c SyncComponent[T]) Executor[T]

CreateSyncExecutor returns an Executor encapsulating the executing task that would be handled by the given SyncComponent.

func CreateSyncOrchestratingExecutor

func CreateSyncOrchestratingExecutor(doFn func(ctx context.Context) error) Executor[any]

CreateSyncOrchestratingExecutor returns a component that is meant for orchestrating some logic without returning any values beside throwing an error if necessary.

func CreateSyncOrchestratingExecutorWithResult

func CreateSyncOrchestratingExecutorWithResult[T any](doFn func(ctx context.Context) (T, error)) (Executor[T], async.Task[T])

CreateSyncOrchestratingExecutorWithResult returns a component that is meant for orchestrating some logic that returns some values and throws an error if necessary.

func (Executor[T]) GetExecutingTask

func (e Executor[T]) GetExecutingTask() async.Task[T]

func (Executor[T]) InvokeExecutingTask

func (e Executor[T]) InvokeExecutingTask(ctx context.Context) error

type ExecutorWithLoading

type ExecutorWithLoading[V any, T any] struct {
	// contains filtered or unexported fields
}

ExecutorWithLoading encapsulates the tasks that need to be executed to carry out the business logic of a synchronous component with loading logic.

func CreateSyncExecutorWithLoading

func CreateSyncExecutorWithLoading[V any, T any](c SyncComponentWithLoading[V, T]) ExecutorWithLoading[V, T]

CreateSyncExecutorWithLoading returns an ExecutorWithLoading encapsulating the loading & executing tasks that would be handled by the given component.

func (ExecutorWithLoading[V, T]) GetExecutingTask

func (e ExecutorWithLoading[V, T]) GetExecutingTask() async.Task[T]

func (ExecutorWithLoading[V, T]) InvokeExecutingTask

func (e ExecutorWithLoading[V, T]) InvokeExecutingTask(ctx context.Context) error

type IExecutor

type IExecutor interface {
	InvokeExecutingTask(ctx context.Context) error
	// contains filtered or unexported methods
}

type LoadData

type LoadData[V any] struct {
	Data V
	Err  error
}

LoadData contains the data and/or the error that was returned by the loading task. SyncComponentWithLoading is responsible for handling this error in its executing logic.

type MockAsyncComponent

type MockAsyncComponent[T interface{}] struct {
	mock.Mock
}

MockAsyncComponent is an autogenerated mock type for the AsyncComponent type

func NewMockAsyncComponent

func NewMockAsyncComponent[T interface{}](t interface {
	mock.TestingT
	Cleanup(func())
}) *MockAsyncComponent[T]

NewMockAsyncComponent creates a new instance of MockAsyncComponent. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations. The first argument is typically a *testing.T value.

func (*MockAsyncComponent[T]) Execute

func (_m *MockAsyncComponent[T]) Execute(ctx context.Context) (T, error)

Execute provides a mock function with given fields: ctx

type MockIExecutor

type MockIExecutor struct {
	mock.Mock
}

MockIExecutor is an autogenerated mock type for the IExecutor type

func NewMockIExecutor

func NewMockIExecutor(t interface {
	mock.TestingT
	Cleanup(func())
}) *MockIExecutor

NewMockIExecutor creates a new instance of MockIExecutor. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations. The first argument is typically a *testing.T value.

func (*MockIExecutor) InvokeExecutingTask

func (_m *MockIExecutor) InvokeExecutingTask(ctx context.Context) error

InvokeExecutingTask provides a mock function with given fields: ctx

type MockSyncComponent

type MockSyncComponent[T interface{}] struct {
	mock.Mock
}

MockSyncComponent is an autogenerated mock type for the SyncComponent type

func NewMockSyncComponent

func NewMockSyncComponent[T interface{}](t interface {
	mock.TestingT
	Cleanup(func())
}) *MockSyncComponent[T]

NewMockSyncComponent creates a new instance of MockSyncComponent. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations. The first argument is typically a *testing.T value.

func (*MockSyncComponent[T]) ExecuteSync

func (_m *MockSyncComponent[T]) ExecuteSync(ctx context.Context) (T, error)

ExecuteSync provides a mock function with given fields: ctx

type MockSyncComponentWithLoading

type MockSyncComponentWithLoading[V interface{}, T interface{}] struct {
	mock.Mock
}

MockSyncComponentWithLoading is an autogenerated mock type for the SyncComponentWithLoading type

func NewMockSyncComponentWithLoading

func NewMockSyncComponentWithLoading[V interface{}, T interface{}](t interface {
	mock.TestingT
	Cleanup(func())
}) *MockSyncComponentWithLoading[V, T]

NewMockSyncComponentWithLoading creates a new instance of MockSyncComponentWithLoading. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations. The first argument is typically a *testing.T value.

func (*MockSyncComponentWithLoading[V, T]) ExecuteSync

func (_m *MockSyncComponentWithLoading[V, T]) ExecuteSync(ctx context.Context, data LoadData[V]) (T, error)

ExecuteSync provides a mock function with given fields: ctx, data

func (*MockSyncComponentWithLoading[V, T]) Load

func (_m *MockSyncComponentWithLoading[V, T]) Load(ctx context.Context) (V, error)

Load provides a mock function with given fields: ctx

type SyncComponent

type SyncComponent[T any] interface {
	ExecuteSync(ctx context.Context) (T, error)
}

SyncComponent represents those components that must be executed sequentially together with other SyncComponent. These components must not be executed concurrently with other AsyncComponent.

Engineers should use SyncComponentWithLoading instead if they need to perform some loading before executing the main logic.

type SyncComponentWithLoading

type SyncComponentWithLoading[V any, T any] interface {
	Load(ctx context.Context) (V, error)
	ExecuteSync(ctx context.Context, data LoadData[V]) (T, error)
}

SyncComponentWithLoading represents those components that must be executed sequentially together with other SyncComponent. They must not be executed concurrently with other AsyncComponent.

Engineers should use SyncComponent instead if they don't need to perform any loading before executing the main logic.