ReactiveMongo с Akka, Scala и веб-сокетами

Я искал простой сервер веб-сокетов для одного из моих проектов, чтобы проверить некоторые вещи с реактивным монго . Однако, оглядываясь по сторонам, я не мог найти простую базовую реализацию, не включив полную структуру. Наконец я наткнулся на один из проектов активаторов Typesage: http://typesafe.com/activator/template/akka-spray-websocket . Несмотря на то, что название подразумевает, что спрей необходим, на самом деле он использует материал websocket отсюда: https://github.com/TooTallNate/Java-WebSocket , который обеспечивает очень простую в использовании базовую реализацию websocket.

Поэтому в этой статье я покажу вам, как вы можете настроить очень простой сервер веб-сокетов (не требуя дополнительных фреймворков) вместе с Akka и ReactiveMongo. На следующих скриншотах показано, к чему мы стремимся:

На этом скриншоте вы видите простой клиент веб-сокета, который общается с нашим сервером. Наш сервер имеет следующие функциональные возможности:

Все, что отправляет клиент, возвращается назад.
Любые входные данные, добавленные в конкретную (ограниченную) коллекцию в mongoDB , автоматически передаются всем слушателям.

Вы можете вырезать и вставить весь код из этой статьи, но, вероятно, проще получить код из git. Вы можете найти его в github здесь: https://github.com/josdirksen/smartjava/tree/master/ws-akka

Начиная

Первое, что нам нужно сделать, это настроить наше рабочее пространство, поэтому давайте начнем с рассмотрения конфигурации sbt:

organization  := "org.smartjava"
  
version       := "0.1"
  
scalaVersion  := "2.11.2"
  
scalacOptions := Seq("-unchecked", "-deprecation", "-encoding", "utf8")
  
libraryDependencies ++= {
  val akkaV = "2.3.6"
  Seq(
    "com.typesafe.akka"   %%  "akka-actor"    % akkaV,
    "org.java-websocket" % "Java-WebSocket" % "1.3.1-SNAPSHOT",
    "org.reactivemongo" %% "reactivemongo" % "0.10.5.0.akka23"
  )
}
  
resolvers ++= Seq("Code Envy" at "http://codenvycorp.com/repository/"
  ,"Typesafe" at "http://repo.typesafe.com/typesafe/releases/")

Здесь нет ничего особенного, мы просто указываем наши зависимости и добавляем некоторые средства распознавания, чтобы sbt знал, откуда извлекать зависимости. Прежде чем мы рассмотрим код, давайте сначала посмотрим на структуру каталогов и файл нашего проекта:

├── build.sbt
└── src
    └── main
        ├── resources
        │   ├── application.conf
        │   └── log4j2.xml
        └── scala
            ├── Boot.scala
            ├── DB.scala
            ├── WSActor.scala
            └── WSServer.scala

В каталоге src / main / resources мы храним наши файлы конфигурации, а в src / main / scala мы храним все наши файлы scala. Давайте начнем с просмотра файлов конфигурации. Для этого проекта мы используем два:

Файл Application.conf содержит конфигурацию нашего проекта и выглядит следующим образом:

akka {
  loglevel = "DEBUG"
}
  
mongo {
  db = "scala"
  collection = "rmongo"
  location = "localhost"
}
  
ws-server {
  port = 9999
}

Как видите, мы просто определяем уровень журнала, как использовать mongo и на каком порту мы хотим, чтобы наш сервер веб-сокетов слушал. И нам также нужен файл log4j2.xml, так как библиотека responsetivemongo использует его для ведения журнала:

<?xml version="1.0" encoding="UTF-8"?>
<Configuration status="INFO">
    <Appenders>
        <Console name="Console" target="SYSTEM_OUT">
            <PatternLayout pattern="%d{HH:mm:ss.SSS} [%t] %-5level %logger{36} - %msg%n"/>
        </Console>
    </Appenders>
    <Loggers>
        <Root level="INFO">
            <AppenderRef ref="Console"/>
        </Root>
    </Loggers>
</Configuration>

Итак, со скучными вещами в стороне, давайте посмотрим на файлы scala.

Запуск сервера websocket и регистрация путей

Файл Boot.scala выглядит так:

package org.smartjava
  
import akka.actor.{Props, ActorSystem}
  
/**
 * This class launches the system.
 */
object Boot extends App {
  // create the actor system
  implicit lazy val system = ActorSystem("ws-system")
  // setup the mongoreactive connection
  implicit lazy val db = new DB(Configuration.location, Configuration.dbname);
  
  // we'll use a simple actor which echo's everything it finds back to the client.
  val echo = system.actorOf(EchoActor.props(db, Configuration.collection), "echo")
  
  // define the websocket routing and start a websocket listener
  private val wsServer = new WSServer(Configuration.port)
  wsServer.forResource("/echo", Some(echo))
  wsServer.start
  
  // make sure the actor system and the websocket server are shutdown when the client is
  // shutdown
  sys.addShutdownHook({system.shutdown;wsServer.stop})
}
  
// load configuration from external file
object Configuration {
  import com.typesafe.config.ConfigFactory
  
  private val config = ConfigFactory.load
  config.checkValid(ConfigFactory.defaultReference)
  
  val port = config.getInt("ws-server.port")
  val dbname = config.getString("mongo.db")
  val collection = config.getString("mongo.collection")
  val location = config.getString("mongo.location")
}

В этом исходном файле мы видим два объекта. Объект Configuration позволяет нам легко получить доступ к элементам конфигурации из файла application.conf, и объект Boot запустит наш сервер. Комментарии в коде должны в значительной степени объяснить, что происходит, но позвольте мне указать основные моменты:

Мы создаем систему актеров Akka и соединение с нашим экземпляром mongoDB.
Мы определяем актера, которого мы можем зарегистрировать по определенному пути веб-сокета.
Затем мы создаем и запускаем websocketserver и регистрируем путь к только что созданному действующему субъекту.
Наконец мы регистрируем крюк отключения, чтобы очистить все.

Вот и все. Теперь давайте посмотрим на интересную часть кода. Далее следует файл WSServer.scala.

Настройка сервера веб-сокетов

В файле WSServer.scala мы определяем сервер веб-сокетов.

package org.smartjava
  
import akka.actor.{ActorSystem, ActorRef}
import java.net.InetSocketAddress
import org.java_websocket.WebSocket
import org.java_websocket.framing.CloseFrame
import org.java_websocket.handshake.ClientHandshake
import org.java_websocket.server.WebSocketServer
import scala.collection.mutable.Map
import akka.event.Logging
  
/**
 * The WSserver companion objects defines a number of distinct messages sendable by this component
 */
object WSServer {
  sealed trait WSMessage
  case class Message(ws : WebSocket, msg : String) extends WSMessage
  case class Open(ws : WebSocket, hs : ClientHandshake) extends WSMessage
  case class Close(ws : WebSocket, code : Int, reason : String, external : Boolean) 
                                                                                                         extends WSMessage
  case class Error(ws : WebSocket, ex : Exception) extends WSMessage
}
  
/**
 * Create a websocket server that listens on a specific address.
 *
 * @param port
 */
class WSServer(val port : Int)(implicit system : ActorSystem, db: DB ) 
                             extends WebSocketServer(new InetSocketAddress(port)) {
  
  // maps the path to a specific actor.
  private val reactors = Map[String, ActorRef]()
  // setup some logging based on the implicit passed in actorsystem
  private val log = Logging.getLogger(system, this);
  
  // Call this function to bind an actor to a specific path. All incoming
  // connections to a specific path will be routed to that specific actor.
  final def forResource(descriptor : String, reactor : Option[ActorRef]) {
    log.debug("Registring actor:" + reactor + " to " + descriptor);
    reactor match {
      case Some(actor) => reactors += ((descriptor, actor))
      case None => reactors -= descriptor
    }
  }
  
  // onMessage is called when a websocket message is recieved.
  // in this method we check whether we can find a listening
  // actor and forward the call to that.
  final override def onMessage(ws : WebSocket, msg : String) {
  
    if (null != ws) {
      reactors.get(ws.getResourceDescriptor) match {
        case Some(actor) => actor ! WSServer.Message(ws, msg)
        case None => ws.close(CloseFrame.REFUSE)
      }
    }
  }
  
  final override def onOpen(ws : WebSocket, hs : ClientHandshake) {
    log.debug("OnOpen called {} :: {}", ws, hs);
    if (null != ws) {
      reactors.get(ws.getResourceDescriptor) match {
        case Some(actor) => actor ! WSServer.Open(ws, hs)
        case None => ws.close(CloseFrame.REFUSE)
      }
    }
  }
  
  final override def onClose(ws : WebSocket, code : Int, reason : String, external : Boolean) {
    log.debug("Close called {} :: {} :: {} :: {}", ws, code, reason, external);
    if (null != ws) {
      reactors.get(ws.getResourceDescriptor) match {
        case Some(actor) => actor ! WSServer.Close(ws, code, reason, external)
        case None => ws.close(CloseFrame.REFUSE)
      }
    }
  }
  final override def onError(ws : WebSocket, ex : Exception) {
    log.debug("onError called {} :: {}", ws, ex);
    if (null != ws) {
      reactors.get(ws.getResourceDescriptor) match {
        case Some(actor) => actor ! WSServer.Error(ws, ex)
        case None => ws.close(CloseFrame.REFUSE)
      }
    }
  }
}

Большой исходный файл, но не сложный для понимания. Позвольте мне объяснить основные понятия:

Сначала мы определим количество сообщений как классы дел. Это сообщения, которые мы отправили нашим актерам. Они отражают сообщения, которые наш сервер веб-сокетов может получать от клиента.
Сам WSServer происходит от WebSocketServer, предоставляемого библиотекой org.java_websocket.
WSServer определяет одну дополнительную функцию, которая называется forResource. С помощью этой функции мы определяем, какого актера вызывать при получении сообщения на нашем сервере веб-сокетов.
и, наконец, мы переопределяем различные методы *, которые вызываются, когда на нашем сервере веб-сокетов происходит определенное событие.

Теперь давайте посмотрим на функциональность эха

Акко эхо актер

Актер-эхо имеет две роли в этом сценарии. Во-первых, он предоставляет возможность отвечать на входящие сообщения, отвечая тем же сообщением. Кроме того, он также создает дочерний актер (с именем ListenActor), который обрабатывает документы, полученные от mongoDB.

object EchoActor {
  
  // Messages send specifically by this actor to another instance of this actor.
  sealed trait EchoMessage
  
  case class Unregister(ws : WebSocket) extends EchoMessage
  case class Listen() extends EchoMessage;
  case class StopListening() extends EchoMessage
  
  def props(db: DB): Props = Props(new EchoActor(db))
}
  
/**
 * Actor that handles the websocket request
 */
class EchoActor(db: DB) extends Actor with ActorLogging {
  import EchoActor._
  
  val clients = mutable.ListBuffer[WebSocket]()
  val socketActorMapping = mutable.Map[WebSocket, ActorRef]()
  
  override def receive = {
  
    // receive the open request
    case Open(ws, hs) => {
      log.debug("Received open request. Start listening for ", ws)
      clients += ws
  
      // create the child actor that handles the db listening
      val targetActor = context.actorOf(ListenActor.props(ws, db));
  
      socketActorMapping(ws) = targetActor;
      targetActor ! Listen
    }
  
    // recieve the close request
    case Close(ws, code, reason, ext) => {
      log.debug("Received close request. Unregisting actor for url {}", ws.getResourceDescriptor)
  
      // send a message to self to unregister
      self ! Unregister(ws)
      socketActorMapping(ws) ! StopListening
      socketActorMapping remove ws;
    }
  
    // recieves an error message
    case Error(ws, ex) => self ! Unregister(ws)
  
    // receives a text message
    case Message(ws, msg) => {
      log.debug("url {} received msg '{}'", ws.getResourceDescriptor, msg)
      ws.send("You send:" + msg);
    }
  
    // unregister the websocket listener
    case Unregister(ws) => {
      if (null != ws) {
        log.debug("unregister monitor")
        clients -= ws
      }
    }
  }
}

Код этого актера в значительной степени должен объяснить сам. Благодаря этому действующему субъекту и коду мы получили простой сервер веб-сокетов, который использует субъект для обработки сообщений. Прежде чем мы посмотрим на ListenActor, который запускается из сообщения «Open», полученного EchoHandler, давайте посмотрим, как мы подключаемся к mongoDB из нашего объекта DB:

package org.smartjava;
  
import play.api.libs.iteratee.{Concurrent, Enumeratee, Iteratee}
import reactivemongo.api.collections.default.BSONCollection
import reactivemongo.api._
import reactivemongo.bson.BSONDocument
import scala.concurrent.ExecutionContext.Implicits.global
  
/**
 * Contains DB related functions.
 */
class DB(location:String, dbname:String)  {
  
  // get connection to the database
  val db: DefaultDB = createConnection(location, dbname)
  // create a enumerator that we use to broadcast received documents
  val (bcEnumerator, channel) = Concurrent.broadcast[BSONDocument]
  // assign the channel to the mongodb cursor enumerator
  val iteratee = createCursor(getCollection(Configuration.collection))
                    .enumerate()
                    .apply(Iteratee
                        .foreach({doc: BSONDocument => channel.push(doc)}));
  
  /**
   * Return a simple collection
   */
  private def getCollection(collection: String): BSONCollection = {
    db(collection)
  }
  
  /**
   * Create the connection
   */
  private def createConnection(location: String, dbname: String)  : DefaultDB = {
    // needed to connect to mongoDB.
    import scala.concurrent.ExecutionContext
  
    // gets an instance of the driver
    // (creates an actor system)
    val driver = new MongoDriver
    val connection = driver.connection(List(location))
  
    // Gets a reference to the database
    connection(dbname)
  }
  
  /**
   * Create the cursor
   */
  private def createCursor(collection: BSONCollection): Cursor[BSONDocument] = {
    import reactivemongo.api._
    import reactivemongo.bson._
    import scala.concurrent.Future
  
    import scala.concurrent.ExecutionContext.Implicits.global
  
    val query = BSONDocument(
      "currentDate" -> BSONDocument(
        "$gte" -> BSONDateTime(System.currentTimeMillis())
      ));
  
    // we enumerate over a capped collection
    val cursor  = collection.find(query)
      .options(QueryOpts().tailable.awaitData)
      .cursor[BSONDocument]
  
    return cursor
  }
  
  /**
   * Simple function that registers a callback and a predicate on the
   * broadcasting enumerator
   */
  def listenToCollection(f: BSONDocument => Unit,
                         p: BSONDocument => Boolean ) = {
  
    val it = Iteratee.foreach(f)
    val itTransformed = Enumeratee.takeWhile[BSONDocument](p).transform(it);
    bcEnumerator.apply(itTransformed);
  }
}

Большая часть этого кода довольно стандартна, но я бы хотел отметить пару вещей. В начале этого класса мы настроили итератора следующим образом:

val db: DefaultDB = createConnection(location, dbname)
  val (bcEnumerator, channel) = Concurrent.broadcast[BSONDocument]
  val iteratee = createCursor(getCollection(Configuration.collection))
                    .enumerate()
                    .apply(Iteratee
                        .foreach({doc: BSONDocument => channel.push(doc)}));

Что мы делаем здесь, так это то, что мы сначала создаем широковещательный перечислитель, используя функцию Concurrent.broadcast Этот перечислитель может передавать элементы, предоставленные каналом, нескольким потребителям (итераторам). Затем мы создаем итератор на перечислителе, предоставляемом нашим курсором ReactiveMongo, где мы используем только что созданный канал для передачи документов любому итератору, подключенному к bcEnumerator.
Мы подключаем итераторов к bcEnumerator в функции listenToCollection:

def listenToCollection(f: BSONDocument => Unit,
                         p: BSONDocument => Boolean ) = {
  
    val it = Iteratee.foreach(f)
    val itTransformed = Enumeratee.takeWhile[BSONDocument](p).transform(it);
    bcEnumerator.apply(itTransformed);
  }

В этой функции мы передаем функцию и предикат. Функция выполняется всякий раз, когда документ добавляется в монго, а предикат используется для определения, когда прекратить отправку сообщений итерируемому.

Единственная недостающая часть — это ListenActor

ListenActor, который отвечает на сообщения от Mongo

В следующем коде показан субъект, отвечающий за ответы на сообщения от mongoDB. Когда он получает сообщение Listen, он регистрируется с помощью функции listenToCollection. Всякий раз, когда сообщение передается из Монго, оно отправляет сообщение самому себе, чтобы далее распространить его на веб-сокет.

object ListenActor {
  case class ReceiveUpdate(msg: String);
  def props(ws: WebSocket, db: DB): Props = Props(new ListenActor(ws, db))
}
class ListenActor(ws: WebSocket, db: DB) extends Actor with ActorLogging {
  
  var predicateResult = true;
  
  override def receive = {
    case Listen => {
  
      log.info("{} , {} , {}", ws, db)
  
      // function to call when we receive a message from the reactive mongo
      // we pass this to the DB cursor
      val func = ( doc: BSONDocument) => {
        self ! ReceiveUpdate(BSONDocument.pretty(doc));
      }
  
      // the predicate that determines how long we want to retrieve stuff
      // we do this while the predicateResult is true.
      val predicate = (d: BSONDocument) => {predicateResult} :Boolean
      Some(db.listenToCollection(func, predicate))
    }
  
    // when we recieve an update we just send it over the websocket
    case ReceiveUpdate(msg) => {
      ws.send(msg);
    }
  
    case StopListening => {
      predicateResult = false;
  
      // and kill ourselves
      self ! PoisonPill
    }
  }
}

Теперь, когда мы сделали все это, мы можем запустить этот пример. При запуске вы увидите что-то вроде этого:

[DEBUG] [11/22/2014 15:14:33.856] [main] [EventStream(akka://ws-system)] logger log1-Logging$DefaultLogger started
[DEBUG] [11/22/2014 15:14:33.857] [main] [EventStream(akka://ws-system)] Default Loggers started
[DEBUG] [11/22/2014 15:14:35.104] [main] [WSServer(akka://ws-system)] Registring actor:Some(Actor[akka://ws-system/user/echo#1509664759]) to /echo
15:14:35.211 [reactivemongo-akka.actor.default-dispatcher-5] INFO  reactivemongo.core.actors.MongoDBSystem - The node set is now available
15:14:35.214 [reactivemongo-akka.actor.default-dispatcher-5] INFO  reactivemongo.core.actors.MongoDBSystem - The primary is now available

Далее, когда мы подключаем веб-сокет, мы видим следующее:

[DEBUG] [11/22/2014 15:15:18.957] [WebSocketWorker-32] [WSServer(akka://ws-system)] OnOpen called org.java_websocket.WebSocketImpl@3161f479 :: org.java_websocket.handshake.HandshakeImpl1Client@6d9a6e19
[DEBUG] [11/22/2014 15:15:18.965] [ws-system-akka.actor.default-dispatcher-2] [akka://ws-system/user/echo] Received open request. Start listening for  WARNING arguments left: 1
[INFO] [11/22/2014 15:15:18.973] [ws-system-akka.actor.default-dispatcher-5] [akka://ws-system/user/echo/$a] org.java_websocket.WebSocketImpl@3161f479 , org.smartjava.DB@73fd64

Теперь давайте вставим сообщение в коллекцию Монго, которую мы создали с помощью следующей команды:

db.createCollection( "rmongo", { capped: true, size: 100000 } )

И давайте вставим сообщение:

> db.rmongo.insert({"test": 1234567, "currentDate": new Date()})
WriteResult({ "nInserted" : 1 })

Что приводит к этому в нашем клиенте websocket:

Если вас интересуют исходные файлы, посмотрите следующий каталог в GitHub: https://github.com/josdirksen/smartjava/tree/master/ws-akka

Ссылка:

ReactiveMongo с Akka, Scala и веб-сокетами от нашего партнера JCG Йоса Дирксена в блоге Smart Java .

ReactiveMongo с Akka, Scala и веб-сокетами

Начиная

Запуск сервера websocket и регистрация путей

Настройка сервера веб-сокетов

Акко эхо актер

ListenActor, который отвечает на сообщения от Mongo

Категории

Последние статьи

Рефакторинг Hudson God Class

Альтернативы синтаксиса Java лямбда

Morphia и MongoDB: развивающиеся структуры документов

OpenShift Express: развертывание приложения Java EE (с поддержкой AS7)

Интеграция jqGrid, REST, AJAX и Spring MVC