//设置package包名称以及导入依赖的类
package com.hxm.earlgrey.jobs

import org.scalatest.{BeforeAndAfter, FunSuite}
import org.mongodb.scala.Document


class RepositoryTest extends FunSuite with BeforeAndAfter {
  val repository = Repository()
  val jobId = "scalaTestId"

  before {
    repository.deleteJob(jobId)
  }

  test("job insert and query") {
    val doc = Document("_id" -> jobId, "type" -> Job.Type.FlowQuery, "name" -> "scalatest", "status" -> Job.Status.Created)
    val job = new Job(doc)
    repository.insertJob(job)
    val j = repository.findJob(jobId)
    assert(j.isDefined)
    assert(j.get.doc.getString("_id") == jobId)
  }

} 

//设置package包名称以及导入依赖的类
package com.gbm.mywatchlist.utils.json

import com.gbm.mywatchlist.models.{Login, User}
import com.gbm.mywatchlist.utils.Password._
import org.bson.BsonValue
import org.mongodb.scala.Document
import org.mongodb.scala.bson.BsonString
import play.api.libs.json.Reads._
import play.api.libs.json._

object UserParser {

  private val reads: Reads[User] = new Reads[User] {
    override def reads(json: JsValue): JsResult[User] = {
      val firstName = (json \ "firstName").as[String](minLength[String](2))
      val lastName = (json \ "lastName").as[String](minLength[String](2))
      val emaiL = (json \ "email").as[String](email)
      JsSuccess(User(firstName = firstName, lastName = lastName, email = emaiL, password = ""))
    }
  }

  private val writes: Writes[User] = new Writes[User] {
    override def writes(user: User): JsValue = {
      Json.obj(
        "id" -> user.id,
        "firstName" -> user.firstName,
        "lastName" -> user.lastName,
        "email" -> user.email
      )
    }
  }

  implicit val userFormatterController = Format(reads, writes)

  implicit val userFormatterRepo = Json.format[User]

  implicit val loginFormatterController = Json.format[Login]

  implicit val userDocumentFormat = new UserDocumentFormat

  class UserDocumentFormat extends DocumentFormat[User] {

    override protected[json] def extractFields(user: User): List[(String, BsonValue)] = {
      "firstName" -> BsonString(user.firstName) :: "lastName" -> BsonString(user.lastName) ::
        "email" -> BsonString(user.email) :: "password" -> BsonString(user.password.encryptPassword) ::
        user.id.map(id => "_id" -> BsonString(id) :: Nil).getOrElse(Nil)
    }

    override protected[json] def buildEntity(document: Document): User = {
      val id = Option(document("_id").asObjectId().getValue.toHexString)
      val firstName = document("firstName").asString().getValue
      val lastName = document("lastName").asString().getValue
      val email = document("email").asString().getValue
      User(id = id, firstName = firstName, lastName = lastName, email = email, password = "")
    }
  }

} 

//设置package包名称以及导入依赖的类
package com.gbm.mywatchlist.utils.json

import com.gbm.mywatchlist.models.{Movie, Login, User}
import com.gbm.mywatchlist.utils.Password._
import org.bson.BsonValue
import org.mongodb.scala.Document
import org.mongodb.scala.bson.BsonString
import play.api.libs.json.Reads._
import play.api.libs.json._

object MovieParser {

  implicit val readsOMDB: Reads[Movie] = new Reads[Movie] {
    override def reads(json: JsValue): JsResult[Movie] = {
      val title = json \ "Title"
      val year = json \ "Year"
      val poster = json \ "Poster"
      val imdbId = json \ "imdbID"

      JsSuccess(Movie(title = title.as[String], year = year.as[String], poster = poster.as[String], imdbId = imdbId.as[String]))
    }
  }

  implicit val movieWrites = Json.writes[Movie]

  implicit val movieFormatter = Json.format[Movie]

  implicit val movieDocumentFormat = new MovieDocumentFormat

  class MovieDocumentFormat extends DocumentFormat[Movie] {

    override protected[json] def extractFields(movie: Movie): List[(String, BsonValue)] = {
      "title" -> BsonString(movie.title) :: "year" -> BsonString(movie.year) ::
        "poster" -> BsonString(movie.poster) :: "imdbId" -> BsonString(movie.imdbId) ::
        movie.userId.map(id => "userId" -> BsonString(id)).get ::
        movie.id.map(id => "_id" -> BsonString(id) :: Nil).getOrElse(Nil)
    }

    override protected[json] def buildEntity(document: Document): Movie = {
      val id = Option(document("_id").asObjectId().getValue.toHexString)
      val title = document("title").asString().getValue
      val year = document("year").asString().getValue
      val poster = document("poster").asString().getValue
      val imdbId = document("imdbId").asString().getValue
      val userId = Option(document("userId").asString().getValue)
      Movie(id = id, title = title, year = year, poster = poster, imdbId = imdbId, userId = userId)
    }
  }

} 

//设置package包名称以及导入依赖的类
package shine.st.blog.dao

import org.mongodb.scala.model.Filters.equal
import org.mongodb.scala.{Document, FindObservable, Observable}
import shine.st.blog.protocol.document.Categories
import shine.st.blog.utils.MongoUtils.ImplicitObservable
import spray.json._


trait CategoriesCollectionDao extends CollectionDao {
  override val collectionName: String = "categories"

  type T = Categories

  implicit class CategoriesObservable(val observable: Observable[Document]) extends ImplicitObservable[Document] {
    type R = T
    override val converter: (Document) => R = (doc) => {
      val jsonSource = doc.toJson
      val jsonAst = jsonSource.parseJson
      jsonAst.convertTo[R]
    }
  }

  def findByAncestors(someAncestor: String) = {
    find(equal("ancestors", someAncestor)).getResults
  }

  override def convert(observable: FindObservable[Document]) = observable.getResults

  //  FIXME: duplicate toJson
  override def toJson(data: T): JsValue = data.toJson
}

object CategoriesCollectionDao extends CategoriesCollectionDao 

//设置package包名称以及导入依赖的类
package shine.st.blog.dao

import org.mongodb.scala.model.Filters.{equal, or}
import org.mongodb.scala.{Document, FindObservable, Observable}
import shine.st.blog.protocol._
import shine.st.blog.protocol.document.Post
import shine.st.blog.utils.MongoUtils.ImplicitObservable
import spray.json._


trait PostCollectionDao extends CollectionDao {
  override val collectionName: String = "post"

  type T = Post

  implicit class PostObservable(val observable: Observable[Document]) extends ImplicitObservable[Document] {
    type R = T
    override val converter: (Document) => R = (doc) => {
      val jsonSource = doc.toJson
      val jsonAst = jsonSource.parseJson
      jsonAst.convertTo[R]
    }
  }

  def findByPath(path: String) = {
    //FIXME
    find(equal("path", path)).getHeadResult.get
  }

  def findByCategoryId(categoryId: List[String]) = {
    val condition = or(categoryId.map(c => equal("category_id", c)): _*)
    find(condition).getResults
  }

  override def convert(observable: FindObservable[Document]) = observable.getResults

  //  FIXME: duplicate toJson
  override def toJson(data: T): JsValue = data.toJson
}

object PostCollectionDao extends PostCollectionDao 

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

import de.khamrakulov.configs.MongoConstants
import de.khamrakulov.models.Shard
import org.mongodb.scala.{Document, MongoClient}

import scala.concurrent.Await
import scala.concurrent.duration._


trait ShardService {
  def getShards: Seq[Shard]
}

object ShardService {

  def apply(client: MongoClient) = new ShardServiceImpl(client)

  final class ShardServiceImpl(client: MongoClient) extends ShardService {

    private val collection = client.getDatabase(MongoConstants.CONFIG_DATABASE)
      .getCollection(MongoConstants.SHARDS_COLLECTION)

    override def getShards = {
      val shards = collection.find().map(parseShardInformation).toFuture()
      Await.result(shards, 10.seconds)
    }

    private def parseShardInformation(item: Document): Shard = {
      val document = item.toBsonDocument
      val shardId = document.getString(MongoConstants.SHARDS_ID).getValue
      val serversDefinition = document.getString(MongoConstants.SHARDS_HOST).getValue
      val servers = if (serversDefinition.contains("/")) serversDefinition.substring(serversDefinition.indexOf('/') + 1) else serversDefinition
      Shard(shardId, "mongodb://" + servers)
    }
  }

} 

//设置package包名称以及导入依赖的类
package org.psesd.srx.services.prs

import org.mongodb.scala.Document
import org.mongodb.scala.bson.BsonValue


object SslOrganization extends SslEntity {

  def apply(authorizedEntityId: String, externalServiceId: BsonValue, createdAt: BsonValue = null): Document = {
    val authorizedEntityXml = getAuthorizedEntity(authorizedEntityId)
    val timestamp = bsonTimeStamp

    var organization = Document("name" -> (authorizedEntityXml \ "authorizedEntity" \ "name").text,
      "website" -> (authorizedEntityXml \ "authorizedEntity" \ "mainContact" \ "webAddress").text,
      "url" -> (authorizedEntityXml \ "authorizedEntity" \ "mainContact" \ "webAddress").text,
      "authorizedEntityId" -> (authorizedEntityXml \ "authorizedEntity" \ "id").text.toInt,
      "externalServiceId" -> externalServiceId,
      "updated_at" -> timestamp)

    if (createdAt != null) {
      organization ++= Document("created_at" -> createdAt)
    } else {
      organization ++= Document("created_at" -> timestamp)
    }

    organization
  }

  def name(authorizedEntityId: String): String = {
    val authorizedEntityXml = getAuthorizedEntity(authorizedEntityId)
    (authorizedEntityXml \ "authorizedEntity" \ "name").text
  }

} 

//设置package包名称以及导入依赖的类
package teleporter.integration

import javax.sql.DataSource

import kafka.message.MessageAndMetadata
import org.apache.kafka.clients.producer.ProducerRecord
import org.elasticsearch.action.ActionRequest
import org.mongodb.scala.Document
import teleporter.integration.component.hbase.{HbaseAction, HbaseOut}
import teleporter.integration.component.jdbc.Action
import teleporter.integration.component.kudu.KuduAction


package object component {
  type KafkaMessage = MessageAndMetadata[Array[Byte], Array[Byte]]
  type KafkaRecord = ProducerRecord[Array[Byte], Array[Byte]]
  type JdbcMessage = Map[String, Any]
  type JdbcRecord = Seq[Action]
  type JdbcFunction = DataSource ? Unit
  type ElasticRecord = ActionRequest[_ <: AnyRef]
  type MongoMessage = Document
  type KuduRecord = KuduAction
  type HbaseRecord = HbaseAction
  type HbaseResult = HbaseOut
} 

//设置package包名称以及导入依赖的类
package teleporter.integration.cluster.broker.mongo

import akka.Done
import org.apache.logging.log4j.scala.Logging
import org.mongodb.scala.model.Filters._
import org.mongodb.scala.model.Updates._
import org.mongodb.scala.model._
import org.mongodb.scala.{Document, MongoCollection, MongoWriteException}
import teleporter.integration.cluster.broker.PersistentProtocol.KeyValue
import teleporter.integration.cluster.broker.PersistentService

import scala.concurrent.duration._
import scala.concurrent.{Await, ExecutionContext}


class MongoDBService(collection: MongoCollection[Document])(implicit ec: ExecutionContext) extends PersistentService with Logging {
  val timeout: FiniteDuration = 1.minutes
  val keyField = "_id"
  val valueField = "value"

  override def id(): Long =
    Await.result(collection.findOneAndUpdate(equal(keyField, "id"), inc(valueField, 1L), FindOneAndUpdateOptions().upsert(true)).map(_ (valueField).asInt64().longValue())
      .toFuture().map(_.head), timeout)

  override def range(key: String, start: Int, limit: Int): Seq[KeyValue] =
    Await.result(collection.find(regex(keyField, key)).skip(start).limit(limit)
      .map(doc ? KeyValue(doc(keyField).asString().getValue, doc(valueField).asString().getValue)).toFuture(), timeout)

  override def apply(key: String): KeyValue = get(key) match {
    case None ? throw new NoSuchElementException("key not found: " + key)
    case Some(v) ? v
  }

  override def get(key: String): Option[KeyValue] =
    Await.result(collection.find(equal(keyField, key))
      .map(doc ? KeyValue(doc(keyField).asString().getValue, doc(valueField).asString().getValue)).toFuture().map(_.headOption), timeout)

  override def unsafePut(key: String, value: String): Unit =
    Await.result(collection.updateOne(equal(keyField, key), set(valueField, value), UpdateOptions().upsert(true))
      .toFuture().map(_ ? Done), timeout)

  override def delete(key: String): Unit =
    Await.result(collection.deleteOne(equal(keyField, key)).toFuture().map(_ ? Done), timeout)

  override def unsafeAtomicPut(key: String, expect: String, update: String): Boolean =
    try {
      Await.result(collection.updateOne(Document(keyField ? key, valueField ? expect), set(valueField, update), UpdateOptions().upsert(true))
        .toFuture().map { results ?
        results.size == 1 && results.headOption.exists(r ? r.getMatchedCount == 1 || r.getModifiedCount == 1 || r.getUpsertedId != null)
      }, timeout)
    } catch {
      case e: MongoWriteException ?
        logger.warn(s"AtomicPut expect value not match, ${e.getMessage}")
        false
    }
}

object MongoDBService {
  def apply(collection: MongoCollection[Document])(implicit ec: ExecutionContext): MongoDBService = new MongoDBService(collection)
} 

//设置package包名称以及导入依赖的类
package operation.trivia.db

import com.mongodb.client.model.Filters
import operation.trivia.entities.Question
import org.mongodb.scala.bson.{BsonArray, BsonString}
import org.mongodb.scala.model.Updates
import org.mongodb.scala.{Completed, Document, MongoClient, MongoDatabase}
import org.scalatest.{BeforeAndAfter, FunSuite, Matchers}


class MongoDBTrialsTest extends FunSuite with Matchers with BeforeAndAfter {

  val mongoClient: MongoDatabase =
    MongoClient().getDatabase("operation_trivia")

  test("Test reading collections from Mongo DB") {
    mongoClient.listCollectionNames().subscribe((s: String) => println(s))
    Thread.sleep(4000)
  }


  test("Test reading collections from Mongo DB get limit of the first one") {

    def createQuestion(doc: Document): Option[Question] = {
      import collection.JavaConverters._
      for (
        item         <- doc.get("question").map(_.asString.getValue);
        actualAnswer <- doc.get("actualAnswer").map(_.asNumber().intValue());
        items        <- doc.get("answers").map(bson => bson.asArray()
                        .getValues.asScala.map(i => i.asString().getValue))
                        .map(_.toList)
      ) yield Question(item, items, actualAnswer)
    }

    mongoClient.getCollection("questions")
      .findOneAndUpdate(Filters.eq("taken", false), Updates.set("taken", true)).map(createQuestion)
      .subscribe((q: Option[Question]) => println(q))

    println("Done")
    Thread.sleep(4000)
  }

  test("Test create a Question") {
    def createDocument(question:Question): Document = {
      Document("question" -> question.item,
        "actualAnswer" -> question.actualAnswer,
        "answers" -> BsonArray(question.possibleAnswers.map(s => BsonString(s))))
    }

    mongoClient.getCollection("questions").insertOne(createDocument(Question("What is the capitol of Wyoming?",
                                                                    List("Cheyenne","Casper","Jackson","Sheridan"), 0)))
                                           .subscribe((c:Completed) => println(c))

    println("Done")
    Thread.sleep(4000)
  }
}