Minimalist Java Actors

Actor.java

// ©2012 Viktor Klang
// 6,016 bytes jarred.
package java.klang;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
public class Actor {
  public static interface Fun<T, R> { public R apply(T t); }
  public static interface Effect extends Fun<Behavior, Behavior> { };
  public static interface Behavior extends Fun<Object, Effect> { };
  public static interface Address { Address tell(Object msg); };
  static abstract class AtomicRunnableAddress implements Runnable, Address { protected final AtomicInteger on = new AtomicInteger(); };
  public final static Effect Become(final Behavior behavior) { return new Effect() { public Behavior apply(Behavior old) { return behavior; } }; }
  public final static Effect Stay = new Effect() { public Behavior apply(Behavior old) { return old; } };
  public final static Effect Die  = Become(new Behavior() { public Effect apply(Object msg) { return Stay; } });
  public static Address create(final Fun<Address, Behavior> initial, final Executor e) {
    final Address a = new AtomicRunnableAddress() {
      private final ConcurrentLinkedQueue<Object> mb = new ConcurrentLinkedQueue<Object>();
      private Behavior behavior = new Behavior() { public Effect apply(Object msg) { return (msg instanceof Address) ? Become(initial.apply((Address)msg)) : Stay; } };
      public final Address tell(Object msg) { if (mb.offer(msg)) async(); return this; }
      public final void run() { try { behavior = behavior.apply(mb.poll()).apply(behavior); } finally { on.set(0); async(); } }
      private final void async() { if(!mb.isEmpty() && on.compareAndSet(0, 1)) try { e.execute(this); } catch(RuntimeException re) { on.set(0); throw re; } }
    };
    return a.tell(a); // Make self-aware
  }
}