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

Making thread transient in Event and adding WTF comment

Browse source
This commit is contained in:
Viktor Klang 2011-03-18 17:26:53 +01:00
parent 2752d7a44e
commit d903498ba9
2 changed files with 23 additions and 6 deletions
Download patch file Download diff file Expand all files Collapse all files

4
akka-actor/src/main/scala/akka/actor/Actor.scala
View file

@ -117,7 +117,7 @@ object EventHandler extends ListenerManagement {
val DebugLevel = 4
sealed trait Event {
val thread: Thread = Thread.currentThread
@transient val thread: Thread = Thread.currentThread
}
case class Error(cause: Throwable, instance: AnyRef, message: String = "") extends Event
case class Warning(instance: AnyRef, message: String = "") extends Event
@ -145,7 +145,7 @@ object EventHandler extends ListenerManagement {
def notify(event: => AnyRef) = notifyListeners(event)
def notify[T <: Event : ClassManifest](event: => T) {
if (classManifest[T].erasure.asInstanceOf[Class[_ <: Event]] == level) notifyListeners(event)
if (classManifest[T].erasure.asInstanceOf[Class[_ <: Event]] == level) notifyListeners(event) //WTF?
}
def error(cause: Throwable, instance: AnyRef, message: => String) = {

25
akka-actor/src/main/scala/akka/dispatch/Future.scala
View file

@ -72,7 +72,7 @@ object Futures {
*/
def fold[T,R](zero: R, timeout: Long = Actor.TIMEOUT)(futures: Iterable[Future[T]])(foldFun: (R, T) => R): Future[R] = {
if(futures.isEmpty) {
(new DefaultCompletableFuture[R](timeout)) completeWithResult zero
new AlreadyCompletedFuture[R](Right(zero))
} else {
val result = new DefaultCompletableFuture[R](timeout)
val results = new ConcurrentLinkedQueue[T]()
@ -109,7 +109,7 @@ object Futures {
*/
def reduce[T, R >: T](futures: Iterable[Future[T]], timeout: Long = Actor.TIMEOUT)(op: (R,T) => T): Future[R] = {
if (futures.isEmpty)
(new DefaultCompletableFuture[R](timeout)).completeWithException(new UnsupportedOperationException("empty reduce left"))
new AlreadyCompletedFuture[R](Left(new UnsupportedOperationException("empty reduce left")))
else {
val result = new DefaultCompletableFuture[R](timeout)
val seedFound = new AtomicBoolean(false)
@ -319,7 +319,7 @@ sealed trait Future[+T] {
fa complete v.asInstanceOf[Either[Throwable, A]]
else {
try {
f(v.right.get) onComplete (fa.completeWith(_))
fa.completeWith(f(v.right.get))
} catch {
case e: Exception =>
EventHandler notify EventHandler.Error(e, this)
@ -515,8 +515,25 @@ class DefaultCompletableFuture[T](timeout: Long, timeunit: TimeUnit) extends Com
}
private def notify(func: Future[T] => Unit) {
func(this)
try {
func(this)
} catch {
case e: Exception => EventHandler notify EventHandler.Error(e, this)
}
}
private def currentTimeInNanos: Long = MILLIS.toNanos(System.currentTimeMillis)
}
sealed class AlreadyCompletedFuture[T](suppliedValue: Either[Throwable, T]) extends CompletableFuture[T] {
val value = Some(suppliedValue)
def complete(value: Either[Throwable, T]): CompletableFuture[T] = this
def onComplete(func: Future[T] => Unit): Future[T] = { func(this); this }
def awaitResult: Option[Either[Throwable, T]] = value
def resultWithin(time: Long, unit: TimeUnit): Option[Either[Throwable, T]] = value
def await : Future[T] = this
def awaitBlocking : Future[T] = this
def isExpired: Boolean = false
def timeoutInNanos: Long = 0
}