resource

Documentation

Overview

Package resource provides supervised, serialized access to acquired resources.

Getting started

An Actor acquires a resource for each execution attempt, receives operations directly, runs them one at a time, and releases the resource when the attempt ends. Acquisition failures and asynchronous Actor.Cast failures are returned to the supervisor. Actor.Call operation errors are also returned to their callers before the resource actor stops.

The resource can be anything that should be owned and accessed serially: a serial port, a Modbus client, a network connection, or a device handle. The acquire function should create a fresh value and the release function should close or disconnect it:

device := resource.NewActor(
	"serial",
	func(ctx context.Context) (*Device, error) {
		return openDevice(ctx)
	},
	func(ctx context.Context, device *Device) error {
		return device.Close(ctx)
	},
)

supervisor := sup.NewSupervisor("root", sup.Transient).AddActors(device)

Use Actor.SetReleaseTimeout before the first Actor.Run call to bound the release context when the default timeout is not appropriate.

Use Actor.Call when the caller needs a typed result:

value, err := device.Call(ctx, func(ctx context.Context, device Device) (int, error) {
	return device.Read(ctx)
})

Use Actor.Cast when the caller only needs direct handoff. A Cast operation error stops the resource actor so its supervisor can apply the restart policy. Both operations wait across acquisition attempts and supervisor restart delays until the actor receives their request, and stop waiting when their context is canceled. Requests that have already been received are never replayed on a new resource execution.

A resource is released with a fresh timeout context when Run ends. That context preserves values from the actor context but is not canceled by the actor's shutdown cancellation.

Index

• Variables
• type Actor
  • func NewActor[T any](id string, acquire func(context.Context) (T, error), ...) *Actor[T]
  • func (a *Actor[T]) Call[R any](ctx context.Context, operation func(context.Context, T) (R, error)) (R, error)
  • func (a *Actor[T]) Cast(ctx context.Context, operation func(context.Context, T) error) error
  • func (a *Actor[T]) ID() string
  • func (a *Actor[T]) Run(ctx context.Context) (runErr error)
  • func (a *Actor[T]) SetReleaseTimeout(timeout time.Duration) *Actor[T]

Examples

• Actor.Call

Variables

var ErrExecutionStopped = errors.New("resource: execution stopped before call completed")

ErrExecutionStopped is returned when an accepted Actor.Call cannot complete because its resource execution stopped before producing a reply. The operation is not retried because it may already have produced side effects.

Types

type Actor

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

Actor acquires a resource for each execution attempt and serializes Actor.Call and Actor.Cast operations against it until its context is canceled or an operation fails. A supervisor can restart it to acquire a fresh resource.

func NewActor

func NewActor[T any](
	id string,
	acquire func(context.Context) (T, error),
	release func(context.Context, T) error,
) *Actor[T]

NewActor creates a resource actor. The acquire function is called once per execution attempt. It should honor its context and return a fresh resource because the previous resource is released when its actor stops. The release function is called exactly once after a successful acquisition, including when the resource actor fails. Its context preserves the actor context's values, ignores its cancellation, and has a fresh release timeout deadline. Operations made before the first Run and between execution attempts wait for the actor to become ready. A request that has not been accepted carries over to the next execution attempt. The release timeout defaults to five seconds.

Optional configuration is added with methods before the first Actor.Run call.

func (*Actor[T]) Call

func (a *Actor[T]) Call[R any](
	ctx context.Context,
	operation func(context.Context, T) (R, error),
) (R, error)

Call sends a request, waits for its typed result, and returns it. The operation is serialized with all other operations on the actor's resource. It waits across execution attempts until the actor receives the request. The caller's context controls both handoff and execution. Operation errors are returned to the caller and stop the resource actor so supervision can decide whether to restart it.

Example

package main

import (
	"bytes"
	"context"
	"fmt"

	"github.com/webermarci/sup"
	"github.com/webermarci/sup/resource"
)

func main() {
	ctx, cancel := context.WithCancel(context.Background())
	buffer := resource.NewActor(
		"buffer",
		func(context.Context) (*bytes.Buffer, error) {
			return new(bytes.Buffer), nil
		},
		func(context.Context, *bytes.Buffer) error {
			return nil
		},
	)

	supervisor := sup.NewSupervisor("root", sup.Temporary).
		AddActors(buffer)
	done := make(chan error, 1)
	go func() { done <- supervisor.Run(ctx) }()

	length, err := buffer.Call(ctx,
		func(_ context.Context, buffer *bytes.Buffer) (int, error) {
			return buffer.WriteString("hello")
		})
	if err != nil {
		panic(err)
	}
	fmt.Println("length:", length)

	if err := buffer.Cast(ctx,
		func(_ context.Context, buffer *bytes.Buffer) error {
			return buffer.WriteByte('!')
		}); err != nil {
		panic(err)
	}

	value, err := buffer.Call(ctx,
		func(_ context.Context, buffer *bytes.Buffer) (string, error) {
			return buffer.String(), nil
		})
	if err != nil {
		panic(err)
	}
	fmt.Println("value:", value)

	cancel()
	if err := <-done; err != nil {
		panic(err)
	}

}

Output:
length: 5
value: hello!

func (*Actor[T]) Cast

func (a *Actor[T]) Cast(
	ctx context.Context,
	operation func(context.Context, T) error,
) error

Cast sends a fire-and-forget operation, waits across execution attempts, and returns once the actor receives it. The operation is serialized with all other operations on the actor's resource and may run after Cast returns. The context passed to operation is the actor's execution context; ctx controls handoff. If the operation returns an error, the resource actor stops and supervision decides whether to restart it.

func (*Actor[T]) ID

func (a *Actor[T]) ID() string

ID returns the actor id.

func (*Actor[T]) Run

func (a *Actor[T]) Run(ctx context.Context) (runErr error)

Run acquires one resource and processes serialized operations until the actor context is canceled or an operation returns an error. Cancellation is a clean shutdown and returns nil. Actor.Call errors are returned to their callers, and both Actor.Call and Actor.Cast errors are returned from Run for supervision.

func (*Actor[T]) SetReleaseTimeout

func (a *Actor[T]) SetReleaseTimeout(timeout time.Duration) *Actor[T]

SetReleaseTimeout sets the maximum duration of the context passed to the release function and returns the actor for chaining.