Poor Man's Concurrency Monad

Concurrent.java

import java.util.ArrayList;
import java.util.List;
import java.util.function.Function;
import java.util.function.Supplier;
import java.util.stream.Collectors;

public class Concurrent<A> {

	private final Function<Function<A, Action>, Action> func;

	public Concurrent(Function<Function<A, Action>, Action> func) {
		this.func = func;
	}

//	public <A, B> Function<Function<B, Action>, Action> flatMap(
//			Function<Function<A, Action>, Action> func,
//			Function<A, Function<Function<B, Action>, Action>> g) {
//		return new Function<Function<B, Action>, Action>() {
//			@Override
//			public Action apply(Function<B, Action> bToAction) {
//				return func.apply((A action) -> g.apply(action).apply(bToAction));
//			}
//		};
//	}

//	public <A, B> Function<Function<B, Action>, Action> flatMap(
//			Function<Function<A, Action>, Action> func,
//			Function<A, Function<Function<B, Action>, Action>> g) {
//		return (Function<B, Action> b) -> func.apply((A action) -> g.apply(action).apply(b));
//	}

	public static <A> Concurrent<A> of(A a) {
		return new Concurrent<A>((Function<A, Action> cont) -> cont.apply(a));
	}

	public <B> Concurrent<B> flatMap(Function<A, Concurrent<B>> mapper) {
		return new Concurrent<>((Function<B, Action> bToAction) -> func.apply((A action) -> mapper.apply(action).func.apply(bToAction)));
	}

	public <B> Concurrent<B> andThen(Concurrent<B> after) {
		return this.flatMap(a -> after);
	}

	private Function<Function<Function<A, Action>, Action>, Action> actionFunction() {
		return (Function<Function<A, Action>, Action> f) -> f.apply(any -> new Stop());
	}

	public Action action() {
		return actionFunction().apply(func);
	}

	private static <A> Function<Function<A, Action>, Action> stopFunction() {
		return (Function<A, Action> functionAction) -> new Stop();
	}

	public static <A> Concurrent<A> stop() {
		return new Concurrent<>(stopFunction());
	}

	private static <A> Function<Function<A, Action>, Action> atomPrime(Supplier<A> x) {
		return (Function<A, Action> c) -> new Atom(new Supplier<Action>() {
            @Override
            public Action get() {
                return new Atom(() -> c.apply(x.get()));
            }
        });
	}

	public static <A> Concurrent<A> atom(Supplier<A> ioA) {
		return new Concurrent<>(atomPrime(ioA));
	}

	public Concurrent<Unit> fork() {
		return new Concurrent<>((Function<Unit, Action> unitActionFunction) -> new Fork(action(), unitActionFunction.apply(Unit.VALUE)));
	}

	public static <A> Concurrent<A> par(Concurrent<A> c1, Concurrent<A> c2) {
		return new Concurrent<>((Function<A, Action> aActionFunction) -> new Fork(c1.action(), c2.action()));
	}

	public Supplier<Unit> run() {
		List<Action> list = new ArrayList<>();
		list.add(this.action());
		return roundRobin(list);
	}

	private static Supplier<Unit> roundRobin(List<Action> list) {
		if (list.isEmpty()) return () -> Unit.VALUE;

		Action action = list.get(0);
		List<Action> remainingList = list.stream().skip(1).collect(Collectors.toList());
		if (action instanceof Stop) {
			return roundRobin(remainingList);
		} else if (action instanceof Fork) {
			Fork fork = (Fork) action;
			List<Action> forkedList = new ArrayList<>();
			forkedList.add(fork.getA1());
			forkedList.add(fork.getA2());
			forkedList.addAll(remainingList);
			return roundRobin(forkedList);
		} else {
			Atom atom = (Atom) action;
			Action next = atom.getAtom().get();
			List<Action> nextList = new ArrayList<>();
			nextList.addAll(remainingList);
			nextList.add(next);
			return roundRobin(nextList);
		}
	}
}