First half of fix for #1518.

4 files changed, 47 insertions, 15 deletions

diff --git a/src/actors/scala/actors/Actor.scala b/src/actors/scala/actors/Actor.scala
index 2adf3f5ffd..553019932d 100644
--- a/src/actors/scala/actors/Actor.scala
+++ b/src/actors/scala/actors/Actor.scala
@@ -439,7 +439,7 @@ trait Actor extends AbstractActor with ReplyReactor with ReplyableActor {
 
     var done = false
     while (!done) {
-      val qel = mailbox.extractFirst((m: Any) => f.isDefinedAt(m))
+      val qel = mailbox.extractFirst((m: Any, replyTo: OutputChannel[Any]) => f.isDefinedAt(m))
       if (null eq qel) {
         synchronized {
           // in mean time new stuff might have arrived
@@ -484,7 +484,7 @@ trait Actor extends AbstractActor with ReplyReactor with ReplyableActor {
     }
 
     // first, remove spurious TIMEOUT message from mailbox if any
-    mailbox.extractFirst((m: Any) => m == TIMEOUT)
+    mailbox.extractFirst((m: Any, replyTo: OutputChannel[Any]) => m == TIMEOUT)
 
     val receiveTimeout = () => {
       if (f.isDefinedAt(TIMEOUT)) {
@@ -496,7 +496,7 @@ trait Actor extends AbstractActor with ReplyReactor with ReplyableActor {
 
     var done = false
     while (!done) {
-      val qel = mailbox.extractFirst((m: Any) => f.isDefinedAt(m))
+      val qel = mailbox.extractFirst((m: Any, replyTo: OutputChannel[Any]) => f.isDefinedAt(m))
       if (null eq qel) {
         val todo = synchronized {
           // in mean time new stuff might have arrived
@@ -580,7 +580,7 @@ trait Actor extends AbstractActor with ReplyReactor with ReplyableActor {
     }
 
     // first, remove spurious TIMEOUT message from mailbox if any
-    mailbox.extractFirst((m: Any) => m == TIMEOUT)
+    mailbox.extractFirst((m: Any, replyTo: OutputChannel[Any]) => m == TIMEOUT)
 
     val receiveTimeout = () => {
       if (f.isDefinedAt(TIMEOUT)) {
@@ -592,7 +592,7 @@ trait Actor extends AbstractActor with ReplyReactor with ReplyableActor {
 
     var done = false
     while (!done) {
-      val qel = mailbox.extractFirst((m: Any) => f.isDefinedAt(m))
+      val qel = mailbox.extractFirst((m: Any, replyTo: OutputChannel[Any]) => f.isDefinedAt(m))
       if (null eq qel) {
         val todo = synchronized {
           // in mean time new stuff might have arrived
diff --git a/src/actors/scala/actors/MessageQueue.scala b/src/actors/scala/actors/MessageQueue.scala
index 540a992640..469b24c1c1 100644
--- a/src/actors/scala/actors/MessageQueue.scala
+++ b/src/actors/scala/actors/MessageQueue.scala
@@ -92,30 +92,30 @@ class MessageQueue(protected val label: String) {
 
   /** Removes the n-th message that satisfies the predicate <code>p</code>.
    */
-  def remove(n: Int)(p: Any => Boolean): Option[(Any, OutputChannel[Any])] =
+  def remove(n: Int)(p: (Any, OutputChannel[Any]) => Boolean): Option[(Any, OutputChannel[Any])] =
     removeInternal(n)(p) map (x => (x.msg, x.session))
 
   /** Extracts the first message that satisfies the predicate <code>p</code>
    *  or <code>null</code> if <code>p</code> fails for all of them.
    */
-  def extractFirst(p: Any => Boolean): MessageQueueElement =
+  def extractFirst(p: (Any, OutputChannel[Any]) => Boolean): MessageQueueElement =
     removeInternal(0)(p) orNull
 
-  private def removeInternal(n: Int)(p: Any => Boolean): Option[MessageQueueElement] = {
+  private def removeInternal(n: Int)(p: (Any, OutputChannel[Any]) => Boolean): Option[MessageQueueElement] = {
     var pos = 0
 
     def foundMsg(x: MessageQueueElement) = {
       changeSize(-1)
       Some(x)
     }
-    def test(msg: Any): Boolean =
-      p(msg) && (pos == n || { pos += 1 ; false })
+    def test(msg: Any, session: OutputChannel[Any]): Boolean =
+      p(msg, session) && (pos == n || { pos += 1 ; false })
 
     if (isEmpty)    // early return
       return None
 
     // special handling if returning the head
-    if (test(first.msg)) {
+    if (test(first.msg, first.session)) {
       val res = first
       first = first.next
       if (res eq last)
@@ -128,7 +128,7 @@ class MessageQueue(protected val label: String) {
       var prev = first
 
       while (curr != null) {
-        if (test(curr.msg)) {
+        if (test(curr.msg, curr.session)) {
           prev.next = curr.next
           if (curr eq last)
             last = prev
@@ -161,12 +161,12 @@ private[actors] trait MessageQueueTracer extends MessageQueue
     printQueue("GET %s" format res)
     res
   }
-  override def remove(n: Int)(p: Any => Boolean): Option[(Any, OutputChannel[Any])] = {
+  override def remove(n: Int)(p: (Any, OutputChannel[Any]) => Boolean): Option[(Any, OutputChannel[Any])] = {
     val res = super.remove(n)(p)
     printQueue("REMOVE %s" format res)
     res
   }
-  override def extractFirst(p: Any => Boolean): MessageQueueElement = {
+  override def extractFirst(p: (Any, OutputChannel[Any]) => Boolean): MessageQueueElement = {
     val res = super.extractFirst(p)
     printQueue("EXTRACT_FIRST %s" format res)
     res
diff --git a/src/actors/scala/actors/Reactor.scala b/src/actors/scala/actors/Reactor.scala
index 4a1f5ac051..7fda8678c1 100644
--- a/src/actors/scala/actors/Reactor.scala
+++ b/src/actors/scala/actors/Reactor.scala
@@ -135,7 +135,7 @@ trait Reactor extends OutputChannel[Any] {
     var tmpMbox = startMbox
     var done = false
     while (!done) {
-      val qel = tmpMbox.extractFirst(handlesMessage)
+      val qel = tmpMbox.extractFirst((msg: Any, replyTo: OutputChannel[Any]) => handlesMessage(msg))
       if (tmpMbox ne mailbox)
         tmpMbox.foreach((m, s) => mailbox.append(m, s))
       if (null eq qel) {
diff --git a/src/actors/scala/actors/ReplyReactor.scala b/src/actors/scala/actors/ReplyReactor.scala
index 762ceef4d6..03c97ebdae 100644
--- a/src/actors/scala/actors/ReplyReactor.scala
+++ b/src/actors/scala/actors/ReplyReactor.scala
@@ -61,4 +61,36 @@ trait ReplyReactor extends Reactor with ReplyableReactor {
       scheduleActor(continuation, item._1)
   }
 
+  // assume continuation != null
+  private[actors] override def searchMailbox(startMbox: MessageQueue,
+                                             handlesMessage: Any => Boolean,
+                                             resumeOnSameThread: Boolean) {
+    var tmpMbox = startMbox
+    var done = false
+    while (!done) {
+      val qel = tmpMbox.extractFirst((msg: Any, replyTo: OutputChannel[Any]) => {
+        senders = List(replyTo)
+        handlesMessage(msg)
+      })
+      if (tmpMbox ne mailbox)
+        tmpMbox.foreach((m, s) => mailbox.append(m, s))
+      if (null eq qel) {
+        synchronized {
+          // in mean time new stuff might have arrived
+          if (!sendBuffer.isEmpty) {
+            tmpMbox = new MessageQueue("Temp")
+            drainSendBuffer(tmpMbox)
+            // keep going
+          } else {
+            waitingFor = handlesMessage
+            done = true
+          }
+        }
+      } else {
+        resumeReceiver((qel.msg, qel.session), resumeOnSameThread)
+        done = true
+      }
+    }
+  }
+
 }