//设置package包名称以及导入依赖的类
package services

import models.db.User
import models.rest.{UserId, UserRest}
import play.modules.reactivemongo.MongoController
import reactivemongo.api.collections.bson.BSONCollection
import reactivemongo.bson.{BSONDocument, BSONObjectID}

import scala.concurrent.Future


trait RegistrationService {
  def insertUser(userData: UserRest): Future[UserId]
  def findUser(uuid: String): Future[Option[UserRest]]
}

trait MongoRegistrationService extends RegistrationService {
  self: MongoController =>

  import scala.concurrent.ExecutionContext.Implicits.global

  def collection: BSONCollection = db.collection[BSONCollection]("users")

  def insertUser(userData: UserRest): Future[UserId] = {

    val insertRec = User(_id = BSONObjectID.generate, firstName = userData.firstName, lastName = userData.lastName)

    collection.insert(insertRec).map { _ =>
      UserId(insertRec._id.stringify)
    }
  }

  def findUser(uuid: String): Future[Option[UserRest]] = {
      BSONObjectID.parse(uuid).map { objId =>
        val query = BSONDocument("_id" -> objId)

        collection.find(query).one[User].map { result =>
          result.map { user => UserRest(firstName = user.firstName, lastName = user.lastName) }
        }
      } getOrElse(Future.successful(None))
  }
} 

//设置package包名称以及导入依赖的类
package com.jjabuk.bookstore.catalog.protocols

import akka.http.scaladsl.marshallers.sprayjson.SprayJsonSupport
import com.jjabuk.bookstore.catalog.protocols.CatalogueProtocol.{Book, BookAdded, BooksFound}
import reactivemongo.bson.BSONObjectID
import spray.json.{DefaultJsonProtocol, JsArray, JsObject, JsString, JsValue, RootJsonFormat}

trait JsonSupport extends SprayJsonSupport with DefaultJsonProtocol {

  implicit val BookAddedFormat = jsonFormat1(BookAdded.apply)

  implicit object BooksFoundFormat extends RootJsonFormat[BooksFound] {
    override def read(json: JsValue): BooksFound = ???

    override def write(b: BooksFound): JsValue = JsObject(
      "books" -> JsArray(b.books.map(book => BookFormat.write(book)).toVector)
    )
  }

  implicit object BookFormat extends RootJsonFormat[Book] {
    override def read(value: JsValue) = {
      val uuid = fromField[Option[String]](value, "uuid")
      val isbn = fromField[String](value, "isbn")
      val title = fromField[String](value, "title")
      val review = fromField[Option[String]](value, "review")
      val publisher = fromField[Option[String]](value, "publisher")
      Book(uuid.getOrElse(BSONObjectID.generate().stringify), isbn, title, review, publisher)
    }

    override def write(obj: Book): JsValue = JsObject(
      "uuid" -> JsString(obj.uuid),
      "isbn" -> JsString(obj.isbn),
      "title" -> JsString(obj.title),
      "review" -> JsString(obj.review.getOrElse("")),
      "publisher" -> JsString(obj.publisher.getOrElse(""))
    )
  }
} 

//设置package包名称以及导入依赖的类
package repos

import scala.concurrent.{ExecutionContext, Future}
import reactivemongo.bson.{BSONDocument, BSONObjectID}
import play.api.libs.json.{JsObject, Json}
import reactivemongo.api.commands.WriteResult
import play.modules.reactivemongo.ReactiveMongoApi
import javax.inject._
import play.modules.reactivemongo.json._
import play.modules.reactivemongo.json.collection.JSONCollection
import reactivemongo.api.ReadPreference

trait PostRepo {
	def find() (implicit ec: ExecutionContext): Future[List[JsObject]]

	def select(selector: BSONDocument) (implicit ec: ExecutionContext): Future[Option[JsObject]]

	def update(selector: BSONDocument, update: BSONDocument) (implicit ec: ExecutionContext): Future[WriteResult]

	def remove(document: BSONDocument) (implicit ec: ExecutionContext): Future[WriteResult]

	def save(document: BSONDocument) (implicit ec: ExecutionContext): Future[WriteResult]

}

class PostRepoImpl @Inject() (reactiveMongoApi: ReactiveMongoApi) extends PostRepo {
  def collection = reactiveMongoApi.db.collection[JSONCollection]("posts");
  
  override def find() (implicit ec: ExecutionContext): Future[List[JsObject]] = {
    val genericQueryBuilder = collection.find(Json.obj());
    val cursor = genericQueryBuilder.cursor[JsObject](ReadPreference.Primary);
    cursor.collect[List]()
  }

  def select(selector: BSONDocument)(implicit ec: ExecutionContext): Future[Option[JsObject]] = {
    collection.find(selector).one[JsObject]
  }

  def update(selector: BSONDocument, update: BSONDocument)(implicit ec: ExecutionContext): Future[WriteResult] = {
    collection.update(selector,update)
  }

  def remove(document: BSONDocument)(implicit ec: ExecutionContext): Future[WriteResult] = {
    collection.remove(document)
  }

  def save(document: BSONDocument)(implicit ec: ExecutionContext): Future[WriteResult] = {
    collection.update(BSONDocument("_id" -> document.get("_id").getOrElse(BSONObjectID.generate)), document, upsert = true)
  }
  
} 

//设置package包名称以及导入依赖的类
package repos

import javax.inject.Inject

import play.api.libs.json.{JsObject, Json}
import play.modules.reactivemongo.ReactiveMongoApi
import play.modules.reactivemongo.json._
import reactivemongo.api.ReadPreference
import reactivemongo.api.commands.WriteResult
import reactivemongo.bson.{BSONDocument, BSONObjectID}
import reactivemongo.play.json.collection.JSONCollection

import scala.concurrent.{ExecutionContext, Future}

trait RepoLike {
  def find()(implicit ec: ExecutionContext): Future[List[JsObject]]

  def select(selector: BSONDocument)(implicit ec: ExecutionContext): Future[Option[JsObject]]

  def update(selector: BSONDocument, update: BSONDocument)(implicit ec: ExecutionContext): Future[WriteResult]

  def remove(document: BSONDocument)(implicit ec: ExecutionContext): Future[WriteResult]

  def save(document: BSONDocument)(implicit ec: ExecutionContext): Future[WriteResult]
}

abstract class Repo @Inject()(reactiveMongoApi: ReactiveMongoApi) extends RepoLike {
  val collectionName: String

  override def find()(implicit ec: ExecutionContext): Future[List[JsObject]] = {
    collection.find(Json.obj())
      .cursor[JsObject](ReadPreference.Primary)
      .collect[List](100)
  }

  override def select(selector: BSONDocument)(implicit ec: ExecutionContext): Future[Option[JsObject]] = {
    collection.find(selector).one[JsObject]
  }

  override def update(selector: BSONDocument, update: BSONDocument)(implicit ec: ExecutionContext): Future[WriteResult] = {
    collection.update(selector, update)
  }

  def collection: JSONCollection = reactiveMongoApi.db.collection[JSONCollection](collectionName)

  override def remove(document: BSONDocument)(implicit ec: ExecutionContext): Future[WriteResult] = {
    collection.remove(document)
  }

  override def save(document: BSONDocument)(implicit ec: ExecutionContext): Future[WriteResult] = {
    collection.update(BSONDocument("_id" -> document.get("_id").getOrElse(BSONObjectID.generate)), document, upsert = true)
  }
} 

//设置package包名称以及导入依赖的类
package json

import org.joda.time.{DateTimeZone, DateTime}
import play.api.libs.json._
import reactivemongo.bson.BSONObjectID


object MongoImplicits {
  // Reads / Writes implicit for BSONObjectID
  implicit val objectIdRead: Reads[BSONObjectID] =
    (__ \ "$oid").read[String].map { oid =>
      BSONObjectID(oid)
    }
  implicit val objectIdWrite: Writes[BSONObjectID] = new Writes[BSONObjectID] {
    def writes(objectId: BSONObjectID): JsValue = Json.obj(
      "$oid" -> objectId.stringify
    )
  }
  // Read / write implicit for DateTime
  implicit val dateTimeRead: Reads[DateTime] =
    (__ \ "$date").read[Long].map { dateTime =>
      new DateTime(dateTime, DateTimeZone.UTC)
    }

  implicit val dateTimeWrite: Writes[DateTime] = new Writes[DateTime] {
    def writes(dateTime: DateTime): JsValue = Json.obj(
      "$date" -> dateTime.getMillis
    )
  }

  implicit val objectIdFormats = Format(objectIdRead, objectIdWrite)
  implicit val dateTimeFormats = Format(dateTimeRead, dateTimeWrite)

} 

//设置package包名称以及导入依赖的类
package extensions

import reactivemongo.api.Cursor
import play.api.libs.json._
import scala.concurrent.{ExecutionContext, Future}
import reactivemongo.bson.BSONObjectID

object ReactiveMongoExtensions {

  implicit class ListExtensions[T](val futureList: Future[List[T]]) extends AnyVal {
    def toJsArray(implicit ec: ExecutionContext, writes: Writes[T]): Future[JsArray] = {
      futureList.map { futureList =>
        futureList.foldLeft(JsArray(List()))( (obj, item) => obj ++ Json.arr(item))
      }
    }
  }

  implicit class CursorExtensions[T](val cursor: Cursor[T]) extends AnyVal {
    def toJsArray(implicit ec: ExecutionContext, writes: Writes[T]): Future[JsArray] = {
      cursor.toList.toJsArray
    }
  }

  implicit class BSONObjectIdExtensions(val string: String) extends AnyVal {
    def toObjectID: BSONObjectID = BSONObjectID(string)
  }

} 

//设置package包名称以及导入依赖的类
package models

import reactivemongo.api._
import play.modules.reactivemongo._
import play.modules.reactivemongo.json.collection.JSONCollection
import play.api.libs.json.Json

import play.api.Play.current
import reactivemongo.bson.BSONObjectID
import play.modules.reactivemongo.json.BSONFormats._

case class Beach(
	name: String,
	image: Image,
  _id: BSONObjectID = BSONObjectID.generate
) {
  val id = _id.stringify
}

object Beach {
  def collection = ReactiveMongoPlugin.db.collection[JSONCollection]("beaches")

  // Json Format
  implicit val format = Json.format[Beach]
} 

//设置package包名称以及导入依赖的类
package com.jjabuk.bookstore.catalog

import com.jjabuk.bookstore.catalog.protocols.CatalogueProtocol.{Book, BookAdded}
import com.jjabuk.bookstore.persistence.MongoDb
import org.slf4j.{Logger, LoggerFactory}
import reactivemongo.api.collections.bson.BSONCollection
import reactivemongo.bson.{BSONDocument, BSONDocumentWriter, BSONObjectID}
import reactivemongo.core.errors.DatabaseException

import scala.concurrent.ExecutionContext.Implicits.global
import scala.concurrent.Future

trait BookCommands {
  def addBook(book: Book): Future[BookAdded]
}

object MongoBookCommands extends BookCommands with MongoDb {

  def logger: Logger = LoggerFactory.getLogger(this.getClass)

  def books: Future[BSONCollection] = database map (_.collection("books"))

  implicit def bookWriter = new BSONDocumentWriter[Book] {
    def write(book: Book): BSONDocument =
      BSONDocument(
        "_id" -> BSONObjectID.parse(book.uuid).getOrElse(BSONObjectID.generate),
        "isbn" -> book.isbn,
        "title" -> book.title,
        "review" -> book.review.orNull,
        "publisher" -> book.publisher.orNull
      )
  }

  def addBook(book: Book): Future[BookAdded] = {
    logger.debug(s"Adding a book $book")

    books
      .flatMap(_
        .insert(book)
        .map(doc => {
          BookAdded(book.uuid)
        }))
      .recover({
        case de: DatabaseException =>
          logger.warn("DatabaseException: " + de.getMessage())
          logger.warn("Cause: " + de.getCause())
          logger.warn("Code: " + de.code)
          BookAdded(book.uuid)
      })
  }
} 

//设置package包名称以及导入依赖的类
package com.jjabuk.bookstore.catalog

import com.jjabuk.bookstore.catalog.protocols.CatalogueProtocol.{Book, BooksFound}
import com.jjabuk.bookstore.persistence.MongoDb
import org.slf4j.LoggerFactory
import reactivemongo.api.ReadPreference
import reactivemongo.api.collections.bson.BSONCollection
import reactivemongo.bson.{BSONDocument, BSONDocumentReader, BSONObjectID}

import scala.concurrent.ExecutionContext.Implicits.global
import scala.concurrent.{ExecutionContext, Future}

trait BookQueries {
  def findAll(implicit ec: ExecutionContext): Future[BooksFound]

  def findByUuid(uuid: String)(implicit ec: ExecutionContext): Future[Option[BooksFound]]
}

object MongoBookQueries extends BookQueries with MongoDb {

  def logger = LoggerFactory.getLogger(this.getClass)

  def books: Future[BSONCollection] = database map (_.collection("books"))

  implicit def bookReader = new BSONDocumentReader[Book] {

    def read(doc: BSONDocument): Book =
      Book(
        uuid = doc.getAs[BSONObjectID]("_id").get.stringify,
        isbn = doc.getAs[String]("isbn").get,
        title = doc.getAs[String]("title").get,
        review = doc.getAs[String]("review"),
        publisher = doc.getAs[String]("publisher")
      )
  }

  def findAll(implicit ec: ExecutionContext): Future[BooksFound] = {
    logger.debug(s"Find all books")

    books flatMap {
      _.find(BSONDocument()).cursor[Book](ReadPreference.Primary).collect[List]()
    } map (books => BooksFound(books))
  }

  def findByUuid(uuid: String)(implicit ec: ExecutionContext): Future[Option[BooksFound]] = {
    logger.debug(s"Find a book by uuid=$uuid")

    val selector = BSONDocument("_id" -> BSONObjectID.parse(uuid).get)

    books
      .flatMap(_.find(selector).one[Book])
      .map {
        case Some(book) => Some(BooksFound(List(book)))
        case _ => Option.empty[BooksFound]
      }
  }

} 

//设置package包名称以及导入依赖的类
package it.agilelab.bigdata.wasp.testApp

import it.agilelab.bigdata.wasp.core.WaspSystem
import it.agilelab.bigdata.wasp.core.logging.WaspLogger
import it.agilelab.bigdata.wasp.core.models.{IndexModel, TopicModel}
import it.agilelab.bigdata.wasp.web.Global
import play.api.libs.json.Json
import reactivemongo.bson.BSONObjectID

object HttpApp extends App with WaspSystem {

  def loggerActorProps = Global.loggerActorProps
  val logger = WaspLogger("HttpApp")

  val avroCustom = s"""
      {"type":"record",
      "namespace":"Finance",
      "name":"Stock",
      "fields":[${TopicModel.schema_base},
        {"name":"stock_name","type":"string"},
        {"name":"bid","type":"double"},
        {"name":"ask","type":"double"},
        {"name":"percent_change","type":"double"},
        {"name":"volume","type":"double"},
        {"name":"currency","type":"string"}
  ]}"""

  val jsonES = s"""
      {"yahoofinance":
          {"properties":{
            ${IndexModel.schema_base_elastic},
        "stock_name":{"type":"string","index":"not_analyzed","store":"true","enabled":"true"},
        "bid":{"type":"double","index":"not_analyzed","store":"true","enabled":"true"},
        "ask":{"type":"double","index":"not_analyzed","store":"true","enabled":"true"},
        "percent_change":{"type":"double","index":"not_analyzed","store":"true","enabled":"true"},
        "volume":{"type":"double","index":"not_analyzed","store":"true","enabled":"true"},
        "currency":{"type":"string","index":"not_analyzed","store":"true","enabled":"true"}
          }}}"""

  val originaljson = Json.parse(jsonES)
  val id_pipeline = Some(BSONObjectID.generate)
  val id_producer = Some(BSONObjectID.generate)


} 

//设置package包名称以及导入依赖的类
package it.agilelab.bigdata.wasp.consumers.MlModels

import it.agilelab.bigdata.wasp.core.models.MlModelOnlyInfo
import org.apache.spark.ml.Transformer
import org.apache.spark.ml.param.Params
import org.joda.time.DateTime
import reactivemongo.bson.BSONObjectID




case class TransformerWithInfo(name: String, version: String,
                                transformer: Transformer with Params,
                                timestamp: Long = DateTime.now().getMillis,
                                favorite: Boolean = false, description: String = "",
                                _id: Option[BSONObjectID] = None,
                                modelFileId: Option[BSONObjectID] = None) {
  val className: String = transformer.getClass.getName
  def toOnlyInfo(modelFileId: BSONObjectID) = {
    MlModelOnlyInfo(_id = _id, name = name, version = version, className = Some(className),
      timestamp = Some(timestamp), favorite = favorite, description = description,
      modelFileId = Some(modelFileId)
    )
  }
  def toOnlyInfo = {
    MlModelOnlyInfo(_id = _id, name = name, version = version, className = Some(className),
      timestamp = Some(timestamp), favorite = favorite, description = description,
      modelFileId = modelFileId)
  }
}

object TransformerWithInfo {
  def create(mlModelOnlyInfo: MlModelOnlyInfo, transformer: Transformer with Params): TransformerWithInfo = {

    TransformerWithInfo(
      _id = mlModelOnlyInfo._id,
      name = mlModelOnlyInfo.name,
      version = mlModelOnlyInfo.version,
      transformer = transformer,
      timestamp = mlModelOnlyInfo.timestamp.getOrElse(DateTime.now().getMillis),
      favorite = mlModelOnlyInfo.favorite,
      description = mlModelOnlyInfo.description,
      modelFileId = mlModelOnlyInfo.modelFileId
    )
  }
} 

//设置package包名称以及导入依赖的类
package it.agilelab.bigdata.wasp.consumers

import it.agilelab.bigdata.wasp.core.WaspSystem
import it.agilelab.bigdata.wasp.core.models.IndexModel
import it.agilelab.bigdata.wasp.core.utils.BSONFormats
import reactivemongo.bson.BSONObjectID


//TODO: Index di prova per batch, rimuovere
private[consumers] object BatchOutputIndex {

  val index_name = "BatchOutput"

  def apply() = IndexModel(
    name = IndexModel.normalizeName(index_name),
    creationTime = WaspSystem.now,
    schema = BSONFormats.fromString(indexSchema),
    _id = Some(BSONObjectID.generate)
  )

  private val indexSchema = s"""
      {"yahoofinance":
          {"properties":{
            ${IndexModel.schema_base_elastic},
        "stock_name":{"type":"string","index":"not_analyzed","store":"true","enabled":"true"},
        "bid":{"type":"double","index":"not_analyzed","store":"true","enabled":"true"},
        "ask":{"type":"double","index":"not_analyzed","store":"true","enabled":"true"},
        "percent_change":{"type":"double","index":"not_analyzed","store":"true","enabled":"true"},
        "volume":{"type":"double","index":"not_analyzed","store":"true","enabled":"true"},
        "currency":{"type":"string","index":"not_analyzed","store":"true","enabled":"true"}
          }}}"""
} 

//设置package包名称以及导入依赖的类
package it.agilelab.bigdata.wasp.pipegraph.metro.models

import com.typesafe.config.{Config, ConfigFactory}
import it.agilelab.bigdata.wasp.core.WaspSystem
import it.agilelab.bigdata.wasp.core.models.{IndexModel, StrategyModel, _}
import reactivemongo.bson.BSONObjectID


object MetroPipegraphModel {

  lazy val metroPipegraphName = "MetroPipegraph6"

  lazy val metroPipegraph = MetroPipegraph()

  lazy val conf: Config = ConfigFactory.load

  lazy val defaultDataStoreIndexed = conf.getString("default.datastore.indexed")

  private[wasp] object MetroPipegraph {

    def apply() =
      PipegraphModel(
          name = MetroPipegraphModel.metroPipegraphName,
          description = "Los Angeles Metro Pipegraph",
          owner = "user",
          system = false,
          creationTime = WaspSystem.now,
          etl = List(
              ETLModel("write on index",
                       List(ReaderModel(MetroTopicModel.metroTopic._id.get,
                                        MetroTopicModel.metroTopic.name,
                                        TopicModel.readerType)),
                       WriterModel.IndexWriter(MetroIndexModel.metroIndex._id.get,
                                               MetroIndexModel.metroIndex.name, defaultDataStoreIndexed),
                       List(),
                       Some(StrategyModel("it.agilelab.bigdata.wasp.pipegraph.metro.strategies.MetroStrategy", None))
              )
          ),
          rt = Nil,
          dashboard = None,
          isActive = false,
          _id = Some(BSONObjectID.generate))

  }
} 

//设置package包名称以及导入依赖的类
package it.agilelab.bigdata.wasp.pipegraph.metro.models

import it.agilelab.bigdata.wasp.core.WaspSystem
import it.agilelab.bigdata.wasp.core.models.IndexModel
import it.agilelab.bigdata.wasp.core.utils.BSONFormats
import reactivemongo.bson.BSONObjectID


object MetroIndexModel {

  lazy val metroIndex = MetroIndex()

  private[wasp] object MetroIndex {

    val index_name = "Metro"

    def apply() = IndexModel(
        name = IndexModel.normalizeName(index_name),
        creationTime = WaspSystem.now,
        // Elastic
        schema = BSONFormats.fromString(indexSchemaElastic),
        // Solr
        //schema = BSONFormats.fromString(indexSchemaSolr),
        _id = Some(BSONObjectID.generate),
        query = None,
        // Work only with Solr
        numShards = Some(2),
        // Work only with Solr
        replicationFactor = Some(1)
    )

    private val indexSchemaElastic = s"""
      {"lametro":
          {
          "properties":{
            ${IndexModel.schema_base_elastic},
            "last_update":{"type":"double","index":"not_analyzed","store":"true","enabled":"true"},
            "vehicle_code":{"type":"integer","index":"not_analyzed","store":"true","enabled":"true"},
            "position":{"type":"geo_point","geohash": true, "geohash_prefix" : true, "geohash_precision": 7, "index":"not_analyzed","store":"true","enabled":"true"},
            "predictable":{"type":"boolean","index":"not_analyzed","store":"true","enabled":"true"}
          }}}"""

    private val indexSchemaSolr = s"""
      { "properties":
        [
          ${IndexModel.schema_base_solr},
          { "name":"last_update", "type":"tdouble", "stored":true },
          { "name":"vehicle_code", "type":"tint", "stored":true },
          { "name" :"position", "type" : "string", "stored" : true },
          { "name":"predictable", "type":"boolean", "stored":true }
        ]
      }"""

  }

} 

//设置package包名称以及导入依赖的类
package it.agilelab.bigdata.wasp.pipegraph.metro.models

import it.agilelab.bigdata.wasp.core.WaspSystem
import it.agilelab.bigdata.wasp.core.models.TopicModel
import it.agilelab.bigdata.wasp.core.utils.BSONFormats
import reactivemongo.bson.BSONObjectID


object MetroTopicModel {

  lazy val metroTopic = MetroTopic()

  private[wasp] object MetroTopic {

    val topic_name = "Metro"

    def apply() = TopicModel(
      name = TopicModel.name(topic_name),
      creationTime = WaspSystem.now,
      partitions = 3,
      replicas = 1,
      topicDataType = "avro",    // avro or json
      schema = Some(BSONFormats.fromString(topicSchema).get),
      _id = Some(BSONObjectID.generate)
    )

    private val topicSchema = s"""
      {"type":"record",
      "namespace":"PublicTransportsTracking",
      "name":"LAMetro",
      "fields":[${TopicModel.schema_base},
        {"name":"last_update","type":"double"},
        {"name":"vehicle_code","type":"int"},
        {"name":"position", "type":{"name":"positionData", "type":"record", "fields":[{"name":"lat","type":"double"}, {"name":"lon","type":"double"}]}},
        {"name":"predictable","type":"boolean"}
  ]}"""

  }

} 

//设置package包名称以及导入依赖的类
package it.agilelab.bigdata.wasp.core.models.configuration

import it.agilelab.bigdata.wasp.core.utils.{ConfigManager, ConnectionConfig}
import reactivemongo.bson.BSONObjectID


case class SolrConfigModel(
                            connections: Seq[ConnectionConfig],
                            apiEndPoint: Option[ConnectionConfig],
                            name: String,
                            _id: Option[BSONObjectID],
                            cluster_name: String
                          )


object SolrConfigModel {
  val default = SolrConfigModel(
    connections = Seq(ConnectionConfig("", "localhost", 8983, None, Some(Map.empty[String, String]))),
    apiEndPoint = Some(ConnectionConfig("http", "localhost", 8983, None, Some(Map("zookeeperRootNode" -> "/solr")))),
    ConfigManager.solrConfigName,
    None,
    "wasp"
  )
} 

//设置package包名称以及导入依赖的类
package it.agilelab.bigdata.wasp.core.models

import it.agilelab.bigdata.wasp.core.utils.ConfigManager
import reactivemongo.bson.{BSONDocument, BSONObjectID, BSONString}

object IndexModel {
  val readerType = "index"

  val schema_base_elastic =
    """
    "id_event":{"type":"double","index":"not_analyzed","store":"true","enabled":"true"},
    "source_name":{"type":"string","index":"not_analyzed","store":"true","enabled":"true"},
    "Index_name":{"type":"string","index":"not_analyzed","store":"true","enabled":"true"},
    "metric_name":{"type":"string","index":"not_analyzed","store":"true","enabled":"true"},
    "timestamp":{"type":"date","format":"date_time","index":"not_analyzed","store":"true","enabled":"true"},
    "latitude":{"type":"double","index":"not_analyzed","store":"true","enabled":"true"},
    "longitude":{"type":"double","index":"not_analyzed","store":"true","enabled":"true"},
    "value":{"type":"double","index":"not_analyzed","store":"true","enabled":"true"},
    "payload":{"type":"string","index":"not_analyzed","store":"true","enabled":"true"}
               """

  val schema_base_solr = """
                         { "name":"id_event", "type":"tdouble", "stored":true },
                         { "name":"source_name", "type":"string", "stored":true },
                         { "name":"topic_name", "type":"string","stored":true },
                         { "name":"metric_name", "type":"string","stored":true },
                         { "name":"timestamp", "type":"string","stored":true },
                         { "name":"latitude", "type":"tdouble", "stored":true },
                         { "name":"longitude", "type":"tdouble", "stored":true },
                         { "name":"value", "type":"string", "stored":true },
                         { "name":"payload", "type":"string", "stored":true }
                    """

  def normalizeName(basename: String) = s"${basename.toLowerCase}_index"
}

case class IndexModel(override val name: String,
                      creationTime: Long,
                      schema: Option[BSONDocument],
                      _id: Option[BSONObjectID] = None,
                      query: Option[String] = None,
                      numShards: Option[Int] = Some(1),
                      replicationFactor: Option[Int] = Some(1))
    extends Model {

  def resource = s"${ConfigManager.buildTimedName(name)}/$dataType"

  def collection = s"${ConfigManager.buildTimedName(name)}"

  def dataType =
    schema
      .map(
          bson =>
            bson.elements.headOption
              .getOrElse(("undefined", BSONString("undefined")))
              ._1)
      .getOrElse("undefined")

} 

//设置package包名称以及导入依赖的类
package it.agilelab.bigdata.wasp.core.models

import reactivemongo.bson.{BSONDocument, BSONObjectID}

object TopicModel {
  val readerType = "topic"

  val schema_base = """
      {"name":"id_event","type":"double"},
      {"name":"source_name","type":"string"},
      {"name":"topic_name","type":"string"},
      {"name":"metric_name","type":"string"},
      {"name":"timestamp","type":"string","java-class" : "java.util.Date"},
      {"name":"latitude","type":"double"},
      {"name":"longitude","type":"double"},
      {"name":"value","type":"double"},
      {"name":"payload","type":"string"}
                    """

  def name(basename: String) = s"${basename.toLowerCase}.topic"
}

case class TopicModel(override val name: String,
                      creationTime: Long,
                      partitions: Int,
                      replicas: Int,
                      topicDataType: String, // avro, json, xml
                      schema: Option[BSONDocument],
                      _id: Option[BSONObjectID] = None) extends Model 

//设置package包名称以及导入依赖的类
package it.agilelab.bigdata.wasp.core.models

import it.agilelab.bigdata.wasp.core.WaspSystem
import reactivemongo.bson.BSONObjectID

object JobStateEnum extends Enumeration {
  type JobState = Value
  val PENDING = "PENDING"
  val PROCESSING = "PROCESSING"
  val SUCCESSFUL = "SUCCESSFUL"
  val FAILED = "FAILED"

}

case class BatchJobModel(
                          override val name: String,
                          description: String,
                          owner: String,
                          system: Boolean,
                          creationTime: Long,
                          etl: ETLModel,
                          var state: String,
                          _id: Option[BSONObjectID] = None
                          ) extends Model {

} 

//设置package包名称以及导入依赖的类
package it.agilelab.bigdata.wasp.core.bl

import it.agilelab.bigdata.wasp.core.models.{BSONConversionHelper, IndexModel}
import it.agilelab.bigdata.wasp.core.utils.WaspDB
import reactivemongo.bson.{BSONObjectID, BSONString}
import reactivemongo.api.commands.WriteResult

import scala.concurrent.ExecutionContext.Implicits.global
import scala.concurrent.Future

trait IndexBL {
  def getByName(name: String): Future[Option[IndexModel]]

  def getById(id: String): Future[Option[IndexModel]]

  def persist(indexModel: IndexModel): Future[WriteResult]

}

class IndexBLImp(waspDB: WaspDB) extends IndexBL with BSONConversionHelper {

  private def factory(t: IndexModel) =
    new IndexModel(t.name, t.creationTime, t.schema, t._id, t.query, t.numShards, t.replicationFactor)

  def getByName(name: String) = {
    waspDB
      .getDocumentByField[IndexModel]("name", new BSONString(name))
      .map(index => {
        index.map(p => factory(p))
      })
  }

  def getById(id: String) = {
    waspDB
      .getDocumentByID[IndexModel](BSONObjectID(id))
      .map(Index => {
        Index.map(p => factory(p))
      })
  }

  override def persist(indexModel: IndexModel): Future[WriteResult] =
    waspDB.insert[IndexModel](indexModel)
}