Initial commit with reactive-stocks

.gitignore

@@ -0,0 +1,17 @@
+logs
+project/project
+project/target
+target
+tmp
+.history
+dist
+/.idea
+/*.iml
+/out
+/.idea_modules
+/.classpath
+/.project
+/RUNNING_PID
+/.settings
+/project/*-shim.sbt
+/activator-sbt-atmos-akka-shim.sbt

LICENSE

@@ -0,0 +1,13 @@
+Copyright 2013 Typesafe, Inc.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.

activator.properties

@@ -0,0 +1,4 @@
+name=reactive-stocks
+title=Reactive Stocks
+description=The Reactive Stocks application uses Java, Scala, Play Framework, and Akka to illustrate a reactive app.  The tutorial in this example will teach you the reactive basics including Reactive Composition and Reactive Push.
+tags=Sample,java,scala,playframework,akka,reactive

app/actors/StockActor.scala

@@ -0,0 +1,82 @@
+package actors
+
+import akka.actor.{Props, ActorRef, Actor}
+import utils.{StockQuote, FakeStockQuote}
+import java.util.Random
+import scala.collection.immutable.{HashSet, Queue}
+import scala.collection.JavaConverters._
+import scala.concurrent.duration._
+import scala.concurrent.ExecutionContext.Implicits.global
+import play.libs.Akka
+
+/**
+ * There is one StockActor per stock symbol.  The StockActor maintains a list of users watching the stock and the stock
+ * values.  Each StockActor updates a rolling dataset of randomly generated stock values.
+ */
+
+class StockActor(symbol: String) extends Actor {
+
+  lazy val stockQuote: StockQuote = new FakeStockQuote
+
+  protected[this] var watchers: HashSet[ActorRef] = HashSet.empty[ActorRef]
+
+  // A random data set which uses stockQuote.newPrice to get each data point
+  var stockHistory: Queue[java.lang.Double] = {
+    lazy val initialPrices: Stream[java.lang.Double] = (new Random().nextDouble * 800) #:: initialPrices.map(previous => stockQuote.newPrice(previous))
+    initialPrices.take(50).to[Queue]
+  }
+
+  // Fetch the latest stock value every 75ms
+  val stockTick = context.system.scheduler.schedule(Duration.Zero, 75.millis, self, FetchLatest)
+
+  def receive = {
+    case FetchLatest =>
+      // add a new stock price to the history and drop the oldest
+      val newPrice = stockQuote.newPrice(stockHistory.last.doubleValue())
+      stockHistory = stockHistory.drop(1) :+ newPrice
+      // notify watchers
+      watchers.foreach(_ ! StockUpdate(symbol, newPrice))
+    case WatchStock(_) =>
+      // send the stock history to the user
+      sender ! StockHistory(symbol, stockHistory.asJava)
+      // add the watcher to the list
+      watchers = watchers + sender
+    case UnwatchStock(_) =>
+      watchers = watchers - sender
+      if (watchers.size == 0) {
+        stockTick.cancel()
+        context.stop(self)
+      }
+  }
+}
+
+class StocksActor extends Actor {
+  def receive = {
+    case watchStock @ WatchStock(symbol) =>
+      // get or create the StockActor for the symbol and forward this message
+      context.child(symbol).getOrElse {
+        context.actorOf(Props(new StockActor(symbol)), symbol)
+      } forward watchStock
+    case unwatchStock @ UnwatchStock(Some(symbol)) =>
+      // if there is a StockActor for the symbol forward this message
+      context.child(symbol).foreach(_.forward(unwatchStock))
+    case unwatchStock @ UnwatchStock(None) =>
+      // if no symbol is specified, forward to everyone
+      context.children.foreach(_.forward(unwatchStock))
+  }
+}
+
+object StocksActor {
+  lazy val stocksActor: ActorRef = Akka.system.actorOf(Props(classOf[StocksActor]))
+}
+
+
+case object FetchLatest
+
+case class StockUpdate(symbol: String, price: Number)
+
+case class StockHistory(symbol: String, history: java.util.List[java.lang.Double])
+
+case class WatchStock(symbol: String)
+
+case class UnwatchStock(symbol: Option[String])

app/actors/UserActor.java

@@ -0,0 +1,59 @@
+package actors;
+
+import akka.actor.UntypedActor;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.node.ArrayNode;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import play.Play;
+import play.libs.Json;
+import play.mvc.WebSocket;
+
+import java.util.List;
+
+/**
+ * The broker between the WebSocket and the StockActor(s).  The UserActor holds the connection and sends serialized
+ * JSON data to the client.
+ */
+
+public class UserActor extends UntypedActor {
+
+    private final WebSocket.Out<JsonNode> out;
+
+    public UserActor(WebSocket.Out<JsonNode> out) {
+        this.out = out;
+
+        // watch the default stocks
+        List<String> defaultStocks = Play.application().configuration().getStringList("default.stocks");
+
+        for (String stockSymbol : defaultStocks) {
+            StocksActor.stocksActor().tell(new WatchStock(stockSymbol), getSelf());
+        }
+    }
+
+    public void onReceive(Object message) {
+        if (message instanceof StockUpdate) {
+            // push the stock to the client
+            StockUpdate stockUpdate = (StockUpdate)message;
+            ObjectNode stockUpdateMessage = Json.newObject();
+            stockUpdateMessage.put("type", "stockupdate");
+            stockUpdateMessage.put("symbol", stockUpdate.symbol());
+            stockUpdateMessage.put("price", stockUpdate.price().doubleValue());
+            out.write(stockUpdateMessage);
+        }
+        else if (message instanceof StockHistory) {
+            // push the history to the client
+            StockHistory stockHistory = (StockHistory)message;
+
+            ObjectNode stockUpdateMessage = Json.newObject();
+            stockUpdateMessage.put("type", "stockhistory");
+            stockUpdateMessage.put("symbol", stockHistory.symbol());
+
+            ArrayNode historyJson = stockUpdateMessage.putArray("history");
+            for (Object price : stockHistory.history()) {
+                historyJson.add(((Number)price).doubleValue());
+            }
+
+            out.write(stockUpdateMessage);
+        }
+    }
+}

app/assets/javascripts/index.coffee

@@ -0,0 +1,100 @@
+$ ->
+  ws = new WebSocket $("body").data("ws-url")
+  ws.onmessage = (event) ->
+    message = JSON.parse event.data
+    switch message.type
+      when "stockhistory"
+        populateStockHistory(message)
+      when "stockupdate"
+        updateStockChart(message)
+      else
+        console.log(message)
+
+  $("#addsymbolform").submit (event) ->
+    event.preventDefault()
+    # send the message to watch the stock
+    ws.send(JSON.stringify({symbol: $("#addsymboltext").val()}))
+    # reset the form
+    $("#addsymboltext").val("")
+
+getPricesFromArray = (data) ->
+  (v[1] for v in data)
+
+getChartArray = (data) ->
+  ([i, v] for v, i in data)
+
+getChartOptions = (data) ->
+  series:
+    shadowSize: 0
+  yaxis:
+    min: getAxisMin(data)
+    max: getAxisMax(data)
+  xaxis:
+    show: false
+
+getAxisMin = (data) ->
+  Math.min.apply(Math, data) * 0.9
+
+getAxisMax = (data) ->
+  Math.max.apply(Math, data) * 1.1
+
+populateStockHistory = (message) ->
+  chart = $("<div>").addClass("chart").prop("id", message.symbol)
+  chartHolder = $("<div>").addClass("chart-holder").append(chart)
+  chartHolder.append($("<p>").text("values are simulated"))
+  detailsHolder = $("<div>").addClass("details-holder")
+  flipper = $("<div>").addClass("flipper").append(chartHolder).append(detailsHolder).attr("data-content", message.symbol)
+  flipContainer = $("<div>").addClass("flip-container").append(flipper).click (event) ->
+    handleFlip($(this))
+  $("#stocks").prepend(flipContainer)
+  plot = chart.plot([getChartArray(message.history)], getChartOptions(message.history)).data("plot")
+
+updateStockChart = (message) ->
+  if ($("#" + message.symbol).size() > 0)
+    plot = $("#" + message.symbol).data("plot")
+    data = getPricesFromArray(plot.getData()[0].data)
+    data.shift()
+    data.push(message.price)
+    plot.setData([getChartArray(data)])
+    # update the yaxes if either the min or max is now out of the acceptable range
+    yaxes = plot.getOptions().yaxes[0]
+    if ((getAxisMin(data) < yaxes.min) || (getAxisMax(data) > yaxes.max))
+      # reseting yaxes
+      yaxes.min = getAxisMin(data)
+      yaxes.max = getAxisMax(data)
+      plot.setupGrid()
+    # redraw the chart
+    plot.draw()
+
+handleFlip = (container) ->
+  if (container.hasClass("flipped"))
+    container.removeClass("flipped")
+    container.find(".details-holder").empty()
+  else
+    container.addClass("flipped")
+    # fetch stock details and tweet
+    $.ajax
+      url: "/sentiment/" + container.children(".flipper").attr("data-content")
+      dataType: "json"
+      context: container
+      success: (data) ->
+        detailsHolder = $(this).find(".details-holder")
+        detailsHolder.empty()
+        switch data.label
+          when "pos"
+            detailsHolder.append($("<h4>").text("The tweets say BUY!"))
+            detailsHolder.append($("<img>").attr("src", "/assets/images/buy.png"))
+          when "neg"
+            detailsHolder.append($("<h4>").text("The tweets say SELL!"))
+            detailsHolder.append($("<img>").attr("src", "/assets/images/sell.png"))
+          else
+            detailsHolder.append($("<h4>").text("The tweets say HOLD!"))
+            detailsHolder.append($("<img>").attr("src", "/assets/images/hold.png"))
+      error: (jqXHR, textStatus, error) ->
+        detailsHolder = $(this).find(".details-holder")
+        detailsHolder.empty()
+        detailsHolder.append($("<h2>").text("Error: " + JSON.parse(jqXHR.responseText).error))
+    # display loading info
+    detailsHolder = container.find(".details-holder")
+    detailsHolder.append($("<h4>").text("Determing whether you should buy or sell based on the sentiment of recent tweets..."))
+    detailsHolder.append($("<div>").addClass("progress progress-striped active").append($("<div>").addClass("bar").css("width", "100%")))