2

我正在尝试在 akka-http websocket 服务器和简单的 javascript 之间建立基本连接。

大约 20 个连接中有 1 个连接成功，其余连接失败。我不知道为什么连接的设置如此不可靠。

应用程序.scala：

import akka.actor.ActorSystem
import akka.http.scaladsl.Http
import akka.stream.ActorMaterializer
import services.WebService

import scala.concurrent.Await
import scala.concurrent.duration._
import java.util.concurrent.TimeoutException


object Application extends App {

  implicit val system = ActorSystem("api")
  implicit val materializer = ActorMaterializer()

  import system.dispatcher

  val config = system.settings.config
  val interface = config.getString("app.interface")
  val port = config.getInt("app.port")

  val service = new WebService

  val binding = Http().bindAndHandle(service.route, interface, port)

  try {
    Await.result(binding, 1 second)
    println(s"server online at http://$interface:$port/")
  } catch {
    case exc: TimeoutException =>
      println("Server took to long to startup, shutting down")
      system.shutdown()
  }
}

WebService.scala：

import actors.{PublisherActor, SubscriberActor}
import akka.actor.{Props, ActorSystem}
import akka.http.scaladsl.model.ws.{Message, TextMessage}

import akka.http.scaladsl.server.Directives
import akka.stream.Materializer
import akka.stream.scaladsl.{Source, Flow}
import scala.concurrent.duration._


class WebService(implicit fm: Materializer, system: ActorSystem) extends Directives {

  import system.dispatcher
  system.scheduler.schedule(15 second, 15 second) {
    println("Timer message!")
  }

  def route =
    get {
      pathSingleSlash {
        getFromResource("web/index.html")
      } ~
        path("helloworld") {
          handleWebsocketMessages(websocketActorFlow)
        }
    }

  def websocketActorFlow: Flow[Message, Message, Unit] =
    Flow[Message].collect({
      case TextMessage.Strict(msg) =>
        println(msg)
        TextMessage.Strict(msg.reverse)
    })
}

客户端：

<input type="text" id="inputMessage"/><br/>
<input type="button" value="Send!" onClick="sendMessage()"/><br/>
<span id="response"></span>
<script type="application/javascript">

    var connection;
    function sendMessage() {
        connection.send(document.getElementById("inputMessage").value);
    }

    document.addEventListener("DOMContentLoaded", function (event) {
        connection = new WebSocket("ws://localhost:8080/helloworld");

        connection.onopen = function (event) {
            connection.send("connection established");
        };

        connection.onmessage = function (event) {
            console.log(event.data);
            document.getElementById("response").innerHTML = event.data;
        }
    });
</script>

如果与服务器的连接失败，我会在 5 秒后收到一条超时消息，其中包含以下内容：

[DEBUG] [07/23/2015 07:59:54.517] [api-akka.actor.default-dispatcher-27] [akka://api/user/$a/flow-76-3-publisherSource-prefixAndTail] Cancelling akka.stream.impl.MultiStreamOutputProcessor$SubstreamOutput@a54778 (after: 5000 ms)

无论连接失败还是成功，我总是会收到以下日志消息：

[DEBUG] [07/23/2015 07:59:23.849] [api-akka.actor.default-dispatcher-4] [akka://api/system/IO-TCP/selectors/$a/0] New connection accepted
4

1 回答 1

0

仔细查看该错误消息......它来自我没想到的来源，当我只希望处理单个流时，它是一些“MultiStreamOutputProcessor”。

这告诉我 - 以及webSocketActorFlow- 也许你正在接收消息并且它们没有被流捕获，所以你最终得到了你从未预料到的子流。

因此，它可能不是“仅在某些时候工作”，而是“大部分时间都在工作，但无法处理您在流程中要求的所有输入，并且您留下了不可选择的流，必须先死。

看看你是否可以 a) 确保你掌握了流，这样你就不会落伍了，b) 创可贴调整超时，以及 c)检测到此类事件并取消处理下游

https://groups.google.com/forum/#!topic/akka-user/x-tARRaJ0LQ

akka {流{物化器{订阅超时{超时= 30s}}}}}

http://grokbase.com/t/gg/akka-user/1561gr0jgt/debug-message-cancelling-akka-stream-impl-multistreamoutputprocessor-after-5000-ms

于 2016-04-04T01:39:24.693 回答