Provides a lock per key, and enables sync execution per key.

LockPool.scala

package core

import org.slf4j.LoggerFactory
import scala.collection.mutable

class LockPool{
	type Lock = AnyRef
	case class LockAndCounter(lock: AnyRef, counter: Int)

	val logger = LoggerFactory.getLogger(getClass)
	val pool = mutable.Map[String, LockAndCounter]()

	def syncByKey[T](key: String)(f: => T): T = {
		val lock = retrieveLock(key)
		val res =  try {
			lock.synchronized {
				logger.debug(s"Locked $key")
				val inner = f
				logger.debug(s"Finished f $key")
				inner
			}
		} finally {
			logger.debug(s"Released $key")
			decreaseCounter(key)
		}
		res
	}

	def retrieveLock(key: String): Lock  = synchronized{
		pool.get(key) match {
			case Some(lockAndKey) =>
				val newLockAndKey = lockAndKey.copy(counter = lockAndKey.counter + 1)
				pool.update(key, newLockAndKey)
				newLockAndKey.lock
			case _ =>
				val lockAndCounter = LockAndCounter(new AnyRef{}, 1)
				pool += key -> lockAndCounter
				lockAndCounter.lock
		}
	}

	private def decreaseCounter(key: String) = synchronized{
		pool.get(key) match {
			case Some(lockAndKey) =>
				if(lockAndKey.counter == 1)
					pool -= key
				else
					pool.update(key, lockAndKey.copy(counter = lockAndKey.counter - 1))
			case _ =>
				throw new MatchError(s"Key does not exist in pool: $key")
		}
	}
}