goakt

README

Go-Akt

Minimal actor framework with goodies to build reactive and distributed system in golang using protocol buffers.

If you are not familiar with the actor model, the blog post from Brian Storti here is an excellent and short introduction to the actor model. Also, check reference section at the end of the post for more material regarding actor model

Features

• Send a synchronous message to an actor from a non actor system
• Send an asynchronous(fire-and-forget) message to an actor from a non actor system
• Actor to Actor communication (check the examples' folder)
• Enable/Disable Passivation mode to remove/keep idle actors
• PreStart hook for an actor.
• PostStop hook for an actor for a graceful shutdown
• ActorSystem (WIP)
• Actor to Actor communication
• Restart an actor
• (Un)Watch an actor
• Stop and actor
• Create a child actor
• Supervisory Strategy (Restart and Stop directive)
• Behaviors (Become/BecomeStacked/UnBecome/UnBecomeStacked)
• Logger interface with a default logger
• Examples (check the examples' folder)
• Integration with OpenTelemetry for traces and metrics.
• Remoting
  • Actors can send messages to other actors on a remote system
  • Actors can look up other actors' address on a remote system
• Clustering
• More tests

Installation

go get github.com/Tochemey/messages

Example

Send a fire-forget message to an actor from a non actor system

package main

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

  goakt "github.com/tochemey/messages/actors"
  samplepb "github.com/tochemey/messages/examples/protos/pb/v1"
  "github.com/tochemey/goakt/log"
  "go.uber.org/atomic"
)

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

  // use the messages default logger. real-life implement the logger interface`
  logger := log.DefaultLogger
  // create the actor system configuration. kindly in real-life application handle the error
  config, _ := goakt.NewConfig("SampleActorSystem", "127.0.0.1:0",
    goakt.WithExpireActorAfter(10*time.Second),
    goakt.WithLogger(logger),
    goakt.WithActorInitMaxRetries(3))

  // create the actor system. kindly in real-life application handle the error
  actorSystem, _ := goakt.NewActorSystem(config)

  // start the actor system
  _ = actorSystem.Start(ctx)

  // create an actor
  actor := actorSystem.StartActor(ctx, "Ping", NewPinger())

  startTime := time.Now()

  // send some public to the actor
  count := 100
  for i := 0; i < count; i++ {
    _ = goakt.SendAsync(ctx, actor, new(samplepb.Ping))
  }

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

  // log some stats
  logger.Infof("Actor=%s has processed %d public in %s", actor.ActorPath().String(), actor.ReceivedCount(ctx), time.Since(startTime))

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

type Pinger struct {
  mu     sync.Mutex
  count  *atomic.Int32
  logger log.Logger
}

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

func NewPinger() *Pinger {
  return &Pinger{
    mu: sync.Mutex{},
  }
}

func (p *Pinger) PreStart(ctx context.Context) error {
  // set the logger
  p.mu.Lock()
  defer p.mu.Unlock()
  p.logger = log.DefaultLogger
  p.count = atomic.NewInt32(0)
  p.logger.Info("About to Start")
  return nil
}

func (p *Pinger) Receive(ctx goakt.ReceiveContext) {
  switch ctx.Message().(type) {
  case *samplepb.Ping:
    p.logger.Info("received Ping")
    p.count.Add(1)
  default:
    p.logger.Panic(goakt.ErrUnhandled)
  }
}

func (p *Pinger) PostStop(ctx context.Context) error {
  p.logger.Info("About to stop")
  p.logger.Infof("Processed=%d public", p.count.Load())
  return nil
}

Contribution

Contributions are welcome! The project adheres to Semantic Versioning and Conventional Commits. This repo uses Earthly.

To contribute please:

• Fork the repository
• Create a feature branch
• Submit a pull request

Test & Linter

Prior to submitting a pull request, please run:

earthly +test