raboof/pekko
1
0
Fork 0
Code Issues Pull requests Projects Releases 6 Packages Wiki Activity Actions 11

Saving the planet and shufflin'

Browse source
This commit is contained in:
Viktor Klang 2012-05-18 13:37:26 +02:00
parent 72f12c89cd
commit 5eba9fceef
10 changed files with 177 additions and 148 deletions
Download patch file Download diff file Expand all files Collapse all files

2
akka-actor-tests/src/test/scala/akka/event/EventStreamSpec.scala
View file

@ -19,7 +19,7 @@ object EventStreamSpec {
loglevel = INFO loglevel = INFO
event-handlers = ["akka.event.EventStreamSpec$MyLog", "%s"] event-handlers = ["akka.event.EventStreamSpec$MyLog", "%s"]
} }
""".format(Logging.StandardOutLoggerName)) """.format(Logging.StandardOutLogger.getClass.getName))
val configUnhandled = ConfigFactory.parseString(""" val configUnhandled = ConfigFactory.parseString("""
akka { akka {

12
akka-actor/src/main/scala/akka/actor/UntypedActor.scala
View file

@ -93,11 +93,17 @@ import akka.japi.{ Creator }
abstract class UntypedActor extends Actor { abstract class UntypedActor extends Actor {
/** /**
* To be implemented by concrete UntypedActor. Defines the message handler. * To be implemented by concrete UntypedActor, this defines the behavior of the
* UntypedActor.
*/ */
@throws(classOf[Exception]) @throws(classOf[Exception])
def onReceive(message: Any): Unit def onReceive(message: Any): Unit
/**
* Returns this UntypedActor's UntypedActorContext
* The UntypedActorContext is not thread safe so do not expose it outside of the
* UntypedActor.
*/
def getContext(): UntypedActorContext = context.asInstanceOf[UntypedActorContext] def getContext(): UntypedActorContext = context.asInstanceOf[UntypedActorContext]
/** /**
@ -150,9 +156,7 @@ abstract class UntypedActor extends Actor {
*/ */
override def postRestart(reason: Throwable): Unit = super.postRestart(reason) override def postRestart(reason: Throwable): Unit = super.postRestart(reason)
final protected def receive = { final protected def receive = { case msg ⇒ onReceive(msg) }
case msg ⇒ onReceive(msg)
}
} }
/** /**

6
akka-actor/src/main/scala/akka/dispatch/Dispatchers.scala
View file

@ -97,6 +97,7 @@ class Dispatchers(val settings: ActorSystem.Settings, val prerequisites: Dispatc
} }
} }
//INTERNAL API
private def config(id: String): Config = { private def config(id: String): Config = {
import scala.collection.JavaConverters._ import scala.collection.JavaConverters._
def simpleName = id.substring(id.lastIndexOf('.') + 1) def simpleName = id.substring(id.lastIndexOf('.') + 1)
@ -106,6 +107,7 @@ class Dispatchers(val settings: ActorSystem.Settings, val prerequisites: Dispatc
.withFallback(defaultDispatcherConfig) .withFallback(defaultDispatcherConfig)
} }
//INTERNAL API
private def idConfig(id: String): Config = { private def idConfig(id: String): Config = {
import scala.collection.JavaConverters._ import scala.collection.JavaConverters._
ConfigFactory.parseMap(Map("id" -> id).asJava) ConfigFactory.parseMap(Map("id" -> id).asJava)
@ -123,9 +125,7 @@ class Dispatchers(val settings: ActorSystem.Settings, val prerequisites: Dispatc
* *
* INTERNAL USE ONLY * INTERNAL USE ONLY
*/ */
private[akka] def from(cfg: Config): MessageDispatcher = { private[akka] def from(cfg: Config): MessageDispatcher = configuratorFrom(cfg).dispatcher()
configuratorFrom(cfg).dispatcher()
}
/** /**
* Creates a MessageDispatcherConfigurator from a Config. * Creates a MessageDispatcherConfigurator from a Config.

56
akka-actor/src/main/scala/akka/dispatch/Mailbox.scala
View file

@ -16,7 +16,10 @@ import akka.actor.ActorSystem
class MessageQueueAppendFailedException(message: String, cause: Throwable = null) extends AkkaException(message, cause) class MessageQueueAppendFailedException(message: String, cause: Throwable = null) extends AkkaException(message, cause)
object Mailbox { /**
* INTERNAL API
*/
private[akka] object Mailbox {
type Status = Int type Status = Int
@ -40,6 +43,7 @@ object Mailbox {
* Mailbox and InternalMailbox is separated in two classes because ActorCell is needed for implementation, * Mailbox and InternalMailbox is separated in two classes because ActorCell is needed for implementation,
* but can't be exposed to user defined mailbox subclasses. * but can't be exposed to user defined mailbox subclasses.
* *
* INTERNAL API
*/ */
private[akka] abstract class Mailbox(val actor: ActorCell, val messageQueue: MessageQueue) private[akka] abstract class Mailbox(val actor: ActorCell, val messageQueue: MessageQueue)
extends SystemMessageQueue with Runnable { extends SystemMessageQueue with Runnable {
@ -244,6 +248,10 @@ private[akka] abstract class Mailbox(val actor: ActorCell, val messageQueue: Mes
} }
} }
/**
* A MessageQueue is the user-message "lane" of an Akka Mailbox.
* It needs to atleast support N producers and 1 consumer thread-safely.
*/
trait MessageQueue { trait MessageQueue {
/** /**
* Try to enqueue the message to this queue, or throw an exception. * Try to enqueue the message to this queue, or throw an exception.
@ -325,6 +333,9 @@ private[akka] trait DefaultSystemMessageQueue { self: Mailbox ⇒
def hasSystemMessages: Boolean = systemQueueGet ne null def hasSystemMessages: Boolean = systemQueueGet ne null
} }
/**
* A QueueBasedMessageQueue is a MessageQueue backed by a java.util.Queue
*/
trait QueueBasedMessageQueue extends MessageQueue { trait QueueBasedMessageQueue extends MessageQueue {
def queue: Queue[Envelope] def queue: Queue[Envelope]
def numberOfMessages = queue.size def numberOfMessages = queue.size
@ -340,11 +351,19 @@ trait QueueBasedMessageQueue extends MessageQueue {
} }
} }
/**
* UnboundedMessageQueueSemantics adds unbounded semantics to a QueueBasedMessageQueue,
* i.e. a non-blocking enqueue and dequeue.
*/
trait UnboundedMessageQueueSemantics extends QueueBasedMessageQueue { trait UnboundedMessageQueueSemantics extends QueueBasedMessageQueue {
def enqueue(receiver: ActorRef, handle: Envelope): Unit = queue add handle def enqueue(receiver: ActorRef, handle: Envelope): Unit = queue add handle
def dequeue(): Envelope = queue.poll() def dequeue(): Envelope = queue.poll()
} }
/**
* BoundedMessageQueueSemantics adds bounded semantics to a QueueBasedMessageQueue,
* i.e. blocking enqueue with timeout
*/
trait BoundedMessageQueueSemantics extends QueueBasedMessageQueue { trait BoundedMessageQueueSemantics extends QueueBasedMessageQueue {
def pushTimeOut: Duration def pushTimeOut: Duration
override def queue: BlockingQueue[Envelope] override def queue: BlockingQueue[Envelope]
@ -360,17 +379,28 @@ trait BoundedMessageQueueSemantics extends QueueBasedMessageQueue {
def dequeue(): Envelope = queue.poll() def dequeue(): Envelope = queue.poll()
} }
/**
* DequeBasedMessageQueue refines QueueBasedMessageQueue to be backed by a java.util.Deque
*/
trait DequeBasedMessageQueue extends QueueBasedMessageQueue { trait DequeBasedMessageQueue extends QueueBasedMessageQueue {
def queue: Deque[Envelope] def queue: Deque[Envelope]
def enqueueFirst(receiver: ActorRef, handle: Envelope): Unit def enqueueFirst(receiver: ActorRef, handle: Envelope): Unit
} }
/**
* UnboundedDequeBasedMessageQueueSemantics adds unbounded semantics to a DequeBasedMessageQueue,
* i.e. a non-blocking enqueue and dequeue.
*/
trait UnboundedDequeBasedMessageQueueSemantics extends DequeBasedMessageQueue { trait UnboundedDequeBasedMessageQueueSemantics extends DequeBasedMessageQueue {
def enqueue(receiver: ActorRef, handle: Envelope): Unit = queue add handle def enqueue(receiver: ActorRef, handle: Envelope): Unit = queue add handle
def enqueueFirst(receiver: ActorRef, handle: Envelope): Unit = queue addFirst handle def enqueueFirst(receiver: ActorRef, handle: Envelope): Unit = queue addFirst handle
def dequeue(): Envelope = queue.poll() def dequeue(): Envelope = queue.poll()
} }
/**
* BoundedMessageQueueSemantics adds bounded semantics to a DequeBasedMessageQueue,
* i.e. blocking enqueue with timeout
*/
trait BoundedDequeBasedMessageQueueSemantics extends DequeBasedMessageQueue { trait BoundedDequeBasedMessageQueueSemantics extends DequeBasedMessageQueue {
def pushTimeOut: Duration def pushTimeOut: Duration
override def queue: BlockingDeque[Envelope] override def queue: BlockingDeque[Envelope]
@ -393,14 +423,14 @@ trait BoundedDequeBasedMessageQueueSemantics extends DequeBasedMessageQueue {
} }
/** /**
* Mailbox configuration. * MailboxType is a factory to create MessageQueues for an optionally provided ActorContext
*/ */
trait MailboxType { trait MailboxType {
def create(owner: Option[ActorContext]): MessageQueue def create(owner: Option[ActorContext]): MessageQueue
} }
/** /**
* It's a case class for Java (new UnboundedMailbox) * UnboundedMailbox is the default unbounded MailboxType used by Akka Actors.
*/ */
case class UnboundedMailbox() extends MailboxType { case class UnboundedMailbox() extends MailboxType {
@ -412,6 +442,9 @@ case class UnboundedMailbox() extends MailboxType {
} }
} }
/**
* BoundedMailbox is the default bounded MailboxType used by Akka Actors.
*/
case class BoundedMailbox( final val capacity: Int, final val pushTimeOut: Duration) extends MailboxType { case class BoundedMailbox( final val capacity: Int, final val pushTimeOut: Duration) extends MailboxType {
def this(settings: ActorSystem.Settings, config: Config) = this(config.getInt("mailbox-capacity"), def this(settings: ActorSystem.Settings, config: Config) = this(config.getInt("mailbox-capacity"),
@ -428,17 +461,20 @@ case class BoundedMailbox( final val capacity: Int, final val pushTimeOut: Durat
} }
/** /**
* Extend me to provide the comparator * UnboundedPriorityMailbox is an unbounded mailbox that allows for priorization of its contents.
* Extend this class and provide the Comparator in the constructor.
*/ */
class UnboundedPriorityMailbox( final val cmp: Comparator[Envelope]) extends MailboxType { class UnboundedPriorityMailbox( final val cmp: Comparator[Envelope], final val initialCapacity: Int) extends MailboxType {
def this(cmp: Comparator[Envelope]) = this(cmp, 11)
final override def create(owner: Option[ActorContext]): MessageQueue = final override def create(owner: Option[ActorContext]): MessageQueue =
new PriorityBlockingQueue[Envelope](11, cmp) with QueueBasedMessageQueue with UnboundedMessageQueueSemantics { new PriorityBlockingQueue[Envelope](initialCapacity, cmp) with QueueBasedMessageQueue with UnboundedMessageQueueSemantics {
final def queue: Queue[Envelope] = this final def queue: Queue[Envelope] = this
} }
} }
/** /**
* Extend me to provide the comparator * BoundedPriorityMailbox is a bounded mailbox that allows for priorization of its contents.
* Extend this class and provide the Comparator in the constructor.
*/ */
class BoundedPriorityMailbox( final val cmp: Comparator[Envelope], final val capacity: Int, final val pushTimeOut: Duration) extends MailboxType { class BoundedPriorityMailbox( final val cmp: Comparator[Envelope], final val capacity: Int, final val pushTimeOut: Duration) extends MailboxType {
@ -452,6 +488,9 @@ class BoundedPriorityMailbox( final val cmp: Comparator[Envelope], final val cap
} }
} }
/**
* UnboundedDequeBasedMailbox is an unbounded MailboxType, backed by a Deque.
*/
case class UnboundedDequeBasedMailbox() extends MailboxType { case class UnboundedDequeBasedMailbox() extends MailboxType {
def this(settings: ActorSystem.Settings, config: Config) = this() def this(settings: ActorSystem.Settings, config: Config) = this()
@ -462,6 +501,9 @@ case class UnboundedDequeBasedMailbox() extends MailboxType {
} }
} }
/**
* BoundedDequeBasedMailbox is an bounded MailboxType, backed by a Deque.
*/
case class BoundedDequeBasedMailbox( final val capacity: Int, final val pushTimeOut: Duration) extends MailboxType { case class BoundedDequeBasedMailbox( final val capacity: Int, final val pushTimeOut: Duration) extends MailboxType {
def this(settings: ActorSystem.Settings, config: Config) = this(config.getInt("mailbox-capacity"), def this(settings: ActorSystem.Settings, config: Config) = this(config.getInt("mailbox-capacity"),

2
akka-actor/src/main/scala/akka/event/DeathWatch.scala
View file

@ -12,7 +12,7 @@ import akka.actor._
* A failed subscribe should also only mean that the Classifier (ActorRef) that is listened to is already shut down * A failed subscribe should also only mean that the Classifier (ActorRef) that is listened to is already shut down
* See LocalDeathWatch for semantics * See LocalDeathWatch for semantics
*/ */
trait DeathWatch extends ActorEventBus with ActorClassifier { abstract class DeathWatch extends ActorEventBus with ActorClassifier {
type Event = Terminated type Event = Terminated
protected final def classify(event: Event): Classifier = event.actor protected final def classify(event: Event): Classifier = event.actor

15
akka-actor/src/main/scala/akka/event/EventBus.scala
View file

@ -182,10 +182,9 @@ trait SubchannelClassification { this: EventBus ⇒
*/ */
trait ScanningClassification { self: EventBus ⇒ trait ScanningClassification { self: EventBus ⇒
protected final val subscribers = new ConcurrentSkipListSet[(Classifier, Subscriber)](new Comparator[(Classifier, Subscriber)] { protected final val subscribers = new ConcurrentSkipListSet[(Classifier, Subscriber)](new Comparator[(Classifier, Subscriber)] {
def compare(a: (Classifier, Subscriber), b: (Classifier, Subscriber)): Int = { def compare(a: (Classifier, Subscriber), b: (Classifier, Subscriber)): Int = compareClassifiers(a._1, b._1) match {
val cM = compareClassifiers(a._1, b._1) case 0 ⇒ compareSubscribers(a._2, b._2)
if (cM != 0) cM case other ⇒ other
else compareSubscribers(a._2, b._2)
} }
}) })
@ -238,7 +237,7 @@ trait ActorClassification { this: ActorEventBus with ActorClassifier ⇒
import java.util.concurrent.ConcurrentHashMap import java.util.concurrent.ConcurrentHashMap
import scala.annotation.tailrec import scala.annotation.tailrec
private val empty = TreeSet.empty[ActorRef] private val empty = TreeSet.empty[ActorRef]
protected val mappings = new ConcurrentHashMap[ActorRef, TreeSet[ActorRef]](mapSize) private val mappings = new ConcurrentHashMap[ActorRef, TreeSet[ActorRef]](mapSize)
@tailrec @tailrec
protected final def associate(monitored: ActorRef, monitor: ActorRef): Boolean = { protected final def associate(monitored: ActorRef, monitor: ActorRef): Boolean = {
@ -320,9 +319,9 @@ trait ActorClassification { this: ActorEventBus with ActorClassifier ⇒
*/ */
protected def mapSize: Int protected def mapSize: Int
def publish(event: Event): Unit = { def publish(event: Event): Unit = mappings.get(classify(event)) match {
val receivers = mappings.get(classify(event)) case null ⇒ ()
if (receivers ne null) receivers foreach { _ ! event } case some ⇒ some foreach { _ ! event }
} }
def subscribe(subscriber: Subscriber, to: Classifier): Boolean = associate(to, subscriber) def subscribe(subscriber: Subscriber, to: Classifier): Boolean = associate(to, subscriber)

167
akka-actor/src/main/scala/akka/event/Logging.scala
View file

@ -29,7 +29,7 @@ trait LoggingBus extends ActorEventBus {
import Logging._ import Logging._
private val guard = new ReentrantGuard private val guard = new ReentrantGuard //Switch to ReentrantReadWrite
private var loggers = Seq.empty[ActorRef] private var loggers = Seq.empty[ActorRef]
private var _logLevel: LogLevel = _ private var _logLevel: LogLevel = _
@ -97,7 +97,7 @@ trait LoggingBus extends ActorEventBus {
val myloggers = val myloggers =
for { for {
loggerName ← defaultLoggers loggerName ← defaultLoggers
if loggerName != StandardOutLoggerName if loggerName != StandardOutLogger.getClass.getName
} yield { } yield {
try { try {
system.dynamicAccess.getClassFor[Actor](loggerName) match { system.dynamicAccess.getClassFor[Actor](loggerName) match {
@ -129,7 +129,7 @@ trait LoggingBus extends ActorEventBus {
case _: InvalidActorNameException ⇒ // ignore if it is already running case _: InvalidActorNameException ⇒ // ignore if it is already running
} }
publish(Debug(logName, this.getClass, "Default Loggers started")) publish(Debug(logName, this.getClass, "Default Loggers started"))
if (!(defaultLoggers contains StandardOutLoggerName)) { if (!(defaultLoggers contains StandardOutLogger.getClass.getName)) {
unsubscribe(StandardOutLogger) unsubscribe(StandardOutLogger)
} }
} catch { } catch {
@ -163,6 +163,9 @@ trait LoggingBus extends ActorEventBus {
publish(Debug(simpleName(this), this.getClass, "all default loggers stopped")) publish(Debug(simpleName(this), this.getClass, "all default loggers stopped"))
} }
/**
* INTERNAL API
*/
private def addLogger(system: ActorSystemImpl, clazz: Class[_ <: Actor], level: LogLevel, logName: String): ActorRef = { private def addLogger(system: ActorSystemImpl, clazz: Class[_ <: Actor], level: LogLevel, logName: String): ActorRef = {
val name = "log" + Extension(system).id() + "-" + simpleName(clazz) val name = "log" + Extension(system).id() + "-" + simpleName(clazz)
val actor = system.systemActorOf(Props(clazz), name) val actor = system.systemActorOf(Props(clazz), name)
@ -361,17 +364,33 @@ object LogSource {
*/ */
object Logging { object Logging {
/**
* Returns a 'safe' getSimpleName for the provided object's Class
* @param obj
* @return the simple name of the given object's Class
*/
def simpleName(obj: AnyRef): String = simpleName(obj.getClass) def simpleName(obj: AnyRef): String = simpleName(obj.getClass)
/**
* Returns a 'safe' getSimpleName for the provided Class
* @param obj
* @return the simple name of the given Class
*/
def simpleName(clazz: Class[_]): String = { def simpleName(clazz: Class[_]): String = {
val n = clazz.getName val n = clazz.getName
val i = n.lastIndexOf('.') val i = n.lastIndexOf('.')
n.substring(i + 1) n.substring(i + 1)
} }
object Extension extends ExtensionKey[LogExt] /**
* INTERNAL API
*/
private[akka] object Extension extends ExtensionKey[LogExt]
class LogExt(system: ExtendedActorSystem) extends Extension { /**
* INTERNAL API
*/
private[akka] class LogExt(system: ExtendedActorSystem) extends Extension {
private val loggerId = new AtomicInteger private val loggerId = new AtomicInteger
def id() = loggerId.incrementAndGet() def id() = loggerId.incrementAndGet()
} }
@ -431,12 +450,6 @@ object Logging {
// these type ascriptions/casts are necessary to avoid CCEs during construction while retaining correct type // these type ascriptions/casts are necessary to avoid CCEs during construction while retaining correct type
val AllLogLevels = Seq(ErrorLevel: AnyRef, WarningLevel, InfoLevel, DebugLevel).asInstanceOf[Seq[LogLevel]] val AllLogLevels = Seq(ErrorLevel: AnyRef, WarningLevel, InfoLevel, DebugLevel).asInstanceOf[Seq[LogLevel]]
val errorFormat = "[ERROR] [%s] [%s] [%s] %s\n%s".intern
val errorFormatWithoutCause = "[ERROR] [%s] [%s] [%s] %s".intern
val warningFormat = "[WARN] [%s] [%s] [%s] %s".intern
val infoFormat = "[INFO] [%s] [%s] [%s] %s".intern
val debugFormat = "[DEBUG] [%s] [%s] [%s] %s".intern
/** /**
* Obtain LoggingAdapter for the given actor system and source object. This * Obtain LoggingAdapter for the given actor system and source object. This
* will use the system’s event stream and include the system’s address in the * will use the system’s event stream and include the system’s address in the
@ -624,27 +637,34 @@ object Logging {
// weird return type due to binary compatibility // weird return type due to binary compatibility
def loggerInitialized(): LoggerInitialized.type = LoggerInitialized def loggerInitialized(): LoggerInitialized.type = LoggerInitialized
/**
* LoggerInitializationException is thrown to indicate that there was a problem initializing a logger
* @param msg
*/
class LoggerInitializationException(msg: String) extends AkkaException(msg) class LoggerInitializationException(msg: String) extends AkkaException(msg)
trait StdOutLogger { trait StdOutLogger {
import java.text.SimpleDateFormat import java.text.SimpleDateFormat
import java.util.Date import java.util.Date
val dateFormat = new SimpleDateFormat("MM/dd/yyyy HH:mm:ss.S") private val dateFormat = new SimpleDateFormat("MM/dd/yyyy HH:mm:ss.S")
private val errorFormat = "[ERROR] [%s] [%s] [%s] %s\n%s".intern
private val errorFormatWithoutCause = "[ERROR] [%s] [%s] [%s] %s".intern
private val warningFormat = "[WARN] [%s] [%s] [%s] %s".intern
private val infoFormat = "[INFO] [%s] [%s] [%s] %s".intern
private val debugFormat = "[DEBUG] [%s] [%s] [%s] %s".intern
def timestamp = dateFormat.format(new Date) def timestamp(): String = synchronized { dateFormat.format(new Date) } // SDF isn't threadsafe
def print(event: Any) { def print(event: Any): Unit = event match {
event match { case e: Error ⇒ error(e)
case e: Error ⇒ error(e) case e: Warning ⇒ warning(e)
case e: Warning ⇒ warning(e) case e: Info ⇒ info(e)
case e: Info ⇒ info(e) case e: Debug ⇒ debug(e)
case e: Debug ⇒ debug(e) case e ⇒ warning(Warning(simpleName(this), this.getClass, "received unexpected event of class " + e.getClass + ": " + e))
case e ⇒ warning(Warning(simpleName(this), this.getClass, "received unexpected event of class " + e.getClass + ": " + e))
}
} }
def error(event: Error) = { def error(event: Error): Unit = {
val f = if (event.cause == Error.NoCause) errorFormatWithoutCause else errorFormat val f = if (event.cause == Error.NoCause) errorFormatWithoutCause else errorFormat
println(f.format( println(f.format(
timestamp, timestamp,
@ -654,21 +674,21 @@ object Logging {
stackTraceFor(event.cause))) stackTraceFor(event.cause)))
} }
def warning(event: Warning) = def warning(event: Warning): Unit =
println(warningFormat.format( println(warningFormat.format(
timestamp, timestamp,
event.thread.getName, event.thread.getName,
event.logSource, event.logSource,
event.message)) event.message))
def info(event: Info) = def info(event: Info): Unit =
println(infoFormat.format( println(infoFormat.format(
timestamp, timestamp,
event.thread.getName, event.thread.getName,
event.logSource, event.logSource,
event.message)) event.message))
def debug(event: Debug) = def debug(event: Debug): Unit =
println(debugFormat.format( println(debugFormat.format(
timestamp, timestamp,
event.thread.getName, event.thread.getName,
@ -689,8 +709,8 @@ object Logging {
override val toString = "StandardOutLogger" override val toString = "StandardOutLogger"
override def !(message: Any)(implicit sender: ActorRef = null): Unit = print(message) override def !(message: Any)(implicit sender: ActorRef = null): Unit = print(message)
} }
val StandardOutLogger = new StandardOutLogger val StandardOutLogger = new StandardOutLogger
val StandardOutLoggerName = StandardOutLogger.getClass.getName
/** /**
* Actor wrapper around the standard output logger. If * Actor wrapper around the standard output logger. If
@ -708,7 +728,7 @@ object Logging {
* Returns the StackTrace for the given Throwable as a String * Returns the StackTrace for the given Throwable as a String
*/ */
def stackTraceFor(e: Throwable): String = e match { def stackTraceFor(e: Throwable): String = e match {
case null | Error.NoCause ⇒ "" case null | Error.NoCause | _: NoStackTrace ⇒ ""
case other ⇒ case other ⇒
val sw = new java.io.StringWriter val sw = new java.io.StringWriter
val pw = new java.io.PrintWriter(sw) val pw = new java.io.PrintWriter(sw)
@ -752,51 +772,51 @@ trait LoggingAdapter {
* These actually implement the passing on of the messages to be logged. * These actually implement the passing on of the messages to be logged.
* Will not be called if is...Enabled returned false. * Will not be called if is...Enabled returned false.
*/ */
protected def notifyError(message: String) protected def notifyError(message: String): Unit
protected def notifyError(cause: Throwable, message: String) protected def notifyError(cause: Throwable, message: String): Unit
protected def notifyWarning(message: String) protected def notifyWarning(message: String): Unit
protected def notifyInfo(message: String) protected def notifyInfo(message: String): Unit
protected def notifyDebug(message: String) protected def notifyDebug(message: String): Unit
/* /*
* The rest is just the widening of the API for the user's convenience. * The rest is just the widening of the API for the user's convenience.
*/ */
def error(cause: Throwable, message: String) { if (isErrorEnabled) notifyError(cause, message) } def error(cause: Throwable, message: String): Unit = { if (isErrorEnabled) notifyError(cause, message) }
def error(cause: Throwable, template: String, arg1: Any) { if (isErrorEnabled) notifyError(cause, format1(template, arg1)) } def error(cause: Throwable, template: String, arg1: Any): Unit = { if (isErrorEnabled) notifyError(cause, format1(template, arg1)) }
def error(cause: Throwable, template: String, arg1: Any, arg2: Any) { if (isErrorEnabled) notifyError(cause, format(template, arg1, arg2)) } def error(cause: Throwable, template: String, arg1: Any, arg2: Any): Unit = { if (isErrorEnabled) notifyError(cause, format(template, arg1, arg2)) }
def error(cause: Throwable, template: String, arg1: Any, arg2: Any, arg3: Any) { if (isErrorEnabled) notifyError(cause, format(template, arg1, arg2, arg3)) } def error(cause: Throwable, template: String, arg1: Any, arg2: Any, arg3: Any): Unit = { if (isErrorEnabled) notifyError(cause, format(template, arg1, arg2, arg3)) }
def error(cause: Throwable, template: String, arg1: Any, arg2: Any, arg3: Any, arg4: Any) { if (isErrorEnabled) notifyError(cause, format(template, arg1, arg2, arg3, arg4)) } def error(cause: Throwable, template: String, arg1: Any, arg2: Any, arg3: Any, arg4: Any): Unit = { if (isErrorEnabled) notifyError(cause, format(template, arg1, arg2, arg3, arg4)) }
def error(message: String) { if (isErrorEnabled) notifyError(message) } def error(message: String): Unit = { if (isErrorEnabled) notifyError(message) }
def error(template: String, arg1: Any) { if (isErrorEnabled) notifyError(format1(template, arg1)) } def error(template: String, arg1: Any): Unit = { if (isErrorEnabled) notifyError(format1(template, arg1)) }
def error(template: String, arg1: Any, arg2: Any) { if (isErrorEnabled) notifyError(format(template, arg1, arg2)) } def error(template: String, arg1: Any, arg2: Any): Unit = { if (isErrorEnabled) notifyError(format(template, arg1, arg2)) }
def error(template: String, arg1: Any, arg2: Any, arg3: Any) { if (isErrorEnabled) notifyError(format(template, arg1, arg2, arg3)) } def error(template: String, arg1: Any, arg2: Any, arg3: Any): Unit = { if (isErrorEnabled) notifyError(format(template, arg1, arg2, arg3)) }
def error(template: String, arg1: Any, arg2: Any, arg3: Any, arg4: Any) { if (isErrorEnabled) notifyError(format(template, arg1, arg2, arg3, arg4)) } def error(template: String, arg1: Any, arg2: Any, arg3: Any, arg4: Any): Unit = { if (isErrorEnabled) notifyError(format(template, arg1, arg2, arg3, arg4)) }
def warning(message: String) { if (isWarningEnabled) notifyWarning(message) } def warning(message: String): Unit = { if (isWarningEnabled) notifyWarning(message) }
def warning(template: String, arg1: Any) { if (isWarningEnabled) notifyWarning(format1(template, arg1)) } def warning(template: String, arg1: Any): Unit = { if (isWarningEnabled) notifyWarning(format1(template, arg1)) }
def warning(template: String, arg1: Any, arg2: Any) { if (isWarningEnabled) notifyWarning(format(template, arg1, arg2)) } def warning(template: String, arg1: Any, arg2: Any): Unit = { if (isWarningEnabled) notifyWarning(format(template, arg1, arg2)) }
def warning(template: String, arg1: Any, arg2: Any, arg3: Any) { if (isWarningEnabled) notifyWarning(format(template, arg1, arg2, arg3)) } def warning(template: String, arg1: Any, arg2: Any, arg3: Any): Unit = { if (isWarningEnabled) notifyWarning(format(template, arg1, arg2, arg3)) }
def warning(template: String, arg1: Any, arg2: Any, arg3: Any, arg4: Any) { if (isWarningEnabled) notifyWarning(format(template, arg1, arg2, arg3, arg4)) } def warning(template: String, arg1: Any, arg2: Any, arg3: Any, arg4: Any): Unit = { if (isWarningEnabled) notifyWarning(format(template, arg1, arg2, arg3, arg4)) }
def info(message: String) { if (isInfoEnabled) notifyInfo(message) } def info(message: String) { if (isInfoEnabled) notifyInfo(message) }
def info(template: String, arg1: Any) { if (isInfoEnabled) notifyInfo(format1(template, arg1)) } def info(template: String, arg1: Any): Unit = { if (isInfoEnabled) notifyInfo(format1(template, arg1)) }
def info(template: String, arg1: Any, arg2: Any) { if (isInfoEnabled) notifyInfo(format(template, arg1, arg2)) } def info(template: String, arg1: Any, arg2: Any): Unit = { if (isInfoEnabled) notifyInfo(format(template, arg1, arg2)) }
def info(template: String, arg1: Any, arg2: Any, arg3: Any) { if (isInfoEnabled) notifyInfo(format(template, arg1, arg2, arg3)) } def info(template: String, arg1: Any, arg2: Any, arg3: Any): Unit = { if (isInfoEnabled) notifyInfo(format(template, arg1, arg2, arg3)) }
def info(template: String, arg1: Any, arg2: Any, arg3: Any, arg4: Any) { if (isInfoEnabled) notifyInfo(format(template, arg1, arg2, arg3, arg4)) } def info(template: String, arg1: Any, arg2: Any, arg3: Any, arg4: Any): Unit = { if (isInfoEnabled) notifyInfo(format(template, arg1, arg2, arg3, arg4)) }
def debug(message: String) { if (isDebugEnabled) notifyDebug(message) } def debug(message: String) { if (isDebugEnabled) notifyDebug(message) }
def debug(template: String, arg1: Any) { if (isDebugEnabled) notifyDebug(format1(template, arg1)) } def debug(template: String, arg1: Any): Unit = { if (isDebugEnabled) notifyDebug(format1(template, arg1)) }
def debug(template: String, arg1: Any, arg2: Any) { if (isDebugEnabled) notifyDebug(format(template, arg1, arg2)) } def debug(template: String, arg1: Any, arg2: Any): Unit = { if (isDebugEnabled) notifyDebug(format(template, arg1, arg2)) }
def debug(template: String, arg1: Any, arg2: Any, arg3: Any) { if (isDebugEnabled) notifyDebug(format(template, arg1, arg2, arg3)) } def debug(template: String, arg1: Any, arg2: Any, arg3: Any): Unit = { if (isDebugEnabled) notifyDebug(format(template, arg1, arg2, arg3)) }
def debug(template: String, arg1: Any, arg2: Any, arg3: Any, arg4: Any) { if (isDebugEnabled) notifyDebug(format(template, arg1, arg2, arg3, arg4)) } def debug(template: String, arg1: Any, arg2: Any, arg3: Any, arg4: Any): Unit = { if (isDebugEnabled) notifyDebug(format(template, arg1, arg2, arg3, arg4)) }
def log(level: Logging.LogLevel, message: String) { if (isEnabled(level)) notifyLog(level, message) } def log(level: Logging.LogLevel, message: String) { if (isEnabled(level)) notifyLog(level, message) }
def log(level: Logging.LogLevel, template: String, arg1: Any) { if (isEnabled(level)) notifyLog(level, format1(template, arg1)) } def log(level: Logging.LogLevel, template: String, arg1: Any): Unit = { if (isEnabled(level)) notifyLog(level, format1(template, arg1)) }
def log(level: Logging.LogLevel, template: String, arg1: Any, arg2: Any) { if (isEnabled(level)) notifyLog(level, format(template, arg1, arg2)) } def log(level: Logging.LogLevel, template: String, arg1: Any, arg2: Any): Unit = { if (isEnabled(level)) notifyLog(level, format(template, arg1, arg2)) }
def log(level: Logging.LogLevel, template: String, arg1: Any, arg2: Any, arg3: Any) { if (isEnabled(level)) notifyLog(level, format(template, arg1, arg2, arg3)) } def log(level: Logging.LogLevel, template: String, arg1: Any, arg2: Any, arg3: Any): Unit = { if (isEnabled(level)) notifyLog(level, format(template, arg1, arg2, arg3)) }
def log(level: Logging.LogLevel, template: String, arg1: Any, arg2: Any, arg3: Any, arg4: Any) { if (isEnabled(level)) notifyLog(level, format(template, arg1, arg2, arg3, arg4)) } def log(level: Logging.LogLevel, template: String, arg1: Any, arg2: Any, arg3: Any, arg4: Any): Unit = { if (isEnabled(level)) notifyLog(level, format(template, arg1, arg2, arg3, arg4)) }
final def isEnabled(level: Logging.LogLevel): Boolean = level match { final def isEnabled(level: Logging.LogLevel): Boolean = level match {
case Logging.ErrorLevel ⇒ isErrorEnabled case Logging.ErrorLevel ⇒ isErrorEnabled
@ -812,14 +832,14 @@ trait LoggingAdapter {
case Logging.DebugLevel ⇒ if (isDebugEnabled) notifyDebug(message) case Logging.DebugLevel ⇒ if (isDebugEnabled) notifyDebug(message)
} }
private def format1(t: String, arg: Any) = arg match { private def format1(t: String, arg: Any): String = arg match {
case a: Array[_] if !a.getClass.getComponentType.isPrimitive ⇒ format(t, a: _*) case a: Array[_] if !a.getClass.getComponentType.isPrimitive ⇒ format(t, a: _*)
case a: Array[_] ⇒ format(t, (a map (_.asInstanceOf[AnyRef]): _*)) case a: Array[_] ⇒ format(t, (a map (_.asInstanceOf[AnyRef]): _*))
case x ⇒ format(t, x) case x ⇒ format(t, x)
} }
def format(t: String, arg: Any*) = { def format(t: String, arg: Any*): String = {
val sb = new StringBuilder val sb = new StringBuilder //FIXME add some decent size hint here
var p = 0 var p = 0
var rest = t var rest = t
while (p < arg.length) { while (p < arg.length) {
@ -829,17 +849,15 @@ trait LoggingAdapter {
rest = "" rest = ""
p = arg.length p = arg.length
} else { } else {
sb.append(rest.substring(0, index)) sb.append(rest.substring(0, index)).append(arg(p))
sb.append(arg(p))
rest = rest.substring(index + 2) rest = rest.substring(index + 2)
p += 1 p += 1
} }
} }
sb.append(rest) sb.append(rest).toString
sb.toString
} }
} }
//FIXME DOCUMENT
class BusLogging(val bus: LoggingBus, val logSource: String, val logClass: Class[_]) extends LoggingAdapter { class BusLogging(val bus: LoggingBus, val logSource: String, val logClass: Class[_]) extends LoggingAdapter {
import Logging._ import Logging._
@ -849,14 +867,9 @@ class BusLogging(val bus: LoggingBus, val logSource: String, val logClass: Class
def isInfoEnabled = bus.logLevel >= InfoLevel def isInfoEnabled = bus.logLevel >= InfoLevel
def isDebugEnabled = bus.logLevel >= DebugLevel def isDebugEnabled = bus.logLevel >= DebugLevel
protected def notifyError(message: String) { bus.publish(Error(logSource, logClass, message)) } protected def notifyError(message: String): Unit = bus.publish(Error(logSource, logClass, message))
protected def notifyError(cause: Throwable, message: String): Unit = bus.publish(Error(cause, logSource, logClass, message))
protected def notifyError(cause: Throwable, message: String) { bus.publish(Error(cause, logSource, logClass, message)) } protected def notifyWarning(message: String): Unit = bus.publish(Warning(logSource, logClass, message))
protected def notifyInfo(message: String): Unit = bus.publish(Info(logSource, logClass, message))
protected def notifyWarning(message: String) { bus.publish(Warning(logSource, logClass, message)) } protected def notifyDebug(message: String): Unit = bus.publish(Debug(logSource, logClass, message))
protected def notifyInfo(message: String) { bus.publish(Info(logSource, logClass, message)) }
protected def notifyDebug(message: String) { bus.publish(Debug(logSource, logClass, message)) }
} }

6
akka-actor/src/main/scala/akka/event/LoggingReceive.scala
View file

@ -26,9 +26,7 @@ object LoggingReceive {
*/ */
def apply(r: Receive)(implicit context: ActorContext): Receive = r match { def apply(r: Receive)(implicit context: ActorContext): Receive = r match {
case _: LoggingReceive ⇒ r case _: LoggingReceive ⇒ r
case _ ⇒ case _ ⇒ if (context.system.settings.AddLoggingReceive) new LoggingReceive(None, r) else r
if (context.system.settings.AddLoggingReceive) new LoggingReceive(None, r)
else r
} }
} }
@ -37,7 +35,7 @@ object LoggingReceive {
* @param source the log source, if not defined the actor of the context will be used * @param source the log source, if not defined the actor of the context will be used
*/ */
class LoggingReceive(source: Option[AnyRef], r: Receive)(implicit context: ActorContext) extends Receive { class LoggingReceive(source: Option[AnyRef], r: Receive)(implicit context: ActorContext) extends Receive {
def isDefinedAt(o: Any) = { def isDefinedAt(o: Any): Boolean = {
val handled = r.isDefinedAt(o) val handled = r.isDefinedAt(o)
val (str, clazz) = LogSource.fromAnyRef(source getOrElse context.asInstanceOf[ActorCell].actor) val (str, clazz) = LogSource.fromAnyRef(source getOrElse context.asInstanceOf[ActorCell].actor)
context.system.eventStream.publish(Debug(str, clazz, "received " + (if (handled) "handled" else "unhandled") + " message " + o)) context.system.eventStream.publish(Debug(str, clazz, "received " + (if (handled) "handled" else "unhandled") + " message " + o))

34
akka-actor/src/main/scala/akka/japi/JavaAPI.scala
View file

@ -24,28 +24,14 @@ trait Function2[T1, T2, R] {
* A Procedure is like a Function, but it doesn't produce a return value. * A Procedure is like a Function, but it doesn't produce a return value.
*/ */
trait Procedure[T] { trait Procedure[T] {
def apply(param: T) def apply(param: T): Unit
}
/**
* A Procedure is like a Function, but it doesn't produce a return value.
*/
trait Procedure2[T1, T2] {
def apply(param: T1, param2: T2)
}
/**
* An executable piece of code that takes no parameters and doesn't return any value.
*/
trait SideEffect {
def apply()
} }
/** /**
* An executable piece of code that takes no parameters and doesn't return any value. * An executable piece of code that takes no parameters and doesn't return any value.
*/ */
trait Effect { trait Effect {
def apply() def apply(): Unit
} }
/** /**
@ -67,9 +53,9 @@ sealed abstract class Option[A] extends java.lang.Iterable[A] {
def get: A def get: A
def isEmpty: Boolean def isEmpty: Boolean
def isDefined = !isEmpty def isDefined: Boolean = !isEmpty
def asScala: scala.Option[A] def asScala: scala.Option[A]
def iterator = if (isEmpty) Iterator.empty else Iterator.single(get) def iterator: java.util.Iterator[A] = if (isEmpty) Iterator.empty else Iterator.single(get)
} }
object Option { object Option {
@ -102,18 +88,18 @@ object Option {
* <code>A</code>. * <code>A</code>.
*/ */
final case class Some[A](v: A) extends Option[A] { final case class Some[A](v: A) extends Option[A] {
def get = v def get: A = v
def isEmpty = false def isEmpty: Boolean = false
def asScala = scala.Some(v) def asScala: scala.Some[A] = scala.Some(v)
} }
/** /**
* This case object represents non-existent values. * This case object represents non-existent values.
*/ */
private case object None extends Option[Nothing] { private case object None extends Option[Nothing] {
def get = throw new NoSuchElementException("None.get") def get: Nothing = throw new NoSuchElementException("None.get")
def isEmpty = true def isEmpty: Boolean = true
def asScala = scala.None def asScala: scala.None.type = scala.None
} }
implicit def java2ScalaOption[A](o: Option[A]): scala.Option[A] = o.asScala implicit def java2ScalaOption[A](o: Option[A]): scala.Option[A] = o.asScala

25
akka-actor/src/main/scala/akka/util/LockUtil.scala
View file

@ -7,17 +7,12 @@ package akka.util
import java.util.concurrent.locks.{ ReentrantLock } import java.util.concurrent.locks.{ ReentrantLock }
import java.util.concurrent.atomic.{ AtomicBoolean } import java.util.concurrent.atomic.{ AtomicBoolean }
final class ReentrantGuard { final class ReentrantGuard extends ReentrantLock {
final val lock = new ReentrantLock
@inline @inline
final def withGuard[T](body: ⇒ T): T = { final def withGuard[T](body: ⇒ T): T = {
lock.lock lock()
try { try body finally unlock()
body
} finally {
lock.unlock
}
} }
} }
@ -104,19 +99,13 @@ class Switch(startAsOn: Boolean = false) {
* Executes the provided action and returns its value if the switch is on, waiting for any pending changes to happen before (locking) * Executes the provided action and returns its value if the switch is on, waiting for any pending changes to happen before (locking)
* Be careful of longrunning or blocking within the provided action as it can lead to deadlocks or bad performance * Be careful of longrunning or blocking within the provided action as it can lead to deadlocks or bad performance
*/ */
def whileOnYield[T](action: ⇒ T): Option[T] = synchronized { def whileOnYield[T](action: ⇒ T): Option[T] = synchronized { if (switch.get) Some(action) else None }
if (switch.get) Some(action)
else None
}
/** /**
* Executes the provided action and returns its value if the switch is off, waiting for any pending changes to happen before (locking) * Executes the provided action and returns its value if the switch is off, waiting for any pending changes to happen before (locking)
* Be careful of longrunning or blocking within the provided action as it can lead to deadlocks or bad performance * Be careful of longrunning or blocking within the provided action as it can lead to deadlocks or bad performance
*/ */
def whileOffYield[T](action: ⇒ T): Option[T] = synchronized { def whileOffYield[T](action: ⇒ T): Option[T] = synchronized { if (!switch.get) Some(action) else None }
if (!switch.get) Some(action)
else None
}
/** /**
* Executes the provided action and returns if the action was executed or not, if the switch is on, waiting for any pending changes to happen before (locking) * Executes the provided action and returns if the action was executed or not, if the switch is on, waiting for any pending changes to happen before (locking)
@ -144,9 +133,7 @@ class Switch(startAsOn: Boolean = false) {
* Executes the provided callbacks depending on if the switch is either on or off waiting for any pending changes to happen before (locking) * Executes the provided callbacks depending on if the switch is either on or off waiting for any pending changes to happen before (locking)
* Be careful of longrunning or blocking within the provided action as it can lead to deadlocks or bad performance * Be careful of longrunning or blocking within the provided action as it can lead to deadlocks or bad performance
*/ */
def fold[T](on: ⇒ T)(off: ⇒ T) = synchronized { def fold[T](on: ⇒ T)(off: ⇒ T): T = synchronized { if (switch.get) on else off }
if (switch.get) on else off
}
/** /**
* Executes the given code while holding this switch’s lock, i.e. protected from concurrent modification of the switch status. * Executes the given code while holding this switch’s lock, i.e. protected from concurrent modification of the switch status.