GoAkt

https://tochemey.gitbook.io/goakt
Admin
GoAkt is an actor framework inspired by Erlang and Akka, designed to build reactive and distributed...
Tokens:24,638
Snippets:185
Trust Score:10
Update:9 months ago
Show doc for...
Tokens:
Raw
### Complete Goakt Actor System Example

Source: https://tochemey.gitbook.io/goakt/get-started

Demonstrates a complete Go application using Goakt, including actor system initialization, spawning an actor, sending messages, and handling graceful shutdown. It utilizes Protocol Buffers for message definitions and showcases actor communication patterns.

```go
package main

import (
	"context"
	"os"
	"os/signal"
	"syscall"
	"time"

	goakt "github.com/tochemey/goakt/v3/actor"
	"github.com/tochemey/goakt/v3/log"

	hellopb "github.com/tochemey/goakt-examples/v2/internal/helloworldpb"
)

func main() {
	ctx := context.Background()

	// use the address default log. real-life implement the log interface`
	logger := log.DefaultLogger

	// create the actor system. kindly in real-life application handle the error
	actorSystem, _ := goakt.NewActorSystem(
		"HelloWorld",
		goakt.WithLogger(logger),
		goakt.WithPassivationDisabled(),
		goakt.WithActorInitMaxRetries(3))

	// start the actor system
	if err := actorSystem.Start(ctx); err != nil {
		logger.Fatal(err)
		os.Exit(1)
	}

	// create a Hello actor
	pid, err := actorSystem.Spawn(ctx, "Hello", NewHelloWorld())
	if err != nil {
		logger.Fatal(err)
		os.Exit(1)
	}

	// send an SayHello message to the actor and expect a response
	response, err := goakt.Ask(ctx, pid, new(hellopb.SayHello), time.Second)
	if err != nil {
		logger.Fatal(err)
		os.Exit(1)
	}

	sswitch response.(type) {
	case *hellopb.SayHi:
		logger.Info("received SayHi from actor")
	default:
		logger.Fatal("unexpected response from actor")
	}

	// capture ctrl+c
	interruptSignal := make(chan os.Signal, 1)
	signal.Notify(interruptSignal, os.Interrupt, syscall.SIGINT, syscall.SIGTERM)
	<-interruptSignal

	// stop the actor system
	_ = actorSystem.Stop(ctx)
	os.Exit(0)
}

type HelloWorld struct{}

var _ goakt.Actor = (*HelloWorld)(nil)

// NewHelloWorld creates an instance of HelloWorld Actor
func NewHelloWorld() *HelloWorld {
	return &HelloWorld{}
}

func (x *HelloWorld) PreStart(context.Context) error { return nil }

func (x *HelloWorld) Receive(ctx *goakt.ReceiveContext) {
	sswitch ctx.Message().(type) {
	case *hellopb.SayHello:
		ctx.Response(new(hellopb.SayHi))
	default:
		ctx.Unhandled()
	}
}

func (x *HelloWorld) PostStop(context.Context) error { return nil }

```

--------------------------------

### Implement Actor Interface in Go

Source: https://tochemey.gitbook.io/goakt/get-started

Shows how to implement the `Actor` interface in Go, defining the behavior of an actor. This includes methods for initialization, message handling, and cleanup.

```go
type HelloWorld struct{}

var _ actors.Actor = (*HelloWorld)(nil)

// NewHelloWorld creates an instance
func NewHelloWorld() *HelloWorld {
	return &HelloWorld{}
}

func (x *HelloWorld) PreStart(context.Context) error { return nil }

func (x *HelloWorld) Receive(ctx *actors.ReceiveContext) {
	sswitch ctx.Message().(type) {
	case *goaktpb.PostStart:
	case *hellopb.SayHello:
		// here the receiving actor is responding to the message
		ctx.Response(new(hellopb.SayHi))
	default:
		ctx.Unhandled()
	}
}

func (x *HelloWorld) PostStop(context.Context) error { return nil }

```

--------------------------------

### Instantiate and Start GoAkt Actor System in Go

Source: https://tochemey.gitbook.io/goakt/features/actor-system

This Go code snippet demonstrates how to instantiate and start a GoAkt actor system. It includes configuration options like disabling passivation, enabling remoting, and setting a custom logger. The system is started and listens for interrupt signals to shut down gracefully.

```go
package main

import (
	"context"
	"os"
	"os/signal"
	"sync"
	"syscall"
	"time"

	"github.com/tochemey/goakt/v3/actor"
	"github.com/tochemey/goakt/v3/log"
)

func main() {
	ctx := context.Background()
	host := "0.0.0.0"
	port := 4000

	logger := log.New(log.InfoLevel, os.Stdout)

	// create an actor system. Check the reference
	// doc for additional options
	actorSystem, err := actor.NewActorSystem(
		"ChatSystem",
		actor.WithPassivationDisabled(),
		actor.WithRemoting(host, int32(port)),
		actor.WithLogger(logger))

	if err != nil {
		logger.Fatal(err)
		os.Exit(1)
	}

	// start the actor system
	if err := actorSystem.Start(ctx); err != nil {
		logger.Fatal(err)
		os.Exit(1)
	}

	// capture ctrl+c
	interruptSignal := make(chan os.Signal, 1)
	signal.Notify(interruptSignal, os.Interrupt, syscall.SIGINT, syscall.SIGTERM)
	<-

	// stop the actor system
	_ = actorSystem.Stop(ctx)
	os.Exit(0)
}

```

--------------------------------

### Define Messages with Protocol Buffers

Source: https://tochemey.gitbook.io/goakt/get-started

Defines the structure for messages that actors will handle using Google Protocol Buffers syntax. This is the first step in setting up actor communication.

```protobuf
syntax = "proto3";

package hellopb;

message SayHello {}

message SayHi {}

```

--------------------------------

### Configure Supervisor Strategy in GoAkt

Source: https://tochemey.gitbook.io/goakt/features/actor/supervision

This example demonstrates how to create a supervisor with a specific strategy and directives for various error types. It utilizes NewSupervisor, WithStrategy, and WithDirective to define the failure handling behavior.

```go
supervisor := NewSupervisor(
	WithStrategy(OneForOneStrategy),
	WithDirective(PanicError{}, RestartDirective),
	WithDirective(&runtime.PanicNilError{}, RestartDirective),
	WithDirective(InternalError{}, ResumeDirective),
	WithDirective(SpawnError{}, ResumeDirective),
)
```

--------------------------------

### GoAkt Testkit Probe: Asserting Message Reception

Source: https://tochemey.gitbook.io/goakt/features/testkit

Provides Go code examples for various assertion methods available on the GoAkt testkit probe. These methods allow for precise testing of message reception, including exact matches, timed assertions, and type checking, ensuring actors behave as expected.

```go
// Exactly matches the given protobuf message.
// It fails the test if no message is received or the message does not match.
probe.ExpectMessage(message proto.Message)

// ExpectMessageWithin asserts that the expected message is received within the given duration.
// It fails the test if the timeout is reached or the message differs.
probe.ExpectMessageWithin(duration time.Duration, message proto.Message)

// ExpectNoMessage asserts that no message is received within a short, default time window.
// This is useful for asserting inactivity or idle actors.
probe.ExpectNoMessage()

// ExpectAnyMessage waits for and returns the next message received by the probe.
// It fails the test if no message is received in a reasonable default timeout.
receivedMessage := probe.ExpectAnyMessage()

// ExpectAnyMessageWithin waits for and returns the next message received within the specified duration.
// It fails the test if no message is received in the given time window.
receivedMessageWithin := probe.ExpectAnyMessageWithin(duration time.Duration)

// ExpectMessageOfType asserts that the next received message matches the given message type.
// It fails the test if no message is received or the type does not match.
probe.ExpectMessageOfType(messageType protoreflect.MessageType)

// ExpectMessageOfTypeWithin asserts that a message of the given type is received within the specified duration.
// It fails the test if the type does not match or if the timeout is reached.
probe.ExpectMessageOfTypeWithin(duration time.Duration, messageType protoreflect.MessageType)

// ExpectTerminated asserts that the actor with the specified name has terminated.
// This is useful when verifying actor shutdown behavior.
probe.ExpectTerminated(actorName string)
```

--------------------------------

### Implement Actor Interface in Go

Source: https://tochemey.gitbook.io/goakt/features/actor

Defines the core Actor interface in Go, outlining methods for actor lifecycle management and message handling. These methods include PreStart for initialization, Receive for message processing, and PostStop for cleanup. Errors in PreStart prevent actor startup, while PostStop errors are logged but do not halt shutdown.

```go
type Actor interface {
    // PreStart is called once before the actor starts processing messages.
    //
    // Use this method to initialize dependencies such as database clients,
    // caches, or external service connections and persistent state recovery. If PreStart returns an error,
    // the actor will not be started, and the failure will be handled by its supervisor.
    PreStart(ctx *Context) error
    
    // Receive handles all messages sent to the actor's mailbox.
    //
    // This is the heart of the actor's behavior. Messages can include user-defined
    // commands/events as well as internal system messages such as PostStart or lifecycle signals.
    //
    // Actors can reply to messages using async messaging patterns or configure replies inline
    // where supported. Avoid heavy synchronous workflows as they may degrade throughput in high-load scenarios.
    //
    // Tip: Use pattern matching or typed message handlers to organize complex message workflows.
    Receive(ctx *ReceiveContext)
    
    // PostStop is called when the actor is about to shut down.
    //
    // This lifecycle hook is invoked after the actor has finished processing all messages
    // in its mailbox and is guaranteed to run before the actor is fully terminated.
    //
    // Use this method to perform final cleanup actions such as:
    //   - Releasing resources (e.g., database connections, goroutines, open files)
    //   - Flushing logs or metrics
    //   - Notifying other systems of termination (e.g., via events or pub/sub)
    //
    // This method is especially important passivated actors, as it is also
    // called during passivation when an idle actor is stopped to free up resources.
    //
    // Note: If PostStop returns an error, the error is logged but does not prevent the actor
    // from being stopped. Keep PostStop logic fast and resilient to avoid delaying system shutdowns.
    PostStop(ctx *Context) error
}
```

--------------------------------

### GoAkt Testkit: Creating and Using a Test Probe

Source: https://tochemey.gitbook.io/goakt/features/testkit

Demonstrates the initialization of the GoAkt testkit and a test probe, spawning an actor, and sending messages for testing. The testkit requires a go context and a testing.T instance, while the probe facilitates message interaction with actors. Options can be applied to the probe for customization.

```go
import (
	"context"
	"testing"

	"github.com/Tochemey/goakt/testkit"
	"your_project/internal/actors"
	"your_project/internal/messages/testpb"
)

func TestActorBehavior(t *testing.T) {
	ctx := context.Background()

	// 1. Create an instance of the testkit
	testkit := testkit.New(ctx, t)

	// 2. Create the instance of the actor under test
	ttestkit.Spawn(ctx, "pinger", &actors.PingerActor{})

	// 3. Create an instance of test probe
	probe := testkit.NewProbe(ctx)

	// 4. Use the probe to send a message to the actor under test
	probe.Send("pinger", new(testpb.Ping))

	// 5. Assert expected behavior (example using ExpectMessage)
	probe.ExpectMessage(&testpb.Pong{})

	// 6. Shut down the testkit and the probe
	probe.Stop()
	testkit.Shutdown(ctx)
}
```

--------------------------------

### Statefulness with Extensions

Source: https://tochemey.gitbook.io/goakt/features/grain

Explains how to make Grains stateful by leveraging the **Extension** capability. Extensions can handle persistence, snapshotting, and durable storage, keeping the core Grain logic clean.

```APIDOC
## Stateful Grains using Extensions

### Description
Enables Grains to maintain state across operations by integrating with Goakt's Extension system. Extensions can manage persistence, snapshotting, and durable storage for stateful Grains.

### Method
Extension System Integration

### Endpoint
N/A (Conceptual feature, implementation details vary)

### Parameters
- **Extension Implementation**: Developers need to implement specific extension interfaces to manage persistence logic (e.g., saving/loading state).
- **`WithExtension(ext Extension)`**: Grain option to attach an extension during Grain identity creation.

### Request Example
```go
// Assuming an 'PersistenceExtension' is implemented
var persistenceExt = &PersistenceExtension{ /* ... */ }

identity, err := actorSystem.GrainIdentity(ctx, accountID, func(ctx context.Context) (goakt.Grain, error) {
	return grain, nil
}, 
WithExtension(persistenceExt)) 

// Inside the Grain, methods provided by the extension would be used to save/load state.
```

### Response
#### Success Response
- N/A (State management is handled internally by extensions.)

#### Error Response
- Errors related to persistence operations (e.g., storage failure) would be handled by the specific extension implementation.
```

--------------------------------

### Kubernetes RBAC Configuration for Pod Discovery

Source: https://tochemey.gitbook.io/goakt/features/cluster/service-discovery

This YAML configuration defines the necessary Role-Based Access Control (RBAC) permissions for GoAkt's Kubernetes discovery provider. It creates a 'pod-reader' Role that grants 'get', 'watch', and 'list' permissions on 'pods' resources within the core API group. A 'RoleBinding' named 'read-pods' then binds this Role to the 'default' Service Account, allowing pods to discover each other.

```yaml
kind: Role
apiVersion: rbac.authorization.k8s.io/v1
metadata:
  name: pod-reader
rules:
  - apiGroups: [""] # "" indicates the core API group
    resources: ["pods"]
    verbs: ["get", "watch", "list"]
---
kind: RoleBinding
apiVersion: rbac.authorization.k8s.io/v1
metadata:
  name: read-pods
subjects:
  # Uses the default service account. Consider creating a new one.
  - kind: ServiceAccount
    name: default
roleRef:
  kind: Role
  name: pod-reader
  apiGroup: rbac.authorization.k8s.io

```

--------------------------------

### Create Grain Identity in Go

Source: https://tochemey.gitbook.io/goakt/features/grain

Demonstrates how to create a unique identity for a GoAkt grain. It involves obtaining a grain identity using the ActorSystem, specifying an account ID, and providing a factory function for grain creation with optional activation strategies and roles. Dependencies include the 'goakt' and 'uuid' packages.

```go
grain := &Grain{}

accountID := uuid.NewString()
identity, err := actorSystem.GrainIdentity(ctx, accountID, func(ctx context.Context) (goakt.Grain, error) {
	return grain, nil
}, 
WithActivationStrategy(RoundRobinActivation), 
WithActivationRole("backend"))
```

--------------------------------

### Static Discovery Provider Configuration in Go

Source: https://tochemey.gitbook.io/goakt/features/cluster/service-discovery

Initializes the static discovery provider for goakt, which relies on a predefined list of host addresses. Each address should be in the format 'host:port', specifying the gossip protocol port.

```go
package main

import "github.com/tochemey/goakt/v3/discovery/static"

// define the discovery configuration
config := static.Config{
    Hosts: []string{
    "node1:3322",
    "node2:3322",
    "node3:3322",
    },
}
// instantiate the dnssd discovery provider
disco := static.NewDiscovery(&config)

// pass the service discovery when enabling cluster mode in the actor system

```

--------------------------------

### Configure Kubernetes Discovery in Goakt

Source: https://tochemey.gitbook.io/goakt/features/cluster/service-discovery

This Go code snippet demonstrates how to configure the Kubernetes discovery provider for GoAkt. It defines necessary constants for namespace, application name, and port names, sets up pod labels, and instantiates the `kubernetes.Config` struct. The `kubernetes.NewDiscovery` function is then used to create the discovery provider instance.

```go
package main

import "github.com/tochemey/goakt/v3/discovery/kubernetes"

const (
    namespace          = "default"
    applicationName    = "accounts"
    actorSystemName    = "AccountsSystem"
    discoveryPortName  = "discovery-port"
    peersPortName      = "peers-port"
    remotingPortName   = "remoting-port"
)

// define the pod labels to use
// for instance you can use the kubernetes standard labels
labels := map[string]string{
    "app.kubernetes.io/part-of":   "some-part-of",
    "app.kubernetes.io/component": actorSystemName,
    "app.kubernetes.io/name":      applicationName,
}

// define the discovery config
config := kubernetes.Config{
    Namespace:        namespace,
    DiscoveryPortName: discoveryPortName,
    RemotingPortName: remotingPortName,
    PeersPortName:  peersPortName,
    PodLabels:    labels
}

// instantiate the k8 discovery provider
disco := kubernetes.NewDiscovery(&config)

// pass the service discovery when enabling cluster mode in the actor system

```

--------------------------------

### Grain Identity and Activation

Source: https://tochemey.gitbook.io/goakt/features/grain

Demonstrates how to obtain a Grain identity, which is necessary before sending messages to a Grain. It shows the use of `GrainIdentity` with activation strategies and roles.

```APIDOC
## POST /api/grains

### Description
Creates or retrieves a unique identity for a Grain instance. This identity is required to interact with the Grain.

### Method
POST

### Endpoint
/api/grains

### Parameters
#### Path Parameters
- None

#### Query Parameters
- **accountID** (string) - Required - The unique identifier for the account.
- **activationStrategy** (string) - Optional - The strategy to use for activating the grain (e.g., RoundRobinActivation).
- **activationRole** (string) - Optional - The role to assign to the activated grain (e.g., "backend").

#### Request Body
- **grainFactory** (func(ctx context.Context) (goakt.Grain, error)) - Required - A factory function that creates a new Grain instance.

### Request Example
```go
grain := &MyGrain{ /* ... */ }
accountID := "some-uuid-string"
identity, err := actorSystem.GrainIdentity(ctx, accountID, func(ctx context.Context) (goakt.Grain, error) {
	return grain, nil
}, 
WithActivationStrategy(RoundRobinActivation), 
WithActivationRole("backend"))
```

### Response
#### Success Response (200)
- **identity** (*GrainIdentity) - The unique identifier of the Grain.
- **error** (error) - An error object if the operation failed.

#### Response Example
```json
{
  "identity": {
    "id": "some-uuid-string",
    "type": "MyGrain"
  },
  "error": null
}
```
```

--------------------------------

### DNS Discovery Provider Configuration in Go

Source: https://tochemey.gitbook.io/goakt/features/cluster/service-discovery

Sets up the DNS discovery provider for goakt, useful for local development with Docker. It uses a domain name to discover nodes and can optionally look up IPv6 addresses.

```go
package main

import "github.com/tochemey/goakt/v3/discovery/dnssd"

const domainName = "accounts"

// define the discovery options
config := dnssd.Config{
    dnssd.DomainName: domainName,
    dnssd.IPv6:       false,
}
// instantiate the dnssd discovery provider
disco := dnssd.NewDiscovery(&config)

// pass the service discovery when enabling cluster mode in the actor system

```

--------------------------------

### Subscribe to a Topic in Goakt

Source: https://tochemey.gitbook.io/goakt/features/pubsub

Demonstrates how to subscribe an actor to a specific topic using the `Tell` method with a `Subscribe` message. This requires the actor system to have PubSub enabled using `WithPubsub` option. The local `TopicActor` instance is accessed from the `ActorSystem`.

```go
import (
	"context"
	"github.com/Tochemey/goakt/actor"
	"github.com/Tochemey/goakt/goaktpb"
)

func main() {
	ctx := context.Background()
	// Assume node1 is an initialized ActorSystem with PubSub enabled
	// actor1, err := node1.Spawn(ctx, "actor1", Subscriber())

	topic := "test-topic"
	// Subscribe to the topic
	// err = actor1.Tell(ctx, node1.TopicActor(), &goaktpb.Subscribe{Topic: topic})
}
```

--------------------------------

### Mailbox Configuration

Source: https://tochemey.gitbook.io/goakt/features/grain

Details the mailbox types (unbounded and bounded) available for Grains. It explains how to configure a bounded mailbox using `WithGrainMailboxCapacity` and the implications of reaching mailbox capacity (`ErrMailboxFull`).

```APIDOC
## Mailbox Configuration

### Description
Configures the message queue (mailbox) for a Grain. Grains can use unbounded mailboxes (default) or bounded mailboxes to manage message flow and apply backpressure.

### Method
`WithGrainMailboxCapacity` (GrainOption)

### Endpoint
N/A (Option passed during Grain creation/identity retrieval)

### Parameters
#### `WithGrainMailboxCapacity` Option
- **capacity** (int) - Required - The maximum number of messages the mailbox can hold.

### Request Example
```go
// Creating a Grain with a bounded mailbox of capacity 100
identity, err := actorSystem.GrainIdentity(ctx, accountID, func(ctx context.Context) (goakt.Grain, error) {
	return grain, nil
}, 
WithGrainMailboxCapacity(100))
```

### Response
#### Success Response
- N/A (This is a configuration option during Grain setup.)

#### Error Response
- **ErrMailboxFull** - Returned when attempting to send a message to a Grain whose bounded mailbox is full.
```

--------------------------------

### Define GoAkt Grain Interface

Source: https://tochemey.gitbook.io/goakt/features/grain

Implement the Grain interface to define the behavior of a GoAkt grain. This interface includes methods for activation (OnActivate), message reception (OnReceive), and deactivation (OnDeactivate). Use these methods to manage grain state, handle incoming messages, and release resources.

```go
type Grain interface {
    // OnActivate is called when the grain is loaded into memory.
    // Use this to initialize state or resources.
    //
    // Arguments:
    //   - ctx: context for cancellation and deadlines.
    //   - props: grain properties and system-level references.
    // Returns:
    //   - error: non-nil to indicate activation failure (grain will not be activated).
    OnActivate(ctx context.Context, props *GrainProps) error
    
    // OnReceive is called when the grain receives a message.
    //
    // Arguments:
    //   - ctx: GrainContext containing the message, sender, grain identity, and system references.
    // Behavior:
    //   - Processes the message and updates grain state as needed.
    //   - Always respect cancellation and deadlines via the context in GrainContext.
    //   - Do not retain references to the GrainContext or its fields beyond the method scope.
    OnReceive(ctx *GrainContext)
    
    // OnDeactivate is called before the grain is removed from memory.
    // Use this to persist state and release resources.
    //
    // Arguments:
    //   - ctx: context for cancellation and deadlines.
    //   - props: grain properties and system-level references.
    // Returns:
    //   - error: non-nil to indicate deactivation failure (system may log or handle the failure).
    OnDeactivate(ctx context.Context, props *GrainProps) error
}
```
Tokens:
Raw