我正在使用 Scala 和 Akka 演员编写一个 Interactive Broker API。
我有一个客户端参与者，它连接到服务器并与 IO 管理器通信以发送请求并接收来自 TWS 的响应。连接工作正常，我能够发送请求并获得响应。
然后我会在 1 分钟后自动收到来自 IO 管理器的 PeerClosed 消息。我希望连接保持打开状态，除非我明确关闭它。我试图设置keepOpenOnPeerClosed = true，但它没有任何改变。
这是演员：
class Client(remote: InetSocketAddress, clientId: Int, extraAuth: Boolean, onConnected: Session => Unit, listener: EWrapper) extends Actor {
final val ClientVersion: Int = 63
final val ServerVersion: Int = 38
final val MinServerVerLinking: Int = 70
import Tcp._
import context.system
IO(Tcp) ! Connect(remote)
def receive = {
case CommandFailed(_: Connect) =>
print("connect failed")
context stop self
case c@Connected(remote, local) => {
val connection = sender()
connection ! Register(self, keepOpenOnPeerClosed = true)
context become connected(connection,1)
val clientVersionBytes = ByteString.fromArray(String.valueOf(ClientVersion).getBytes() ++ Array[Byte](0.toByte))
println("Sending Client Version " + clientVersionBytes)
sender() ! Write(clientVersionBytes)
}
}
def connected(connection: ActorRef, serverVersion: Int): Receive = {
case request: Request =>
print("Send request " + request)
connection ! Write(ByteString(request.toBytes(serverVersion)))
case CommandFailed(w: Write) =>
connection ! Close
print("write failed")
case Received(data) => {
println(data)
implicit val is = new DataInputStream(new ByteArrayInputStream(data.toArray))
EventDispatcher.consumers.get(readInt()) match {
case Some(consumer) => {
consumer.consume(listener, serverVersion)
}
case None => {
listener.error(EClientErrors.NoValidId, EClientErrors.UnknownId.code, EClientErrors.UnknownId.msg)
}
}
}
case _ : ConnectionClosed => context stop self
}
如果我使用 IBJts API 连接（使用标准 Java 套接字），我没有相同的行为