Reactive Push, Composition и UI на примере Reactive Stocks - 2

UI реактивного приложения

UI приложения Reactive Stocks фактически состоит из одной страницы и устроен следующим образом. Схематично страница с котировками выглядит так:

Здесь верхняя панель (в приложении используется фреймворк Bootstrap, в котором эта панель называется “navbar”, т.е. навигационный заголовок) содержит только форму добавления новой котировки, которая состоит из двух контролов:

Остальная же часть страницы - один сплошной <div> stocks. При начальной загрузке кнопке “Add stock” назначается обработчик

$("#addsymbolform").submit (event) ->
....
  ws.send(JSON.stringify({symbol: $("#addsymboltext").val()}))
....

который вызовет веб-сервис/ws, т.е. фактически метод controllers.Application.ws, и будет передано JSON-сообщение вида {symbol:<SYMBOL>}, где <SYMBOL> - содержимое текстового контрола, т.е. название котировки. В ответ на запрос пользователя об отслеживании котировки, присылаемый через входной канал WebSocket.In, WebSocket из контроллера Application генерирует сообщение WatchStock, порождается цепочка сообщений, которая заканчивается JSON-сообщением stockhistory, отправляемым UserActor‘ом в выходной канал WebSocket‘а, и который содержит символ котировки и массив из 50 последних значений котировки.

В ответ на это сообщение, вызовется создание и заполнение 50-ю предыдущими значениями нового графика с котировками. Схематично, реализация отображения графиков построено следующим образом (stocks, flip-contaner и остальные заголовки есть названия соотвествующих <div>ов):

Т.е. каждый график - это фактически <div> chart-holder, содержащий еще 2 <div>‘а - собственно chart и details-holder. В свою очередь, <div> chart-holder вложен в <div> flipper, а последний - <div> flip-container. flip-container нужен для следующей функциональности. В приложении Reactive Stocks, если пользователь нажимает на график котировок, то он переворачивается, и далее должно быть отображено т.н. “ожидание” для данной котировки. Это самое ожидание образуется следующим образом - сначала делается поиск в твитере с упоминанием символа этой котировки, а затем делается запрос к специальному сервису, определяющему “настроение” переданного текста, на основании которого показывается картинка с ожиданием, т.е. рекомендация, что нужно делать - “buy”, “sell” или “hold”.

Для того, чтобы по клику переворачивалась картинка и отображалась “рекомендация”, навешивается обработчик handleFlip:

populateStockHistory = (message) ->
....
  flipContainer = $("<div>").addClass("flip-container").append(flipper).click (event) ->
    handleFlip($(this))
....

Как уже отмечалось, есть две функции, обновляющие график, и обе отрабатывают в ответ на JSON-сообщения из входного (с точки зрения клиента) канала WebSocket‘а:

• populateStockHistory - первоначальная, в ответ на JSON-сообщение "stockhistory";
• updateStockChart - вызывается каждый раз в ответ на JSON-сообщение "stockupdate";

Для отрисовки используется библиотека Flot. Использовать ее очень просто - фактически, все, что нужно сделать - это вызвать функцию plot:

$("#placeholder").plot(data, options)

где placeholder - это существующий DOM элемент, например, <div>, data - или массив массивов (пар) координат, или спец. объекты, ну и options - очевидно, что это такое - объект со всякими настройками. Это, собственно, и делается, например, в функции populateStockHistory:

chart.plot([getChartArray(message.history)], getChartOptions(message.history))

Задача функции updateStockChart же лишь добавить одну координату в ответ на "stockupdate", поэтому она извлекает данные из контрола

plot = $("#" + message.symbol).data("plot")

добавляет к ним еще одну координату, вызывает plot.setData и перерисовывает график.

Отрисовка ожидания

Как мы уже выяснили, при нажатии на график он должен перевернутся, и нам должна отобразиться одна из картинок - buy, sell или hold, и все это делается в методе handleFlip Переворот осуществляется с помощью назначения спец. класса и CSS-трансформации, т.е. назначается класс flipped:

container.addClass("flipped")

а этот класс описан в main.less с использованием трансформации, в данном случае - зеркального разворота

&.flipped .flipper {
  .transform(180deg);
}

Затем делается ajax-запрос к эндпойнту /sentiment/<SYMBOL>, где <SYMBOL>, как уже понятно - символ котировки (которой хранится в атрибуте data-content <div>а flipper и соответственно извлекается оттуда). И при успехе запроса, анализируется в JSON-ответе анализируется поле label и в зависимости от возвращаемого значения - pos, neg или neutral - показывается соответствующее ожидание: buy, sell или hold . Для этого в <div> details-holder устанавливается соответствующий текст и иконка, например:

detailsHolder.append($("<h4>").text("The tweets say BUY!"))
detailsHolder.append($("<img>").attr("src", "/assets/images/buy.png"))

Но как именно определяется, какое именно ожидание показать?

Реактивный запрос и реактивная композиция

Когда веб-сервер получает запрос, как правило, выделяется поток (например, из пула) для его обработки. В классической модели поток выделяется на все время обработки запроса вплоть до генерации ответа, даже в том случае, если обработка запроса потребует ожидания от другого, внешнего ресурса, например, веб-сервиса - и это может занять сравнительно большой промежуток времени. Реактивный же запрос с точки зрения клиента выглядит точно также, но внутри реализован таким образом, что реально 1 поток не блокируется на все время ожидания ответа от внешнего ресурса. Это означает, что если мы находимся в режиме ожидания ответа, т.е. поток активно не используется, он может использоваться для каких-нибудь других целей. В приложении Reactive Stocks эндпойнт определения “настроения” котировки реализован именно с помощью реактивных запросов. Эндпойнт /sentiment/<SYMBOL> приводит к вызову в контроллере StockSentiment (см. conf/routes):

GET /sentiment/:symbol controllers.StockSentiment.get(symbol)

Посмотрим на сигнатуру этого метода:

def get(symbol: String): Action[AnyContent] = Action.async {

Блок async говорит нам, что будет возвращен Future[Result], т.е. обработка запроса происходит асинхронно. Внутри блока происходит приблизительно следующее:

1. Происходит вызов к прокси поиска на Твитере, находится некоторое количество твитов с упоминанием котировки
2. Для каждого твита определяется его “настроение”
3. Вычисляется значения “настроения”/”ожидания” для каждого твита
4. Вычисляются средние значения neg, neutral и pos для по вероятностям из предыдущего шага, и по ним определяется общее “настроение” котировки

Более детально: происходит вызов к прокси поиска на Твитере. Фактически это небольшое Scala-приложение twitter-search-proxy, клиент, который делает запросы к Twitter, кэширует их и обрабатывает отказы (которые случаются примерно в 10% случаев). В данном случае оно развернуто в облачном сервисе Heroku. Возвращаемый JSON имеет примерно следующий вид:

{
  "statuses": [
    {
      "metadata": {
        "iso_language_code": "en",
        "result_type": "recent"
      },
      "created_at": "Sun Jun 14 08:44:05 +0000 2015",
      "id": 610004771844562944,
      "text": "$BABA Cloud Services Will Be a Billion-Dollar Business By 2018... http://t.co/GX1v476Liy via @TheStreet #market #retail $GOOG $IBM $MSFT",
    }
  ],
  "search_metadata": {
    "completed_in": 0.054,
    "next_results": "?max_id=609998420472897535&q=GOOG&include_entities=1",
    "query": "GOOG",
    "count": 15
  }
}

где statuses - фактически отдельные найденные твиты. С помощью клиентской библиотеки веб-сервисов Play делается вызов к прокси поиска на Твитере:

WS.url(Play.current.configuration.getString("tweet.url").get.format(symbol)).get.withFilter { response =>
  response.status == OK
}

Для каждого твита определяется его “настроение”. Это делается путем вызова к еще одному веб-сервису. На http://text-processing.com/ есть ряд сервисов, связанных с обработкой текста, на интересует конкретно определение “настроения” - http://text-processing.com/docs/index.html. Вызов делается приблизительно аналогичным образом:

WS.url(Play.current.configuration.getString("sentiment.url").get) post Map("text" -> Seq(text))

Затем подсчитывается средние значения neg, neutral и pos по вероятностям для “настроений”/”ожиданий”, и далее используется простой алгоритм для вычисления общего результата (который помещается в поле label). Если вероятность “нейтрального” “настроения” более 0.5, то результат будет “нейтральным”. В ином случае, поскольку neg и pos в сумме дают 1, то берется тот, который из них больше.

if (neutral > 0.5)
  "neutral"
else if (neg > pos)
  "neg"
else
  "pos"

Все эти действия в сумме совершаются неблокирующим и асинхронным образом, т.е. являются реактивными, в том числе и запрос из браузера (поскольку ajax-запрос является тоже реактивным). Таким образом, вся цепочка запросов является реактивной, что является реактивной композицией:

Action.async {
  for {
    tweets <- getTweets(symbol) 
    futureSentiments = loadSentimentFromTweets(tweets.json)
    sentiments <- Future.sequence(futureSentiments) 
  } yield Ok(sentimentJson(sentiments))
}