ODIn/spark-instrumented-optimizer
1
0
Fork 0
Code Issues Pull requests Projects Releases Wiki Activity

Fix sendMessageReliablySync

Browse source
This commit is contained in:
Matei Zaharia 2012-08-26 16:26:06 -07:00
parent e7a5cbb543
commit 741899b21e
1 changed files with 3 additions and 2 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

5
core/src/main/scala/spark/network/ConnectionManager.scala
View file

@ -14,7 +14,8 @@ import scala.collection.mutable.SynchronizedQueue
import scala.collection.mutable.Queue import scala.collection.mutable.Queue
import scala.collection.mutable.ArrayBuffer import scala.collection.mutable.ArrayBuffer
import akka.dispatch.{Promise, ExecutionContext, Future} import akka.dispatch.{Await, Promise, ExecutionContext, Future}
import akka.util.Duration
case class ConnectionManagerId(host: String, port: Int) { case class ConnectionManagerId(host: String, port: Int) {
def toSocketAddress() = new InetSocketAddress(host, port) def toSocketAddress() = new InetSocketAddress(host, port)
@ -325,7 +326,7 @@ class ConnectionManager(port: Int) extends Logging {
} }
def sendMessageReliablySync(connectionManagerId: ConnectionManagerId, message: Message): Option[Message] = { def sendMessageReliablySync(connectionManagerId: ConnectionManagerId, message: Message): Option[Message] = {
sendMessageReliably(connectionManagerId, message)() Await.result(sendMessageReliably(connectionManagerId, message), Duration.Inf)
} }
def onReceiveMessage(callback: (Message, ConnectionManagerId) => Option[Message]) { def onReceiveMessage(callback: (Message, ConnectionManagerId) => Option[Message]) {