lihaoyi / spraywebsockets Goto Github PK

The following log is shown when I send Sockets.UpgradeServer(Sockets.acceptAllFunction(req), self) to sender. I have experienced this issue when run SocketServer in SocketExample.

[WARN] [02/11/2014 02:04:17.685] [example-akka.actor.default-dispatcher-2] [akka
://example/user/IO-SOCKET/listener-0/0] event pipeline: dropped AckEventWithRece
iver(NoAck(null),HttpResponse(101 Switching Protocols,Empty,List(Upgrade: websoc
ket, Connection: Upgrade, Sec-WebSocket-Accept: ???),HTTP/1.1),Actor[akka://exam
ple/user/server#-1103739125])

i tried to compile the source against spray 1.2-RC2 but was not successful. apparently there were quite a few changes from spray 1.2-M8 which spraywebsockets was compiled against originally. do you have any plan on updating it to be compatible with 1.2-RC2?

I'm using SprayWebSockets to communicate with browser. Everything works on the local environment, when server and client on the same machine. But on the production environment there is a strange behavior: sometimes events from server doesn't reach the client (Google Chrome). Events size and count is the same as on the local environment. After this behavior socket stops working and doesn't send any data from server to connected socket client. At the same time there are log messages for each unreached event:

event pipeline: dropped CommandFailed(Write(ByteString....))

I'm using latest SprayWebSockets.

I haven't had the time or drive to keep up with this project, and by this point I'm as much a stranger to the code as anyone else would be. It would be great to find someone to take over the project so that the hard work people put into PRs and bugfixes can be merged, and it can be brought up to date with the latest version(s) of Spray.

Hi,

I really love your project. Nice job!

I have already an website based on spray.routing.HttpService and now I want to extend some parts of it with Websocket. Is SprayWebSockets possible to multiplex the same port of an existing actor? For example, dispatch all requests with path of /websocket to SprayWebSockets and remain other paths.

Thanks

I'm attempting to create chat-rooms, using Something like this:

class ChatService extends Actor with ChatMarshaller {
    val mediator = DistributedPubSubExtension(context.system).mediator
    def receive = {
        case x: Tcp.Connected => sender ! Register(self)
        case req: HttpRequest => sender ! Sockets.UpgradeServer(Sockets.acceptAllFunction(req), self)
        case Sockets.Upgraded => mediator ! Subscribe("mychatroom", sender)
        case f @ Frame(fin, rsv, Text, maskingKey, data) => {
            mediator ! Publish("mychatroom", Frame(opcode = OpCode.Text, data = ByteString(f.stringData)))
        }
    }
}

as you can see, I'm subscribing the sender to "mychatroom," and I'm also using the DistributedPubSubExtension mediator to to publish the message to all subscribers of "mychatroom"

The problem I'm having is, I'd like to create a unique key for "mychatroom" that's passed in from the request, but I'm not sure how to do that with the example above.

Any ideas?

Hi I'm trying to use spray websockets. When issuing the upgrade message I get the following warning message:

[SocketsSystem-akka.actor.default-dispatcher-2] [akka://SocketsSystem/user/IO-HTTP/listener-0/1] command pipeline: dropped spray.can.server.websockets.Sockets$UpgradeServer@3bc932f2

Then I get a timed out message shortly afterwards.

The case match looks like this

case req @ HttpRequest(
      HttpMethods.GET,
      Uri( _, _, Uri.Path( "/sockets" ), _, _ ),
      headers, _, _ ) ⇒
      sender ! Sockets.UpgradeServer(
        Sockets.acceptAllFunction( req ),
        publisherActor( genNewId() )
      )

Any suggestions? It looks like the warning message is from here: https://github.com/spray/spray/blob/master/spray-io/src/main/scala/spray/io/ConnectionHandler.scala#L31

I can't see how the client can receive Acks.

I'm using this library as a client. The TcpPipelineHandler actor (in Akka) holds my specific pipeline and brokers between my app and the socket (which is a spray web socket). My pipeline has a backpressure buffer (from Akka) that requires Acks to be propagated back to the pipeline from the socket.

I have some code that bridges between a "standard looking" IO client and the web socket that looks something like this:

case Tcp.Write(bytes, ack) =>
    websocket ! Frame(...)

I need to ack back to the sender once the socket has completed its write, in order to ensure that my backpressure buffer does the right thing. However, I don't know how to get an Ack back from the web socket.

For now I'm acking immediately but this, of course, isn't going to work because it's going to signal to the sender that he's good to send another write immediately, which is causing failures:

Failing write command because there is already another write in progress

Am I missing something or is this just not currently possible?

README.md describes how to "Close the Connection" but SocketServer.Close and SocketServer.Closed seem to incorrect. (Tcp.Close and Tcp.Closed?)

A WebSockets server actor listens to certain events related to the connection. Akka's TCP-based IO extension allows us to retrieve the client IP address via the Tcp.Connected event. The sender of this event is supposed to represent the client connection itself.

When using SprayWebSockets, the actor which sends this event no longer represents the client connection. Instead, there is another event which comes later, Sockets.Upgraded, whose sender represents the client connection. The problem is that at this point, there is no way to retrieve the client IP address.

It's easy to verify this behavior with the following program:

import akka.actor._
import akka.io.Tcp, Tcp._
import spray.http.HttpRequest
import spray.can.server.websockets.Sockets
import spray.can.Http

object Simple extends App {

  implicit val system = ActorSystem()

  val act = system.actorOf(Props[Server])
  akka.io.IO(Sockets) ! Http.Bind(act, interface = "localhost", port = 8080)

  class Server extends Actor with ActorLogging {
    def receive = {
      case x: Connected =>
        log.info("Connected: {}", sender)
        sender ! Register(self)
      case x: ConnectionClosed =>
        log.info("Disconnected: {}", sender)
      case Sockets.Upgraded =>
        log.info("Upgraded: {}", sender)
      case req: HttpRequest =>
        sender ! Sockets.UpgradeServer(
          Sockets.acceptAllFunction(req), self)
      case x =>
        log.info("Received data from {}: {}", sender, x.toString)
    }
  }
}

In this example, the actor which sends Tcp.Connected is not the same actor which sends all other events. It should be the same actor...

My application needs to provide both Rest API and websockets support. So for that I'm binding to 2 ports at the same time (Spray 1.2):

IO(Http) ! Http.Bind(serviceActor, "0.0.0.0", port = 38080)
IO(Sockets) ! Http.Bind(websocketActor, "0.0.0.0", port = 38081)

but I get an exception that actor name IO-HTTP is already taken.

Exception in thread "main" akka.actor.InvalidActorNameException: actor name [IO-HTTP] is not unique!
at akka.actor.dungeon.ChildrenContainer$NormalChildrenContainer.reserve(ChildrenContainer.scala:130)
at akka.actor.dungeon.Children$class.reserveChild(Children.scala:77)
at akka.actor.ActorCell.reserveChild(ActorCell.scala:338)
at akka.actor.dungeon.Children$class.makeChild(Children.scala:186)
at akka.actor.dungeon.Children$class.attachChild(Children.scala:42)
at akka.actor.ActorCell.attachChild(ActorCell.scala:338)
at akka.actor.ActorSystemImpl.actorOf(ActorSystem.scala:518)
at spray.can.HttpExt.(Http.scala:152)
at spray.can.server.websockets.SocketExt.(Sockets.scala:110)
at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
at sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:57)
at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
at java.lang.reflect.Constructor.newInstance(Constructor.java:525)
at akka.actor.ReflectiveDynamicAccess$$anonfun$createInstanceFor$2.apply(DynamicAccess.scala:78)
at scala.util.Try$.apply(Try.scala:161)
at akka.actor.ReflectiveDynamicAccess.createInstanceFor(DynamicAccess.scala:73)
at akka.actor.ExtensionKey.createExtension(Extension.scala:153)
at akka.actor.ActorSystemImpl.registerExtension(ActorSystem.scala:654)
at akka.actor.ExtensionId$class.apply(Extension.scala:79)
at akka.actor.ExtensionKey.apply(Extension.scala:149)
at akka.io.IO$.apply(IO.scala:36)
at my.Boot.startup(Boot.scala:29)

It works fine if I try to bind only to one server, whether it's HTTP or Sockets and works fine if I bind using different ActorSystems:

IO(Http) ! Http.Bind(serviceActor, MaierConfig.rest.host, port = MaierConfig.rest.port)
IO(Sockets)(ActorSystem()) ! Http.Bind(websocketActor, MaierConfig.websocket.host, port = MaierConfig.websocket.port)

It looks like the problem is in class spray.can.server.websockets.SocketExt, which extends HttpExt which creates a manager actor and sets a hardcoded name to it - "IO-HTTP". Sockets class then cannot initialize because super initialization is called and name IO-HTTP is not unique. Is there a way to avoid this?