diff --git a/README.md b/README.md index 02f13f1..94e6be0 100644 --- a/README.md +++ b/README.md @@ -8,6 +8,12 @@ Getting started This is just an SBT application; you don't need Play 2. -Run `sbt run`, and browse to http://localhost:9000/ +Run `sbt run` + +Open a browser and add url: http://localhost:9000/websockets + +In the websocket address field enter: ws://localhost:9000/websockets/cpu-info + +Repeat this in other tabs, connect and disconnect and see in the logs what happens. As you might have seen I do not acquire real cpu info, it's just a counter. But the idea should be clear. + -For the WebSocket test-screen, browse to http://localhost:9000/ diff --git a/app/controllers/CpuInfoDistributer.scala b/app/controllers/CpuInfoDistributer.scala new file mode 100644 index 0000000..ccffd7a --- /dev/null +++ b/app/controllers/CpuInfoDistributer.scala @@ -0,0 +1,77 @@ +package controllers; + +import play.api._ +import play.api.mvc._ +import play.api.libs.iteratee.{ Iteratee, Enumerator } +import play.api.libs.concurrent.Promise +import akka.util.duration._ +import akka.util._ +import akka.actor.{ ActorSystem, Actor } +import play.api.libs.iteratee.Enumerator._ +import akka.actor.Props +import akka.actor.Scheduler +import java.util.concurrent.atomic.AtomicInteger + +object CpuInfoWebSocket extends Controller { + + val system = ActorSystem("MySystem") + val cpuLoadActor = system.actorOf(Props[CpuInfoDistributionActor]) + val pusheeCounter = new AtomicInteger + system.scheduler.schedule(0 seconds, 3 seconds, cpuLoadActor, DistributeMessage) + + /** + * Websocket that retrieves a pushee and send it to + * a Actor + */ + def cpuLoadSender() = WebSocket.using[String] { request => + val in = Iteratee.ignore[String] + val count = pusheeCounter.incrementAndGet() + val cpuInfoEnumerator = Enumerator.pushee[String]( + onStart = pushee => cpuLoadActor ! AddMessage(count, pushee), + onComplete = cpuLoadActor ! RemoveMessage(count), + onError = (a, b) => println("error " + a + " " + b)) + (in, cpuInfoEnumerator) + } +} + +/** + * Actor messages + */ +sealed trait CpuMessages +case class AddMessage(id: Int, pushee: Pushee[String]) extends CpuMessages +case class DistributeMessage() extends CpuMessages +case class RemoveMessage(id: Int) extends CpuMessages + +/** + * Distribution actor + */ +class CpuInfoDistributionActor extends Actor { + var pushees: Map[Int, Pushee[String]] = Map.empty + var counter: Int = 0 + + def receive = { + case AddMessage(id, pushee) => { + println("Added pushee with id " + id) + pushees = pushees + (id -> pushee) + } + case DistributeMessage => { + if (!pushees.isEmpty) { + println("Distribute cpu average: " + counter) + counter += 1 + } + pushees.foreach { + case (id, p) => { + println("Send to pushee: " + id) + p.push(counter.toString) + } + } + } + case RemoveMessage(id) => { + println("Remove pushee with id " + id) + pushees = pushees - id + + } + + } + +} diff --git a/conf/routes b/conf/routes index 72deb64..2c9bdf4 100644 --- a/conf/routes +++ b/conf/routes @@ -7,6 +7,7 @@ GET /templates/index controllers.Templates.index GET /responses/simple-without-header controllers.Responses.simpleWithoutHeader() + GET /responses/simple-feeding controllers.Responses.simpleFeeding() GET /responses/chunked-result controllers.Responses.chunkedResult() GET /responses/imperative-chunked controllers.Responses.imperativeChunked() @@ -18,6 +19,7 @@ GET /websockets/logging controllers.WebSockets.logging() GET /websockets/echo controllers.WebSockets.echo() GET /websockets/counter controllers.WebSockets.counter() GET /websockets/echo-and-counter controllers.WebSockets.echoAndCounter() +GET /websockets/cpu-info controllers.CpuInfoWebSocket.cpuLoadSender() GET /twitter/stream controllers.Twitter.stream(keywords)