Hello,
I am trying to implement a simple WebSockets server using akka-http with
separate Sink and Source. According to the documentation I should be using
the handleMessagesWithSinkSource method however if I do so I am not able to
receive any Message. I am able to send some back to the client but nothing
sent to the server is received. Could someone help me out and spot my
mistake? I guess it will be something obvious.
import akka.actor.ActorSystem
import akka.http.scaladsl.Http
import akka.http.scaladsl.Http.IncomingConnection
import akka.http.scaladsl.model.HttpMethods.GET
import akka.http.scaladsl.model.ws.{Message, TextMessage,
UpgradeToWebSocket}
import akka.http.scaladsl.model.{HttpRequest, HttpResponse, Uri}
import akka.stream.ActorMaterializer
import akka.stream.scaladsl.{Sink, Source}
import com.typesafe.config.ConfigFactory
import com.typesafe.scalalogging.StrictLogging
import scala.concurrent.duration.Duration
import scala.concurrent.{Await, ExecutionContextExecutor, Promise}
object Main extends StrictLogging {
def main(args: Array[String]): Unit = {
for {
config <- asResource(ConfigFactory.load())
system <- asResource(ActorSystem("system", config), Duration("10s"))
materializer <- asResource(ActorMaterializer()(system))
} {
implicit val s: ActorSystem = system
implicit val m: ActorMaterializer = materializer
implicit val ec: ExecutionContextExecutor = system.dispatcher
val listenAddress = config.getString("listenAddress")
val listenPort = config.getInt("listenPort")
val server = Http().bind(listenAddress, listenPort)
val requestHandler: HttpRequest => HttpResponse = {
case req @ HttpRequest(GET, Uri.Path("/websocket"), _, _, _) =>
req.header[UpgradeToWebSocket] match {
case Some(upgrade) =>
logger.debug(s"Upgrading: $req")
val inHandler = Sink.foreach[Message](x =>
logger.debug(s"$x"))
upgrade.handleMessagesWithSinkSource(inHandler,
Source.single(TextMessage("hello")))
case None => HttpResponse(400, entity = "Expected 'Upgrade:
websocket' HTTP header")
}
case req: HttpRequest =>
logger.debug(s"Unexpected: $req")
req.discardEntityBytes() // important to drain incoming HTTP
Entity stream
HttpResponse(404, entity = "URL match not found")
}
val connectionHandler = Sink.foreach { connection: IncomingConnection
=>
logger.debug(s"Connection accepted from
${connection.remoteAddress}")
connection.handleWithSyncHandler(requestHandler)
}
val binding = server.to(connectionHandler).run()
binding.failed.foreach { ex =>
logger.error(s"Server could not be bound to
$listenAddress:$listenPort", ex)
}
StdIn.readLine()
}
}
}
Thank you. Regards,
Jakub Janecek
--
>>>>>>>>>> Read the docs: http://akka.io/docs/
>>>>>>>>>> Check the FAQ:
>>>>>>>>>> http://doc.akka.io/docs/akka/current/additional/faq.html
>>>>>>>>>> Search the archives: https://groups.google.com/group/akka-user
---
You received this message because you are subscribed to the Google Groups "Akka
User List" group.
To unsubscribe from this group and stop receiving emails from it, send an email
to [email protected].
To post to this group, send email to [email protected].
Visit this group at https://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.
