/**
 * This file is part of Erjang - A JVM-based Erlang VM
 *
 * Copyright (c) 2009 by Trifork
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *  
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 **/

package erjang.m.ets;

import java.lang.ref.WeakReference;
import java.util.concurrent.Callable;

import clojure.lang.APersistentMap;
import clojure.lang.IPersistentMap;
import clojure.lang.LockingTransaction;
import clojure.lang.Ref;

import erjang.EAtom;
import erjang.EInteger;
import erjang.EObject;
import erjang.EPID;
import erjang.EProc;
import erjang.ERT;
import erjang.ESeq;
import erjang.ETuple;
import erjang.ETuple2;
import erjang.ErlangError;
import erjang.NotImplemented;

/**
 * Abstract super class for an ETS table
 */
abstract class ETable {

	public static final EAtom am_stm = EAtom.intern("stm");
	
	protected final WeakReference<EProc> owner;
	protected final EAtom access;
	protected final int keypos;
	protected final EPID heirPID;
	protected final EObject heirData;
	protected final EInteger tid;
	protected final EAtom aname;
	protected final boolean is_named;
	protected final EAtom type;
	private Ref mapRef;

	ETable(EProc owner, EAtom type, EInteger tid, EAtom aname, EAtom access, int keypos,
			boolean is_named, EPID heir_pid, EObject heir_data, APersistentMap map) {
		this.type = type;
		this.is_named = is_named;
		this.owner = new WeakReference<EProc>(owner);
		this.tid = tid;
		this.aname = aname;
		this.access = access;
		this.keypos = keypos;
		this.heirPID = heir_pid;
		this.heirData = heir_data;
		try {
			this.mapRef = new Ref(map);
		} catch (Exception e) {
			throw new ErlangError(am_stm);
		}
	}

	/**
	 * Allocate table with specified configuration
	 */
	public static ETable allocate(EProc proc, EInteger tid, EAtom aname,
			EAtom type, EAtom access, int keypos, boolean write_concurrency,
			boolean is_named, EPID heir_pid, EObject heir_data) {

		if (type == Native.am_set || type == Native.am_ordered_set) {
			return new ETableSet(proc, type, tid, aname, access, keypos,
					write_concurrency, is_named, heir_pid, heir_data);
		}

		if (type == Native.am_bag || type == Native.am_duplicate_bag) {
			return new ETableBag(proc, type, tid, aname, access, keypos,
					write_concurrency, is_named, heir_pid, heir_data);
		}

		throw new NotImplemented("ets type=" + type + "; access=" + access
				+ "; keypos=" + keypos + "; write_concurrency="
				+ write_concurrency + "; heir_pid=" + heir_pid);
	}

	/**
	 * @param caller
	 * @param access
	 * @return
	 */
	public final boolean allow_access(EProc caller, boolean write_access) {
		if (access == Native.am_protected) {
			if (write_access) {
				return (caller == owner.get());
			} else {
				return true;
			}
		} else if (access == Native.am_public) {
			return true;
		} else if (access == Native.am_private) {
			return (caller == owner.get());
		} else {
			throw new InternalError("invalid access mode");
		}
	}

	EPID owner_pid() {
		EProc o = owner.get();
		if (o == null)
			throw ERT.badarg();
		return o.self_handle();
	}

	ESeq info() {
		ESeq res = ERT.NIL;

		res = res.cons(new ETuple2(Native.am_owner, owner_pid()));
		res = res.cons(new ETuple2(Native.am_heir, heirPID));
		res = res.cons(new ETuple2(Native.am_name, aname));
		res = res.cons(new ETuple2(Native.am_size, ERT.box(size())));
		res = res.cons(new ETuple2(Native.am_node, ERT.getLocalNode().node()));
		res = res.cons(new ETuple2(Native.am_named_table, ERT.box(is_named)));
		res = res.cons(new ETuple2(Native.am_type, type));
		res = res.cons(new ETuple2(Native.am_keypos, ERT.box(keypos)));
		res = res.cons(new ETuple2(Native.am_protection, access));

		return res;
	}

	abstract int size();

	/** utility for subclasses */
	EObject get_key(ETuple value) {
		return value.elm(keypos);
	}

	/** utility for subclasses */
	EObject get_key(EObject value) {
		ETuple tuple;
		if ((tuple = value.testTuple()) == null)
			throw ERT.badarg();
		return tuple.elm(keypos);
	}
	
	IPersistentMap deref() {
		return (IPersistentMap) mapRef.deref();
	}
	

	abstract class WithMap<T> implements Callable<T> {
		
		@Override
		public final T call() throws Exception {
			return run((IPersistentMap)mapRef.deref());
		}

		protected abstract T run(IPersistentMap map);

		protected void set(IPersistentMap map) {
			mapRef.set(map);
		}
		
	}
	
	@SuppressWarnings("unchecked")
	<X> X in_tx(WithMap<X> run) {
		try {
			return (X) LockingTransaction.runInTransaction(run);
		} catch (Exception e) {
			// STM Failure
			throw new ErlangError(am_stm);
		}
	}
	
	


	protected abstract void insert_one(ETuple value);

	protected abstract void insert_many(ESeq values);

	protected abstract void insert_new_one(ETuple value);

	protected abstract void insert_new_many(ESeq values);

	protected abstract EObject lookup(EObject key);

}

@@ -1185,16 +1185,312 @@
 TS table
+.  We implement ETS by using%0A * Clojure's persistent data types.  %0A * %0A * set:           PersistentHashMap %5Bkey, value%5D%0A * ordered_set:   PersistentTreeMap %5Bkey, value%5D%0A * bag%09%09%09  PersistentHashMap %5Bkey, PersistentSet %5Bvalue%5D%5D%0A * duplicate_bag: PersistentHashMap %5Bkey, PersistentList %5Bvalue%5D%5D%0A * 
 %0A */%0Aabs

document ets implementation