A GenServer in Crystal

bus.cr

class Bus(A, M)
  include GenServer

  def initialize
    @listeners = [] of A
  end

  # public interface

  def subscribe(listener : A) : Nil
    async(:subscribe, listener)
  end

  def unsubscribe(listener : A) : Nil
    async(:unsubscribe, listener)
  end

  def broadcast(message : M) : Nil
    async(:broadcast, message)
  end

  def size
    sync(:size)
  end

  # private implementation

  handle_async(:subscribe, listener : A) do
    @listeners << listener
  end

  handle_async(:unsubscribe, listener : A) do
    @listeners.delete(listener)
  end

  handle_async(:broadcast, message : M) do
    @listeners.each(&.send(message))
  end

  handle_sync(:size) do
    @listeners.size
  end
end

example.cr

require "./bus"

class Foo
  def initialize(@id : Int32)
  end

  def send(message : String)
    puts "foo#{@id}: #{message}"
  end
end

bus = Bus(Foo, String).new
spawn { bus.call }

foo1 = Foo.new(1)
foo2 = Foo.new(2)

bus.subscribe(foo1)
bus.subscribe(foo2)
bus.broadcast("lorem ipsum")
p bus.size

bus.unsubscribe(foo1)
bus.broadcast("dolor sit amet")
p bus.size

bus.terminate

gen_server.cr

require "syn/core/future"
require "syn/core/wait_group"

module GenServer
  # :nodoc:
  annotation Async; end

  # :nodoc:
  annotation Sync; end

  # :nodoc:
  annotation Info; end

  macro included
    macro finished
      \{% types = [] of String %}
      \{% methods = [] of String %}

      \{% for type in [@type] + @type.ancestors %}
        \{% type.methods.each do |d|
          if ann = d.annotation(Async) || d.annotation(Sync) || d.annotation(Info)
            types << {Symbol, d.args.map(&.restriction).splat}
            methods << {d, ann}
          end
        end %}
      \{% end %}

      @mailbox : Channel(\{{ types.uniq.join(" | ").id }})
      @mailbox = Channel(\{{ types.uniq.join(" | ").id }}).new(128)

      protected def call : Nil
        while msg = @mailbox.receive?
          case msg[0]
            \{% for tuple in methods %}
              \{%
                d, ann = tuple
                meth =
                  if ann = d.annotation(Async)
                    :async
                  elsif ann = d.annotation(Sync)
                    :sync
                  elsif ann = d.annotation(Info)
                    :info
                  end
                %}
              \{% if ann %}
                when \{{ann.named_args[:name]}}
                  # TODO :check for missing handle_async_name method to report an explanatory error message
                  handle_\{{meth.id}}_\{{ann.named_args[:name].id}}(
                    \{% for arg, idx in d.args %}
                      msg[\{{idx + 1}}]?.unsafe_as(\{{arg.restriction}}),
                    \{% end %}
                  )
              \{% end %}
            \{% end %}
          else
            raise "unknown message #{msg.inspect}"
          end
        end
      end
    end
  end

  macro async(name, *args)
    @mailbox.send({ {{name}}, {{args.splat}} })
    nil
  end

  macro sync(name, *args)
    %future = Syn::Core::Future(typeof(handle_sync_{{name.id}}({% if args.size > 0 %}{{args.splat}}, {% end %}Pointer(Void).null))).new
    @mailbox.send({ {{name}}, {% if args.size > 0 %}{{args.splat}}, {% end %}pointerof(%future).unsafe_as(Pointer(Void)) })
    %future.get
  end

  macro info(name, *args)
    %wg = Syn::Core::WaitGroup.new(1)
    @mailbox.send({ {{name}}, {% if args.size > 0 %}{{args.splat}}, {% end %}pointerof(%wg) })
    %wg.wait
    nil
  end

  macro handle_async(name, *args)
    @[GenServer::Async(name: {{name}})]
    protected def handle_async_{{name.id}}({{args.splat}})
      {{yield}}
    end
  end

  macro handle_sync(name, *args)
    @[GenServer::Sync(name: {{name}})]
    protected def handle_sync_{{name.id}}({% if args.size > 0 %}{{args.splat}}, {% end %}future : Pointer(Void))
      value = {{yield}}
      future.unsafe_as(Pointer(Syn::Core::Future(typeof(value)))).value.set(value)
      value
    end
  end

  macro handle_info(name, *args)
    @[GenServer::Info(name: {{name}})]
    protected def handle_info_{{name.id}}({% if args.size > 0 %}{{args.splat}}, {% end %}wg : Pointer(Syn::Core::WaitGroup))
      {{yield}}.tap { wg.value.done }
    end
  end

  def trap(agent : Agent, exception : Exception?) : Nil
    sync(:trap, agent, exception)
  end

  def terminate : Nil
    info(:terminate)
    @mailbox.close
  end

  handle_info(:trap, agent : Agent, exception : Exception?) { nil }
  handle_info(:terminate) { nil }
end

shard.yml

name: gen_server_example
version: 0.2.0

dependencies:
  syn:
    github: ysbaddaden/syn