Advanced OTP Patterns and Real-world Applications - Learn Elixir Part 5

Congratulations! You've made it to the final article in our Elixir learning series. By now, you should have a solid foundation in Elixir programming. In this article, we'll explore advanced OTP patterns, distributed systems, and real-world best practices that will help you build production-ready applications.

Advanced OTP Patterns

OTP provides several specialized behaviors beyond GenServer. These patterns solve specific problems and are worth understanding for complex applications.

GenStateMachine: Managing State Transitions

Sometimes you need to model systems that have distinct states with different behaviors. A traffic light is a perfect example - it cycles through red, yellow, and green, with different timing for each state.

GenStateMachine makes this straightforward:

defmodule TrafficLight do
  use GenStateMachine, callback_mode: :state_functions

  # Client API
  def start_link do
    GenStateMachine.start_link(__MODULE__, :red, name: __MODULE__)
  end

  def next_light do
    GenStateMachine.call(__MODULE__, :next)
  end

  # State functions - each state has its own function
  def red(:call, :next, _data) do
    {:next_state, :yellow, :red_data, [{:state_timeout, 3000, :timeout}]}
  end

  def yellow(:call, :next, _data) do
    {:next_state, :green, :yellow_data, [{:state_timeout, 5000, :timeout}]}
  end

  def green(:call, :next, _data) do
    {:next_state, :red, :green_data, [{:state_timeout, 10000, :timeout}]}
  end
end

The key insight is that each state has its own function that defines how the system behaves in that state. The :state_timeout tells the system to automatically transition after a certain time.

When to Use GenStateMachine

Use GenStateMachine when:

Your system has distinct states with different behaviors
State transitions are well-defined
You need automatic timeouts
The state logic is complex enough to warrant separation

Examples include: order processing systems, game state management, workflow engines, and protocol implementations.

GenEvent: Event Handling

GenEvent is designed for event handling and logging. It allows multiple handlers to process the same events, making it great for building extensible systems.

defmodule EventLogger do
  use GenEvent

  def init(_args) do
    {:ok, []}
  end

  def handle_event({:log, level, message}, state) do
    timestamp = DateTime.utc_now() |> DateTime.to_iso8601()
    IO.puts("[#{timestamp}] #{level}: #{message}")
    {:ok, [message | state]}
  end

  def handle_call({:get_logs}, state) do
    {:ok, Enum.reverse(state), state}
  end
end

# Usage
defmodule EventExample do
  def start_logger do
    {:ok, pid} = GenEvent.start_link(name: :logger)
    GenEvent.add_handler(pid, EventLogger, [])
    pid
  end

  def log_event(level, message) do
    GenEvent.notify(:logger, {:log, level, message})
  end
end

The beauty of GenEvent is that you can add multiple handlers to the same event manager. Each handler processes the same events independently.

Application Behavior: System Startup

The Application behavior is what ties everything together. It's responsible for starting your application and managing its lifecycle.

defmodule MyApp.Application do
  use Application

  def start(_type, _args) do
    children = [
      # Start the database connection pool
      {MyApp.Repo, []},
      
      # Start the main supervisor
      {MyApp.Supervisor, []},
      
      # Start the web endpoint
      {MyAppWeb.Endpoint, []}
    ]

    opts = [strategy: :one_for_one, name: MyApp.Supervisor]
    Supervisor.start_link(children, opts)
  end

  def config_change(changed, _new, removed) do
    MyAppWeb.Endpoint.config_change(changed, removed)
    :ok
  end
end

The Application behavior ensures that your system starts in the right order and handles configuration changes gracefully.

Distributed Elixir

Elixir's distributed capabilities are what make it truly powerful for building scalable systems. You can run your application across multiple machines, and they can communicate seamlessly.

Understanding Distributed Elixir

Distributed Elixir is built on Erlang's distribution system. Each node (machine running Erlang) can communicate with other nodes as if they were local processes. This makes building distributed systems much simpler than with other languages.

Node Communication

defmodule DistributedExample do
  def start_node do
    # Start a node with a name
    Node.start(:"worker@192.168.1.100")
    Node.set_cookie(:my_cookie)
  end

  def connect_to_node(node_name) do
    case Node.connect(node_name) do
      true -> 
        IO.puts("Connected to #{node_name}")
        :ok
      false -> 
        IO.puts("Failed to connect to #{node_name}")
        :error
    end
  end

  def spawn_remote_process(node_name, module, function, args) do
    Node.spawn(node_name, module, function, args)
  end

  def call_remote_function(node_name, module, function, args) do
    :rpc.call(node_name, module, function, args)
  end
end

The key functions are:

Node.start/1: Start a named node
Node.connect/1: Connect to another node
Node.spawn/4: Spawn a process on a remote node
:rpc.call/4: Call a function on a remote node

Distributed Registry

When you have multiple nodes, you need a way to find processes across the cluster. A distributed registry solves this:

defmodule DistributedRegistry do
  use GenServer

  def start_link do
    GenServer.start_link(__MODULE__, [], name: {:via, Registry, {MyRegistry, __MODULE__}})
  end

  def register(name, pid) do
    GenServer.call(__MODULE__, {:register, name, pid})
  end

  def lookup(name) do
    GenServer.call(__MODULE__, {:lookup, name})
  end

  @impl true
  def init(_args) do
    {:ok, %{}}
  end

  @impl true
  def handle_call({:register, name, pid}, _from, state) do
    new_state = Map.put(state, name, pid)
    {:reply, :ok, new_state}
  end

  @impl true
  def handle_call({:lookup, name}, _from, state) do
    result = Map.get(state, name)
    {:reply, result, state}
  end
end

The {:via, Registry, {MyRegistry, __MODULE__}} syntax tells the system to use a distributed registry for process naming.

Cluster Management

Managing a cluster involves monitoring node health, handling node failures, and ensuring data consistency:

defmodule ClusterManager do
  use GenServer

  def start_link do
    GenServer.start_link(__MODULE__, [], name: __MODULE__)
  end

  def join_cluster(node_name) do
    GenServer.call(__MODULE__, {:join, node_name})
  end

  def get_cluster_status do
    GenServer.call(__MODULE__, :get_status)
  end

  @impl true
  def init(_args) do
    # Monitor node connections
    :net_kernel.monitor_nodes(true)
    {:ok, %{nodes: MapSet.new()}}
  end

  @impl true
  def handle_call({:join, node_name}, _from, state) do
    case Node.connect(node_name) do
      true ->
        new_nodes = MapSet.put(state.nodes, node_name)
        {:reply, :ok, %{state | nodes: new_nodes}}
      false ->
        {:reply, {:error, "Failed to connect"}, state}
    end
  end

  @impl true
  def handle_call(:get_status, _from, state) do
    {:reply, MapSet.to_list(state.nodes), state}
  end

  # Handle node up/down events
  @impl true
  def handle_info({:nodeup, node}, state) do
    IO.puts("Node #{node} joined the cluster")
    new_nodes = MapSet.put(state.nodes, node)
    {:noreply, %{state | nodes: new_nodes}}
  end

  @impl true
  def handle_info({:nodedown, node}, state) do
    IO.puts("Node #{node} left the cluster")
    new_nodes = MapSet.delete(state.nodes, node)
    {:noreply, %{state | nodes: new_nodes}}
  end
end

The :net_kernel.monitor_nodes/1 function tells the system to send messages when nodes join or leave the cluster.

Building Microservices with Elixir

Elixir is excellent for building microservices because of its lightweight processes and built-in distribution capabilities.

Service Discovery

In a microservices architecture, services need to find each other. Here's a simple service discovery pattern:

defmodule ServiceDiscovery do
  use GenServer

  def start_link do
    GenServer.start_link(__MODULE__, [], name: {:via, Registry, {MyRegistry, __MODULE__}})
  end

  def register_service(name, endpoint) do
    GenServer.call(__MODULE__, {:register, name, endpoint})
  end

  def discover_service(name) do
    GenServer.call(__MODULE__, {:discover, name})
  end

  @impl true
  def init(_args) do
    {:ok, %{}}
  end

  @impl true
  def handle_call({:register, name, endpoint}, _from, state) do
    new_state = Map.put(state, name, endpoint)
    {:reply, :ok, new_state}
  end

  @impl true
  def handle_call({:discover, name}, _from, state) do
    result = Map.get(state, name)
    {:reply, result, state}
  end
end

Circuit Breaker Pattern

Circuit breakers prevent cascading failures in distributed systems:

defmodule CircuitBreaker do
  use GenServer

  def start_link(name) do
    GenServer.start_link(__MODULE__, name, name: {:via, Registry, {MyRegistry, name}})
  end

  def call(name, fun) do
    GenServer.call({:via, Registry, {MyRegistry, name}}, {:call, fun})
  end

  @impl true
  def init(name) do
    {:ok, %{
      name: name,
      state: :closed,
      failure_count: 0,
      last_failure_time: nil,
      threshold: 5,
      timeout: 60_000
    }}
  end

  @impl true
  def handle_call({:call, fun}, _from, state) do
    case state.state do
      :open ->
        if should_attempt_reset?(state) do
          # Try to reset the circuit
          attempt_call(fun, %{state | state: :half_open})
        else
          {:reply, {:error, :circuit_open}, state}
        end
      
      :closed ->
        attempt_call(fun, state)
      
      :half_open ->
        attempt_call(fun, state)
    end
  end

  defp attempt_call(fun, state) do
    try do
      result = fun.()
      new_state = %{state | 
        state: :closed, 
        failure_count: 0, 
        last_failure_time: nil
      }
      {:reply, {:ok, result}, new_state}
    rescue
      _ ->
        new_failure_count = state.failure_count + 1
        new_state = %{state | 
          failure_count: new_failure_count,
          last_failure_time: DateTime.utc_now()
        }
        
        if new_failure_count >= state.threshold do
          {:reply, {:error, :circuit_open}, %{new_state | state: :open}}
        else
          {:reply, {:error, :service_failed}, new_state}
        end
    end
  end

  defp should_attempt_reset?(state) do
    case state.last_failure_time do
      nil -> true
      time -> 
        DateTime.diff(DateTime.utc_now(), time, :millisecond) > state.timeout
    end
  end
end

The circuit breaker has three states:

Closed: Normal operation
Open: Service is failing, reject requests
Half-open: Testing if service has recovered

Best Practices and Design Patterns

Process Design Principles

When designing processes in Elixir, follow these principles:

Single Responsibility: Each process should have one clear purpose
Message Passing: Use messages for communication, not shared state
Supervision: Always supervise processes that can fail
Isolation: Failures should be contained within processes

Error Handling Strategies

defmodule RobustWorker do
  use GenServer

  def start_link do
    GenServer.start_link(__MODULE__, [], name: __MODULE__)
  end

  def process_data(data) do
    GenServer.call(__MODULE__, {:process, data})
  end

  @impl true
  def init(_args) do
    {:ok, %{processed_count: 0}}
  end

  @impl true
  def handle_call({:process, data}, _from, state) do
    try do
      result = do_process(data)
      new_state = %{state | processed_count: state.processed_count + 1}
      {:reply, {:ok, result}, new_state}
    rescue
      e ->
        # Log the error but don't crash
        IO.puts("Error processing data: #{inspect(e)}")
        {:reply, {:error, "Processing failed"}, state}
    end
  end

  defp do_process(data) do
    # Simulate processing that might fail
    if :rand.uniform(10) == 1 do
      raise "Random processing error"
    else
      "Processed: #{data}"
    end
  end
end

Configuration Management

Use configuration to make your applications flexible:

# config/config.exs
import Config

config :my_app,
  max_retries: 3,
  timeout: 5000,
  pool_size: 10

config :my_app, MyApp.Worker,
  batch_size: 100,
  interval: 1000

# config/dev.exs
import Config

config :my_app,
  debug: true,
  log_level: :debug

# config/prod.exs
import Config

config :my_app,
  debug: false,
  log_level: :info

Testing Strategies

Testing in Elixir is straightforward and powerful:

defmodule MyAppTest do
  use ExUnit.Case
  doctest MyApp

  setup do
    # Start a test process for each test
    {:ok, pid} = MyApp.Worker.start_link()
    {:ok, worker: pid}
  end

  test "worker processes data correctly", %{worker: worker} do
    result = MyApp.Worker.process_data("test")
    assert {:ok, "Processed: test"} = result
  end

  test "worker handles errors gracefully", %{worker: worker} do
    # Test error handling
    result = MyApp.Worker.process_data("error")
    assert {:error, _} = result
  end

  # Property-based testing
  property "worker always returns valid results" do
    check all data <- string() do
      result = MyApp.Worker.process_data(data)
      assert is_tuple(result)
      assert tuple_size(result) == 2
    end
  end
end

Real-world Project Examples

Chat Application

Here's a simplified chat application that demonstrates many Elixir concepts:

defmodule ChatApp do
  use Application

  def start(_type, _args) do
    children = [
      {Phoenix.PubSub, name: ChatApp.PubSub},
      {ChatApp.RoomSupervisor, []},
      {ChatAppWeb.Endpoint, []}
    ]

    Supervisor.start_link(children, strategy: :one_for_one)
  end
end

defmodule ChatApp.RoomSupervisor do
  use Supervisor

  def start_link do
    Supervisor.start_link(__MODULE__, [], name: __MODULE__)
  end

  @impl true
  def init(_args) do
    children = [
      {ChatApp.RoomRegistry, []},
      {ChatApp.RoomManager, []}
    ]

    Supervisor.init(children, strategy: :one_for_one)
  end
end

defmodule ChatApp.Room do
  use GenServer

  def start_link(room_name) do
    GenServer.start_link(__MODULE__, room_name, name: {:via, Registry, {ChatApp.RoomRegistry, room_name}})
  end

  def join(room_name, user) do
    GenServer.cast({:via, Registry, {ChatApp.RoomRegistry, room_name}}, {:join, user})
  end

  def send_message(room_name, user, message) do
    GenServer.cast({:via, Registry, {ChatApp.RoomRegistry, room_name}}, {:message, user, message})
  end

  @impl true
  def init(room_name) do
    Phoenix.PubSub.subscribe(ChatApp.PubSub, "room:#{room_name}")
    {:ok, %{name: room_name, users: MapSet.new(), messages: []}}
  end

  @impl true
  def handle_cast({:join, user}, state) do
    new_users = MapSet.put(state.users, user)
    broadcast_message(state.name, "System", "#{user} joined the room")
    {:noreply, %{state | users: new_users}}
  end

  @impl true
  def handle_cast({:message, user, message}, state) do
    broadcast_message(state.name, user, message)
    new_messages = [{user, message, DateTime.utc_now()} | state.messages]
    {:noreply, %{state | messages: new_messages}}
  end

  defp broadcast_message(room_name, user, message) do
    Phoenix.PubSub.broadcast(
      ChatApp.PubSub,
      "room:#{room_name}",
      {:message, user, message, DateTime.utc_now()}
    )
  end
end

This chat application demonstrates:

Supervision: RoomSupervisor manages room processes
Registry: RoomRegistry for finding room processes
PubSub: Broadcasting messages to connected clients
GenServer: Managing room state
Process isolation: Each room is a separate process

API Gateway

An API gateway that demonstrates load balancing and circuit breaking:

defmodule ApiGateway do
  use GenServer

  def start_link do
    GenServer.start_link(__MODULE__, [], name: __MODULE__)
  end

  def route_request(path, method, body) do
    GenServer.call(__MODULE__, {:route, path, method, body})
  end

  @impl true
  def init(_args) do
    # Start circuit breakers for each service
    CircuitBreaker.start_link(:user_service)
    CircuitBreaker.start_link(:order_service)
    CircuitBreaker.start_link(:payment_service)
    
    {:ok, %{
      services: %{
        "/users" => :user_service,
        "/orders" => :order_service,
        "/payments" => :payment_service
      }
    }}
  end

  @impl true
  def handle_call({:route, path, method, body}, _from, state) do
    service = find_service(path, state.services)
    
    case service do
      nil ->
        {:reply, {:error, :not_found}, state}
      
      service_name ->
        CircuitBreaker.call(service_name, fn ->
          call_service(service_name, path, method, body)
        end)
        |> then(fn result -> {:reply, result, state} end)
    end
  end

  defp find_service(path, services) do
    Enum.find_value(services, fn {prefix, service} ->
      if String.starts_with?(path, prefix), do: service
    end)
  end

  defp call_service(service_name, path, method, body) do
    # Simulate service call
    case service_name do
      :user_service -> {:ok, %{id: 1, name: "John"}}
      :order_service -> {:ok, %{id: 123, items: ["item1", "item2"]}}
      :payment_service -> {:ok, %{status: "processed"}}
    end
  end
end

What's Next?

Congratulations! You've completed a comprehensive journey through Elixir programming. You now have the knowledge and skills to build robust, scalable applications.

Continuing Your Journey

Here are some ways to continue learning:

Build Real Projects: Start with small applications and gradually increase complexity
Join the Community: Participate in Elixir forums, Discord channels, and meetups
Read the Documentation: Elixir has excellent documentation
Contribute to Open Source: Many Elixir projects welcome contributions
Explore the Ecosystem: Look into libraries like Nerves (IoT), Broadway (data processing), and Livebook (notebooks)

Resources

Official Documentation: https://elixir-lang.org/docs.html
Phoenix Framework: https://hexdocs.pm/phoenix/
Ecto: https://hexdocs.pm/ecto/
Elixir Forum: https://elixirforum.com/
Elixir Weekly: Newsletter with latest updates

Practice Exercises

Try these advanced exercises to solidify your knowledge:

Distributed System: Build a simple distributed key-value store
Microservice: Create a user service with authentication and authorization
Real-time Application: Build a collaborative document editor
Data Processing: Implement a stream processing pipeline

Key Points

Advanced OTP patterns solve specific problems in distributed systems
Distributed Elixir makes building cluster-aware applications straightforward
Microservices benefit from Elixir's lightweight processes and fault tolerance
Best practices focus on process isolation, supervision, and error handling
Real-world applications combine multiple Elixir concepts
The Elixir ecosystem provides tools for almost any application type
Continuous learning and community involvement are key to mastery

You're now ready to build amazing applications with Elixir!

Advanced OTP Patterns and Real-world Applications - Learn Elixir Part 5