// ©2012 Viktor Klang
// 5,804 bytes compressed.
package java.klang

import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.Executor;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.atomic.AtomicBoolean;

public class Actor {
  public static interface Function<T, R> { public R apply(T t); }
  public static interface Effect { Behavior getOrElse(Behavior old); };
  public static interface Behavior extends Function<Object, Effect> { };
  public static interface Address { void tell(Object message); };

  public final static Effect Stay = new Effect() { public Behavior getOrElse(Behavior old) { return old; } };
  public static class Become implements Effect {
    public final Behavior like;
    public Become(Behavior like) { this.like = like; }
    public Behavior getOrElse(Behavior old) { return like; }
  };
  public final static Become Die = new Become(new Behavior() { 
      public Effect apply(Object message) { System.out.println("Dropping message [" + message + "] due to severe case of death."); return Stay; }
  });

  static class AddressImpl extends AtomicBoolean implements Address, Runnable {
    private final String name;
    private final Executor e;
    private final ConcurrentLinkedQueue mbox = new ConcurrentLinkedQueue<Object>();
    private Behavior behavior;

    AddressImpl(final Function<Address, Behavior> initial, final String name, final Executor e) {
      this.name = name;
      this.e = e;
      this.behavior = new Behavior() {
          public Effect apply(Object message) { return (message instanceof Address) ? new Become(initial.apply((Address)message)) : Stay; }
      };
    }
    @SuppressWarnings("unchecked")
    public final void tell(Object message) {
      if (behavior == Die.like) Die.like.apply(message);
      else {
        mbox.offer(message);
        trySchedule();
      }
    }

    public void run() { try { behavior = behavior.apply(mbox.poll()).getOrElse(behavior); } finally { set(false); trySchedule(); } }
    public final String toString() { return "actor://" + name; }
    private final void trySchedule() {
      if(!mbox.isEmpty() && compareAndSet(false, true)) {
        try { e.execute(this); } catch(RejectedExecutionException ree) { set(false); throw ree; }
      }
    }
  }
 
  public static Address create(final Function<Address, Behavior> initial, final String name, final Executor e) {
    final Address a = new AddressImpl(initial, name, e);
    a.tell(a);
    return a;
  }
}