Concurrency test

mongotest.sc

import $ivy.`com.typesafe.akka::akka-actor:2.5.3`, akka.actor._
import $ivy.`com.typesafe.akka::akka-stream:2.5.4`, akka.stream._
import $ivy.`org.reactivemongo::reactivemongo:0.12.5`, reactivemongo.api._, reactivemongo.bson._, reactivemongo.api.collections.bson._ 
import $ivy.`ch.qos.logback:logback-classic:1.2.3`, org.slf4j._
import java.util.concurrent.Executors
import java.util.concurrent.atomic.AtomicLong
import scala.concurrent._
import scala.concurrent.duration._
import scala.language.postfixOps
import scala.util._

  val log = LoggerFactory.getLogger("application")

  implicit val ec = ExecutionContext.fromExecutor(Executors.newFixedThreadPool(20))

  implicit val system = ActorSystem("MainActorSystem")
  implicit val materializer = ActorMaterializer()

  val mongoDriver = new MongoDriver

  val mongoDb = for {
    uri <- Future.fromTry(MongoConnection.parseURI("mongodb://192.168.99.100:32768/limits"))
    con = mongoDriver.connection(uri)
    dn <- Future(uri.db.get)
    db <- con.database(dn)
  } yield db

  val collection = Await.result(mongoDb.map(_.collection[BSONCollection]("test")), 10 seconds)

  val id = BSONDocument("_id" -> 1)

  val atomicLong = new AtomicLong(0L)

  val futures = (1 to 20000) map { _ =>
    val i = 5000 - Random.nextInt(10000)

    val update = BSONDocument("$inc" -> BSONDocument("amount" -> i))

    atomicLong.updateAndGet(_ + i)

    collection.update(id, update, upsert = true).map(wr =>
      if(!wr.ok) { println("errror") }
    )
  }

  Await.result(Future.sequence(futures), 20 minutes)

  println(s"Atomic result: ${atomicLong.get()}")