//设置package包名称以及导入依赖的类
package com.jetprobe.fastgen.io

import org.mongodb.scala.bson.BsonDocument
import org.mongodb.scala.{Completed, MongoClient, MongoCollection, Observer}
import org.mongodb.scala.bson.collection.mutable.Document
import scala.concurrent.duration._
import scala.concurrent.{Await, Future}


case class MongoDBSink(db: String, collection: String, host: String)
    extends DataSink {

  import MongoDBSink._

  override def write(data: Array[String]): Unit = {
    val collection = getCollection(this)

    data
      .grouped(1000)
      .foreach(docs => {
        val observable =
          collection.insertMany(docs.map(str => Document(BsonDocument(str))))
        Await.result(observable.head(), 10 seconds)
      })
    logger.info(s"Total docs inserted : ${data.length}")
  }

}

object MongoDBSink {

  import org.json4s._
  import org.json4s.jackson.JsonMethods._

  def apply(uri: String): MongoDBSink = {
    val splitUri = uri.substring(10).split("/")
    val hostname = "mongodb://" + splitUri(0)
    val database = splitUri(1)
    val collection = splitUri(2).split("\\?")(0)
    MongoDBSink(database, collection, hostname)
  }

  def getCollection(mongo: MongoDBSink): MongoCollection[Document] = {
    val mongoClient = MongoClient(mongo.host)
    mongoClient.getDatabase(mongo.db).getCollection(mongo.collection)
  }

  def jsonStrToMap(jsonStr: String): Map[String, Any] = {
    implicit val formats = org.json4s.DefaultFormats
    parse(jsonStr).extract[Map[String, Any]]
  }
} 

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

import org.mongodb.scala.bson.collection.immutable.Document
import org.mongodb.scala.{Completed, MongoCollection}
import org.mongounit.dsl.MongoRequestBuilding
import org.mongounit.embedded.EmbeddedMongo
import org.scalatest.Suite

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


trait MongoTest extends MongoRequestBuilding with EmbeddedMongo {
  this: MongoTestFrameworkInterface =>

  private def dropCollection(collection: MongoCollection[Document]): Unit = {
    Await.ready(collection.drop().toFuture(), 10.seconds)
  }

  private def importJson(collection: MongoCollection[Document], lines: Seq[String]): Future[Seq[Completed]] = {
    Await.ready(collection.insertMany(lines.map(json => Document(json))).toFuture(), 10.seconds)
  }

  implicit class WithSeqMongoRequestLoading(seqRequest: SeqMongoRequest) {
    def ~>(newReq: SeqMongoRequest): SeqMongoRequest = {
      SeqMongoRequest(newReq, seqRequest)
    }

    def ~>[A](block: => A): Unit = {
      seqRequest.requests.foreach { request =>
        val collection = mongoDB.getDatabase(request.database).getCollection(request.collection)
        importJson(collection, request.jsonList)
      }
      block
      seqRequest.requests.foreach { request =>
        val collection = mongoDB.getDatabase(request.database).getCollection(request.collection)
        dropCollection(collection)
      }
    }
  }

}


trait MongoScalaTest extends MongoTest with MongoTestFrameworkInterface.ScalaTest {
  this: Suite =>

  override def setup() = {
    mongod
  }

  override def cleanUp() = {
    mongoDB.close()
    mongod.stop()
    mongodExe.stop()
  }

} 

//设置package包名称以及导入依赖的类
package com.edu.chat.database

import akka.actor.{Actor, ActorLogging}
import com.edu.chat.Settings
import com.mongodb.ServerAddress
import org.mongodb.scala.bson.collection.immutable.Document
import org.mongodb.scala.connection.ClusterSettings
import org.mongodb.scala.{MongoClient, MongoClientSettings, MongoCollection, MongoCredential, MongoDatabase}

import scala.collection.JavaConverters._

abstract class DatabaseActor extends Actor with ActorLogging {
  private val applicationSettings = Settings(context.system)

  protected val collectionName: String

  protected def loadCollection: MongoCollection[Document] = {
    val clusterSettings = ClusterSettings
      .builder()
      .hosts(List(new ServerAddress(applicationSettings.dbHost, applicationSettings.dbPort)).asJava)
      .build()

    val credentials = List(MongoCredential.createCredential(
      applicationSettings.dbUsername,
      applicationSettings.dbName,
      applicationSettings.dbPassword)).asJava

    val dbSettings = MongoClientSettings
      .builder()
      .clusterSettings(clusterSettings)
      .credentialList(credentials)
      .build()

    val mongoDbClient = MongoClient(dbSettings)

    val database: MongoDatabase = mongoDbClient.getDatabase(applicationSettings.dbName)
    database.getCollection(collectionName)
  }
} 

//设置package包名称以及导入依赖的类
package org.styx.mongo

import com.fasterxml.jackson.databind.ObjectMapper
import org.bson.RawBsonDocument
import org.mongodb.scala.{Completed, MongoCollection}
import org.styx.handler.EventHandler
import org.styx.handler.EventHandler.{SuccessfullyHandled, UnsuccessfullyHandled}
import org.styx.model.Event
import org.styx.state.State
import org.styx.state.State.AggregationId

import scala.concurrent.{ExecutionContext, Future}


trait MongoDBEventHandler[S <: State] extends EventHandler[S] {
  val collection: MongoCollection[MongoDBEvent]
  val objectMapper: ObjectMapper

  override def handle(aggregationId: AggregationId, event: Event[S], actualState: S)(implicit executionContext: ExecutionContext): Future[EventHandler.HandleStatus[S]] = {
    collection
      .insertOne {
        MongoDBEvent(
          eventType = event.getClass.getSimpleName,
          aggregationId = aggregationId,
          data = RawBsonDocument.parse(objectMapper.writeValueAsString(event.data)),
          eventDate = event.eventDate,
          revision = event.revision
        )
      }.head
      .map { case Completed() => SuccessfullyHandled(event) }
      .recover { case e: Exception => UnsuccessfullyHandled(e, event) }
  }
}

object MongoDBEventHandler {
  def apply[S <: State](mongoCollection: MongoCollection[MongoDBEvent], mapper: ObjectMapper): MongoDBEventHandler[S] = new MongoDBEventHandler[S](){
    override val collection: MongoCollection[MongoDBEvent] = mongoCollection
    override val objectMapper: ObjectMapper = mapper
  }
} 

//设置package包名称以及导入依赖的类
package org.styx.mongo

import com.fasterxml.jackson.databind.ObjectMapper
import org.mongodb.scala.MongoCollection
import org.mongodb.scala.bson.Document
import org.styx.handler.EventFetcher
import org.styx.model.Event
import org.styx.state.State
import org.styx.state.State.AggregationId

import scala.collection.Seq
import scala.concurrent.{ExecutionContext, Future}


trait MongoDBEventFetcher[S <: State] extends EventFetcher[S] {
  val collection: MongoCollection[MongoDBEvent]
  val objectMapper: ObjectMapper
  val converter: (MongoDBEvent => Event[S])

  override def get(aggregationId: AggregationId)(implicit executionContext: ExecutionContext): Future[Seq[Event[S]]] = {
    val eventualEvents: Future[Seq[MongoDBEvent]] = collection
      .find(Document("aggregationId" -> aggregationId))
      .sort(Document("eventDate" -> 1))
      .collect()
      .head()

    eventualEvents.map[Seq[Event[S]]] {
      events => events.map(e => converter(e)).distinct
    }
  }
}

object MongoDBEventFetcher {
  def apply[S <: State](col: MongoCollection[MongoDBEvent],
                        mapper: ObjectMapper,
                        eventConverter: (MongoDBEvent => Event[S])): MongoDBEventFetcher[S] = new MongoDBEventFetcher[S]() {
    override val collection: MongoCollection[MongoDBEvent] = col
    override val objectMapper: ObjectMapper = mapper
    override val converter: (MongoDBEvent) => Event[S] = eventConverter
  }
} 

//设置package包名称以及导入依赖的类
package com.stulsoft.pmongodb.example01

import com.stulsoft.pmongodb.example01.Helpers._
import com.typesafe.scalalogging.LazyLogging
import org.bson.codecs.configuration.CodecRegistries.{fromProviders, fromRegistries}
import org.mongodb.scala.bson.ObjectId
import org.mongodb.scala.bson.codecs.DEFAULT_CODEC_REGISTRY
import org.mongodb.scala.bson.codecs.Macros._
import org.mongodb.scala.model.Filters._
import org.mongodb.scala.{MongoClient, MongoCollection, MongoDatabase}


object Example3 extends App with LazyLogging {

  private lazy val codecRegistry = fromRegistries(fromProviders(classOf[Person]), DEFAULT_CODEC_REGISTRY)

  // To directly connect to the default server localhost on port 27017
  val client: MongoClient = MongoClient()
  val database: MongoDatabase = client.getDatabase("testDb").withCodecRegistry(codecRegistry)
  val collection: MongoCollection[Person] = database.getCollection("testCollection3")
  collection.drop().results()

  val person: Person = Person("Ada", "Lovelace")
  collection.insertOne(person).results()

  collection.find().first().printHeadResult()
  collection.find().printResults()
  collection.find(equal("firstName", "Ada")).first().printHeadResult()
  collection.find(equal("lastName", "Lovelace")).first().printHeadResult()

  case class Person(_id: ObjectId, firstName: String, lastName: String)

  object Person {
    def apply(firstName: String, lastName: String): Person =
      Person(new ObjectId(), firstName, lastName)
  }

} 

//设置package包名称以及导入依赖的类
package com.stulsoft.pmongodb.example01

import com.stulsoft.pmongodb.example01.Helpers._
import com.typesafe.scalalogging.LazyLogging
import org.bson.codecs.configuration.CodecRegistries.{fromProviders, fromRegistries}
import org.mongodb.scala.bson.ObjectId
import org.mongodb.scala.bson.codecs.DEFAULT_CODEC_REGISTRY
import org.mongodb.scala.bson.codecs.Macros._
import org.mongodb.scala.model.Filters._
import org.mongodb.scala.{MongoClient, MongoCollection, MongoDatabase}

import com.mongodb.MongoCredential
import org.mongodb.scala.bson.collection.mutable.Document
import org.mongodb.scala.{Completed, FindObservable, MongoClient, MongoClientSettings, MongoCollection, MongoDatabase, Observable,Observer, ReadPreference, ServerAddress}
import org.mongodb.scala.connection.ClusterSettings
import com.mongodb.MongoCredential._
import java.util.logging.{Level, Logger}
import org.mongodb.scala.connection.{NettyStreamFactoryFactory,SslSettings}
import scala.collection.JavaConverters._



object Cloud1 extends App with LazyLogging {

  private lazy val codecRegistry = fromRegistries(fromProviders(classOf[Person]), DEFAULT_CODEC_REGISTRY)


  val clusterSettings: ClusterSettings = ClusterSettings.builder().hosts(List(new ServerAddress("cluster0-shard-00-00-k8h55.mongodb.net:27017"),
    new ServerAddress("cluster0-shard-00-01-k8h55.mongodb.net:27017"),
    new ServerAddress("cluster0-shard-00-02-k8h55.mongodb.net:27017")
  ).asJava).build()
  val user: String = "<USER NAME>>"
  val databasename: String = "testDb"
  val password: Array[Char] = "<PASSWORD>".toCharArray
  val credential: MongoCredential = createCredential(user, databasename, password)
  val settings: MongoClientSettings = MongoClientSettings.builder()
    .clusterSettings(clusterSettings).credentialList(List(credential,credential).asJava).sslSettings(SslSettings.builder().enabled(true).build())
    .streamFactoryFactory(NettyStreamFactoryFactory()).build()
  val client: MongoClient = MongoClient(settings)

  val database: MongoDatabase = client.getDatabase("testDb").withCodecRegistry(codecRegistry)
  val collection: MongoCollection[Person] = database.getCollection("testCollection3")
  collection.drop().results()

  val person: Person = Person("Ada", "Lovelace")
  collection.insertOne(person).results()

  collection.find().first().printHeadResult()
  collection.find().printResults()
  collection.find(equal("firstName", "Ada")).first().printHeadResult()
  collection.find(equal("lastName", "Lovelace")).first().printHeadResult()

  case class Person(_id: ObjectId, firstName: String, lastName: String)

  object Person {
    def apply(firstName: String, lastName: String): Person =
      Person(new ObjectId(), firstName, lastName)
  }

} 

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

import java.util.concurrent.TimeUnit.SECONDS

import org.bson.json.JsonMode
import org.bson.json.JsonWriterSettings
import org.mongodb.scala.MongoClient
import org.mongodb.scala.MongoCollection
import org.mongodb.scala.MongoDatabase
import org.mongodb.scala.bson.collection.immutable.Document

import scala.concurrent.Await
import scala.concurrent.Future
import scala.concurrent.duration.Duration
import scala.language.postfixOps
import scala.util.Failure
import scala.util.Success
import scala.util.Try

class MongoService(mongoQueryParser: MongoQueryParser) {

  // You will only need one instance of class MongoClient even with multiple concurrently executing asynchronous operations.
  val mongoClient: MongoClient = MongoClient("mongodb://localhost")
  val database: MongoDatabase = mongoClient.getDatabase("vm")
  val collection: MongoCollection[Document] = database.getCollection("letters")

  def executeQuery(query: String): Seq[String] = Try {
    val result: Future[Document] = collection.find(mongoQueryParser.parse(query)).first().head()
    List(Await.result(result, Duration(30, SECONDS)).toJson(new JsonWriterSettings(JsonMode.STRICT, true)))
  } match {
    case Success(value) => value
    case Failure(f) => List()
  }
} 

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

import fresco.db.Helpers._
import org.mongodb.scala.bson.collection.immutable.Document
import org.mongodb.scala.{MongoClient, MongoCollection}


object Database {
  val mongoClient = MongoClient()
  val mongoDb = mongoClient.getDatabase("fresco")

  implicit def printResults() = {

  }

  def getStores() = {
    val collection: MongoCollection[Document] = mongoDb.getCollection("stores")
    val doc: Document = Document("_id" -> 0, "name" -> "MongoDB", "type" -> "database",
      "count" -> 1, "info" -> Document("x" -> 203, "y" -> 102))
    // collection.insertOne(doc).results()
    collection.find().printResults()
  }

}

object TestDatabase extends App {
  val stores = Database.getStores()
  println(s"$stores")
} 

//设置package包名称以及导入依赖的类
package g8.akkahttp.eventsvc.data.access

import g8.akkahttp.eventsvc.Config
import g8.akkahttp.eventsvc.data.Event
import org.mongodb.scala.bson.BsonDocument
import org.mongodb.scala.model.Filters
import org.mongodb.scala.{Completed, MongoClient, MongoCollection, Observer}

import scala.concurrent.{Future, Promise}

trait MongoDataStoreImpl extends DataStoreComponent  {

  private val _dataStoreImpl = new DataStoreImpl
  def dataStore: DataStore = _dataStoreImpl

  class DataStoreImpl extends DataStore {

    // EVENTS

    override def createEvents(numToCreate: Int): Future[Boolean] = {
      val p = Promise[Boolean]()
      val database = MongoDataStoreConnection.client.getDatabase("sdsi").withCodecRegistry(Event.codecRegistry)
      val collection: MongoCollection[Event] = database.getCollection("events")

      val events = for (i <- 0 until numToCreate) yield Event.generateRandom

      collection.insertMany(events).subscribe(new Observer[Completed] {
        override def onNext(res: Completed): Unit = {}
        override def onError(e: Throwable): Unit = { println(s"error occurred ${e.getMessage}"); p.failure(e) }
        override def onComplete(): Unit = { println("completed"); p.success(true) }
      })

      p.future
    }

    override def getEventsByType(triggerType: Int): Future[List[BsonDocument]] = {
      val p = Promise[List[BsonDocument]]
      val database = MongoDataStoreConnection.client.getDatabase("sdsi")
      val collection: MongoCollection[BsonDocument] = database.getCollection("events")

      collection.find(Filters.`eq`("trigger.triggerType", triggerType)).collect().subscribe(
        (results: Seq[BsonDocument]) => p.success(results.toList),
        (error: Throwable) => p.failure(error)
      )

      p.future
    }
  }
}

object MongoDataStoreConnection extends Config {

  private val uri: String = config.getString("mongoConnectionUri")
  System.setProperty("org.mongodb.async.type", "netty")

  // TODO: Introduce the standard UUID codec for compatibility with other drivers

  lazy val client: MongoClient = MongoClient(uri)
} 

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

import com.typesafe.config.{Config, ConfigFactory}
import de.neuland.models.ReminderRecord
import org.bson.codecs.configuration.CodecRegistries.{fromProviders, fromRegistries}
import org.bson.codecs.configuration.CodecRegistry
import org.mongodb.scala.bson.codecs.DEFAULT_CODEC_REGISTRY
import org.mongodb.scala.bson.codecs.Macros._
import org.mongodb.scala.model.Filters._
import org.mongodb.scala.{Completed, MongoClient, MongoCollection, MongoDatabase}
import play.api.Logger

import scala.concurrent.Future

class ReminderRepository {
  val conf: Config = ConfigFactory.load

  val codecRegistry: CodecRegistry = fromRegistries(fromProviders(classOf[ReminderRecord]), DEFAULT_CODEC_REGISTRY )
  val mongoClient: MongoClient = MongoClient(conf.getString("db.mongo"))
  val database: MongoDatabase = mongoClient.getDatabase(conf.getString("db.database")).withCodecRegistry(codecRegistry)
  val collection: MongoCollection[ReminderRecord] = database.getCollection("reminders")

  def save(reminderRecord: ReminderRecord): Future[Completed] = {
    Logger.debug("Save " + reminderRecord)
    collection.insertOne(reminderRecord).head()
  }

  def all(): Future[Seq[ReminderRecord]] =  {
    collection.find().toFuture()
  }

  def find(author: String): Future[Seq[ReminderRecord]] = {
    collection.find(equal("author", author)).toFuture()
  }

} 

//设置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)
}