Replika 🦾

A pure functional finite state machine library for Elixir.

Table of Contents 📖

• Overview
• Installation
• Usage
• Error Handling
• Pattern Matching and Guards
• Returning Values
• Global Event Handlers
• Dynamic FSM Creation
• Performance
• Advanced Usage
• Testing Replika State Machines
• Documentation
• License

Overview 🔍

Replika implements finite state machines as immutable data structures that transform during state transitions. This approach provides:

• High Performance - State transitions are simple struct updates (O(1))
• No Process Overhead - Works as a lightweight data structure
• Immutable & Pure - Predictable state transitions without side effects
• Embeddable - Can be stored in ETS, databases, or embedded in other processes
• Pattern Matching - Pattern matching is optimized by the BEAM VM
• Deterministic Memory Usage - Low memory footprint at scale

Installation 📦

Add Replika to your mix.exs dependencies:

def deps do
  [
    {:replika, "~> 0.1.1"}
  ]
end

Usage 🧑‍🏫

This example shows a state machine for a character accessing secured areas. The character starts with no ID card, can pick one up, approach an elevator, and use the card to gain access.

Each state defines valid transitions to other states, and each transition can include data transformation:

defmodule Unit.AccessState do
  use Replika, initial_state: :no_id_card, initial_data: %{inventory: []}

  defstate no_id_card do
    defevent pickup_card do
      next_state(:has_id_card, %{inventory: [:id_card], floors_visited: []})
    end
  end

  defstate has_id_card do
    defevent approach_elevator do
      next_state(:at_elevator, %{inventory: [:id_card], floors_visited: []})
    end
  end

  defstate at_elevator do
    defevent use_card, data: %{inventory: inventory, floors_visited: visited} do
      if :id_card in inventory do
        next_state(:elevator_accessed, %{inventory: inventory, floors_visited: visited})
      else
        next_state(:at_elevator, %{inventory: inventory, floors_visited: visited})
      end
    end
  end

  defstate elevator_accessed do
    defevent select_floor(floor), data: %{inventory: inv, floors_visited: visited} do
      next_state(:elevator_accessed, %{inventory: inv, floors_visited: [floor | visited]})
    end

    defevent get_visited_floors, data: %{floors_visited: visited} do
      respond(visited)
    end
  end
end

We can then interact with the state machine like this:

# Create a new FSM instance
elster = Unit.AccessState.new()

# Progress through states by calling event functions
elster = elster
         |> Unit.AccessState.pickup_card()
         |> Unit.AccessState.approach_elevator()
         |> Unit.AccessState.use_card()

# Inspect the current state
Unit.AccessState.state(elster)  # Returns :elevator_accessed

# Events can return values while changing state
{floors, elster} = Unit.AccessState.get_visited_floors(elster)

The state machine above defines:

• Four distinct states: :no_id_card, :has_id_card, :at_elevator, and :elevator_accessed
• Events that trigger transitions between states
• Data transformations with each state change
• Conditional transitions based on inventory contents

Each event function returns either the updated FSM instance or a tuple containing a response value AND the updated instance.

Error Handling ⚠️

Replika provides clear error messages when you attempt invalid transitions. If you try to execute an event that isn't defined for the current state, you'll get a helpful error message:

try do
  # Starting with no card
  elster = Unit.AccessState.new()

  # Pick up a card, putting us in :has_id_card state
  elster = Unit.AccessState.pickup_card(elster)

  # Try to pick up a card again - but we're already in :has_id_card state!
  # This raises an InvalidTransitionError since pickup_card is only defined
  # for the :no_id_card state
  Unit.AccessState.pickup_card(elster)
rescue
  e in Replika.Error.InvalidTransitionError ->
    IO.puts(e.message)
    # "Invalid transition: cannot execute event 'pickup_card' with args [] in state ':has_id_card'"
end

The error message includes:

• The current state (:has_id_card)
• The invalid event (pickup_card)
• Any arguments passed to the event ([])

This makes debugging much easier compared to normal function clause errors.

Pattern Matching and Guards 🧩

Pattern matching lets you handle different cases within the same state based on data conditions. Combined with guards, this provides powerful control flow:

defstate at_elevator do
  # First clause: match when ID card is in inventory
  # The guard clause (when) ensures this only matches when :id_card is in inventory
  defevent use_card, data: %{inventory: inventory} when :id_card in inventory do
    next_state(:elevator_accessed, %{inventory: inventory})
  end

  # Second clause: match when ID card is NOT in inventory (fallback case)
  # This handles the case where inventory doesn't contain :id_card
  defevent use_card, data: %{inventory: inventory} do
    # Return an error response and remain in the same state
    respond({:error, :missing_id_card}, :at_elevator, %{inventory: inventory})
  end
end

This pattern lets you:

• Define different behaviors for the same event based on data conditions
• Use guard clauses for precise control over which clause matches
• Return different values or transition to different states based on conditions
• Handle error cases elegantly

Returning Values 📤

Events can return values to the caller while optionally changing state. This is useful for querying the FSM or implementing commands that produce a result:

defstate elevator_accessed do
  # Query event: returns data without changing state
  defevent get_visited_floors, data: %{floors_visited: visited} do
    # First arg is the return value, no state change
    respond(visited)
  end

  # Command event: returns data AND changes state
  defevent select_floor(floor), data: %{inventory: inv, floors_visited: visited} do
    # Return a tuple, update state, and modify data
    respond(
      {:travelling_to, floor},                             # Return value
      :elevator_accessed,                                  # New state (unchanged)
      %{inventory: inv, floors_visited: [floor | visited]} # New data
    )
  end
end

Using respond/1, respond/2, or respond/3:

• respond(value) - Returns value without changing state or data
• respond(value, new_state) - Returns value and changes state
• respond(value, new_state, new_data) - Returns value, changes state, and updates data

When using respond, the event call returns a tuple: {return_value, updated_fsm}.

Global Event Handlers 🌐

Sometimes you need to handle unexpected events or provide fallback behavior. The special _ event can catch undefined events:

defmodule SecuritySystem do
  use Replika, initial_state: :armed, initial_data: %{alert_count: 0}

  defstate armed do
    # Defined events
    defevent disarm(code) when code == 1234 do
      next_state(:disarmed)
    end

    # Catch-all for any other event in the :armed state
    # This will catch any undefined event while in the armed state
    defevent _ do
      # Increment alert count and remain armed
      next_state(:armed, %{alert_count: data.alert_count + 1})
    end
  end

  defstate disarmed do
    defevent arm do
      next_state(:armed, %{alert_count: 0})
    end
  end

  # Global catch-all for any undefined event in any state
  # This only triggers if there's no state-specific handler
  defevent _ do
    respond({:error, :invalid_operation})
  end
end

The event handlers are checked in order:

1. Exact event/state match
2. State-specific catch-all (_ event in the current state)
3. Global catch-all (_ event outside any state)

This gives you complete control over error handling and fallback behavior.

Dynamic FSM Creation 🏗️

Replika's macros enable programmatic FSM definition - useful for state machines with regular patterns:

defmodule ProtektorElevator do
  use Replika, initial_state: :b1, initial_data: %{log: []}

  # Define all possible floors
  floors = [:b1, :b2, :b3, :b4, :b5, :b6]

  # Generate states and transitions dynamically
  for current_floor <- floors do
    defstate current_floor do
      # For each floor, create transitions to all other floors
      for target_floor <- floors, target_floor != current_floor do
        # Create a "go_to_X" event for each possible target floor
        defevent :"go_to_#{target_floor}", data: %{log: log} do
          # Log the transition and update state
          next_state(
            target_floor,
            %{log: ["#{current_floor} -> #{target_floor}" | log]}
          )
        end
      end

      # Add inspection event to see travel history
      defevent :travel_history, data: %{log: log} do
        respond(log)
      end
    end
  end
end

This generates a complete elevator control system with:

• One state for each floor
• Events to travel between any floors
• Event naming that matches the destination (e.g., go_to_b6)
• Automatic logging of all floor transitions

Using metaprogramming this way creates sophisticated state machines with minimal code.

Performance ⏱️

Replika FSMs are lightweight data structures that make state transitions in constant time. There's no process creation, message passing, or serialization overhead. This makes Replika well-suited for high-throughput and memory-constrained applications.

Run the benchmarks to compare with gen_state_machine, an Elixir wrapper for Erlang's gen_statem:

mix bench

or

mix run bench/replika_bench.exs

Advanced Usage 🔧

While Replika provides exceptional performance as a pure functional state machine, you can combine it with OTP's process model when you need features like supervision or timeouts.

defmodule Protektor.ElevatorServer do
  use GenServer

  def start_link(args), do: GenServer.start_link(__MODULE__, args)

  def init(args), do: {:ok, Unit.AccessState.new(args)}

  # Expose FSM events as server API
  def pickup_card(pid), do: GenServer.call(pid, :pickup_card)
  def use_card(pid), do: GenServer.call(pid, :use_card)

  # Handle the events in the server
  def handle_call(:pickup_card, _from, fsm) do
    new_fsm = Unit.AccessState.pickup_card(fsm)
    {:reply, :ok, new_fsm}
  end

  def handle_call(:use_card, _from, fsm) do
    new_fsm = Unit.AccessState.use_card(fsm)
    {:reply, :ok, new_fsm}
  end

  # Timeout example
  def handle_info(:security_timeout, fsm) do
    IO.puts("ALERT: Security timeout - initiating lockdown")
    {:noreply, fsm}
  end
end

This hybrid approach allows you to take advantage of:

• OTP Integration - Supervision trees and fault tolerance
• Distribution - Run FSMs across a cluster
• Timeouts - Support for time-based transitions and alerts
• Long-Running Operations - Background tasks and periodic checks

In addition, you still get the performance benefits of Replika's faster transitions, lower resource use, and simpler state logic.

Use this pattern when you need both the reliability of OTP and the performance of Replika.

Testing Replika State Machines 🧪

Testing Replika state machines is straightforward since they're just data structures:

defmodule Unit.AccessStateTest do
  use ExUnit.Case

  test "unit can access elevator with ID card" do
    # Start with no ID card
    unit = Unit.AccessState.new()
    assert Unit.AccessState.state(unit) == :no_id_card

    # Pick up ID card and approach elevator
    unit = unit
           |> Unit.AccessState.pickup_card()
           |> Unit.AccessState.approach_elevator()

    assert Unit.AccessState.state(unit) == :at_elevator

    # Use card to access elevator
    unit = Unit.AccessState.use_card(unit)
    assert Unit.AccessState.state(unit) == :elevator_accessed

    # Check visited floors
    {floors, _unit} = Unit.AccessState.get_visited_floors(unit)
    assert floors == []

    # Select a floor
    {response, unit} = Unit.AccessState.select_floor(unit, "b6")
    assert response == {:travelling_to, "b6"}

    # Verify floor was added to visited floors
    {floors, _unit} = Unit.AccessState.get_visited_floors(unit)
    assert "b6" in floors
  end

  test "raises error for invalid transitions" do
    unit = Unit.AccessState.new()

    assert_raise Replika.Error.InvalidTransitionError, fn ->
      # Can't use card without approaching elevator first
      Unit.AccessState.use_card(unit)
    end

    unit = Unit.AccessState.pickup_card(unit)

    assert_raise Replika.Error.InvalidTransitionError, fn ->
      # Can't pick up card when already have one
      Unit.AccessState.pickup_card(unit)
    end
  end
end

Since Replika state machines are deterministic and free from side-effects, they're easy to test without mocks or complex setups. You can create instances, apply transitions, and verify both state changes and returned values directly.

Documentation 📚

For detailed API documentation, go to:

• Replika - Main module documentation
• Replika.Error - Error handling

License 📄

This project is licensed under: