From 5b2f3b9b9ec8fe80933ece5a442e66634b39308d Mon Sep 17 00:00:00 2001 From: david Date: Fri, 25 Apr 2014 21:13:40 +0200 Subject: [PATCH 1/2] Fixed problem with the creation of ActorSystem: Couch and Database create a new ActorSystem instead of using it from Config. --- src/main/scala/sprouch/Couch.scala | 6 +++--- src/main/scala/sprouch/Database.scala | 4 ++-- 2 files changed, 5 insertions(+), 5 deletions(-) diff --git a/src/main/scala/sprouch/Couch.scala b/src/main/scala/sprouch/Couch.scala index b933d3f..514b04a 100644 --- a/src/main/scala/sprouch/Couch.scala +++ b/src/main/scala/sprouch/Couch.scala @@ -33,7 +33,7 @@ case class SprouchException(error:ErrorResponse) extends Exception * Class that handles the connection to CouchDB. It contains methods for creating, looking up and deleting databases. */ class Couch(config:Config) extends UriBuilder { - implicit val system = ActorSystem() + implicit val system = config.actorSystem import system.dispatcher // execution context for futures private val pipelines = new Pipelines(config) @@ -44,7 +44,7 @@ class Couch(config:Config) extends UriBuilder { * Creates a new database. Fails if the database already exists. */ def createDb(dbName:String):Future[Database] = { - pipeline(Put(dbUri(dbName))).map(_ => new Database(dbName, pipelines)) + pipeline(Put(dbUri(dbName))).map(_ => new Database(dbName, pipelines, config)) } /** * Deletes a database and all containing documents. @@ -56,7 +56,7 @@ class Couch(config:Config) extends UriBuilder { * Looks up a database by its name. */ def getDb(dbName:String):Future[Database] = { - getDbPipeline(Get(dbUri(dbName))).map(_ => new Database(dbName, pipelines)) + getDbPipeline(Get(dbUri(dbName))).map(_ => new Database(dbName, pipelines, config)) } } diff --git a/src/main/scala/sprouch/Database.scala b/src/main/scala/sprouch/Database.scala index a3e9fff..b234803 100644 --- a/src/main/scala/sprouch/Database.scala +++ b/src/main/scala/sprouch/Database.scala @@ -23,9 +23,9 @@ import spray.json.JsValue * Supports CRUD operations on documents and attachments, * creating and querying views, bulk get, update, and delete operations. */ -class Database private[sprouch](val name:String, pipelines:Pipelines) extends UriBuilder { +class Database private[sprouch](val name:String, pipelines:Pipelines, config:Config) extends UriBuilder { import pipelines._ - implicit val system = ActorSystem() + implicit val system = config.actorSystem import system.dispatcher // execution context for futures private def dbUri:String = dbUri(name) From 00a6feb8af382bc68da8e4a2a507346dd15a51b3 Mon Sep 17 00:00:00 2001 From: david Date: Wed, 14 May 2014 21:37:38 +0200 Subject: [PATCH 2/2] Added a bulkPutWithErrors operation that returns the list of errors in the operation. --- src/main/scala/sprouch/Database.scala | 9 +++++ src/main/scala/sprouch/JsonProtocol.scala | 26 +++++++++++--- src/test/scala/sprouch/BulkActions.scala | 41 +++++++++++++++++++++-- 3 files changed, 69 insertions(+), 7 deletions(-) diff --git a/src/main/scala/sprouch/Database.scala b/src/main/scala/sprouch/Database.scala index b234803..35648db 100644 --- a/src/main/scala/sprouch/Database.scala +++ b/src/main/scala/sprouch/Database.scala @@ -71,6 +71,15 @@ class Database private[sprouch](val name:String, pipelines:Pipelines, config:Con crs.zip(docs).map { case (cr, doc) => doc.setRev(cr.rev) } }) } + + def bulkPutWithError[A:RootJsonFormat](docs:Seq[Document[A]]):Future[(Seq[RevedDocument[A]], Seq[ErrorBulkResponse])] = { + val p = pipeline[Seq[BulkResponse]] + p(Post(bulkUri, BulkPut(docs))).map(crs => { + val (revDocs, errors) = crs.zip(docs).partition(_._1.isInstanceOf[CreateResponse]) + (revDocs.map { case (cr, doc) => doc.setRev(cr.asInstanceOf[CreateResponse].rev) }, + errors.map { case (ebr, _) => ebr.asInstanceOf[ErrorBulkResponse] }) + }) + } /** * Deletes the entire database. diff --git a/src/main/scala/sprouch/JsonProtocol.scala b/src/main/scala/sprouch/JsonProtocol.scala index 2fae99f..55c7be5 100644 --- a/src/main/scala/sprouch/JsonProtocol.scala +++ b/src/main/scala/sprouch/JsonProtocol.scala @@ -50,7 +50,9 @@ object JsonProtocol extends DefaultJsonProtocol { case class AllDocsResponse[A](total_rows:Int, offset:Int, rows:Seq[AllDocsRow[A]]) implicit def allDocsResponseFormat[A:RootJsonFormat] = jsonFormat3(AllDocsResponse[A]) case class OkResponse(ok:Boolean) - case class CreateResponse(ok:Option[Boolean], id:String, rev:String) + trait BulkResponse + case class CreateResponse(ok:Option[Boolean], id:String, rev:String) extends BulkResponse + case class ErrorBulkResponse(id:String, error:String, reason:String) extends BulkResponse case class ErrorResponse(status:Int, body:Option[ErrorResponseBody]) case class ErrorResponseBody(error:String, reason:String) case object Empty @@ -62,7 +64,9 @@ object JsonProtocol extends DefaultJsonProtocol { implicit val getDbResponseFormat = jsonFormat10(GetDbResponse) implicit val okResponseFormat = jsonFormat1(OkResponse) implicit val createResponseFormat = jsonFormat3(CreateResponse) + implicit val errorBulkResponseFormat = jsonFormat3(ErrorBulkResponse) implicit val errorResponseFormat = jsonFormat2(ErrorResponseBody) + implicit val bulkResponseFormat = new BulkResponseFormat implicit def revedDocJsonFormat[A:RootJsonFormat]:RootJsonFormat[RevedDocument[A]] = new RevedDocFormat[A] implicit def newDocJsonFormat[A:RootJsonFormat]:RootJsonFormat[NewDocument[A]] = new NewDocFormat[A] implicit def documentFormat[A:RootJsonFormat]:RootJsonFormat[Document[A]] = new AnyDocFormat[A] @@ -125,7 +129,7 @@ object JsonProtocol extends DefaultJsonProtocol { ) JsObject(dataFields ++ docFields ++ otherFields(doc)) } - case js => throw new Exception("data does not serialize to json object: " + js) + case js => throw new Exception("data does not serialize to value object: " + js) } } @@ -137,12 +141,12 @@ object JsonProtocol extends DefaultJsonProtocol { val _id = stringFormat.read(fields("_id")) val attachments = fields.get("_attachments").toList.flatMap { case JsObject(as) => as.map { case (k,v) => k -> attachmentStubFormat.read(v) } - case _ => deserializationError("json array expected") + case _ => deserializationError("value array expected") }.toMap val data = dataFormat.read(o) makeB(fields, _id, data, attachments) } - case _ => deserializationError("json object expected") + case _ => deserializationError("value object expected") } } @@ -166,7 +170,19 @@ object JsonProtocol extends DefaultJsonProtocol { case class BulkPut[A](docs:Seq[Document[A]]) implicit def bulkPutFormat[A:RootJsonFormat] = jsonFormat1(BulkPut[A]) - + + class BulkResponseFormat extends RootJsonFormat[BulkResponse] { + override def read(value: JsValue): BulkResponse = value match { + case o:JsObject if o.fields.contains("error") => + o.convertTo[ErrorBulkResponse] + case o:JsObject => + o.convertTo[CreateResponse] + case _ => deserializationError("value object expected") + } + + override def write(obj: BulkResponse): JsValue = obj.toJson + } + implicit val nothingFormat = new JsonFormat[Nothing] { def read(js:JsValue) = throw new Exception("fields of type nothing should never be used") def write(n:Nothing) = throw new Exception("fields of type nothing should never be used") diff --git a/src/test/scala/sprouch/BulkActions.scala b/src/test/scala/sprouch/BulkActions.scala index 8f16e7d..ac5a613 100644 --- a/src/test/scala/sprouch/BulkActions.scala +++ b/src/test/scala/sprouch/BulkActions.scala @@ -24,6 +24,43 @@ class BulkActions extends FunSuite with CouchSuiteHelpers { } }) } - - + test("get the list of errors when a update conflict happens") { + withNewDb(db => { + val data = Seq(Test(0, "a"), Test(1, "b"), Test(2, "c")).map(new NewDocument(_)) + for { + bulkInserted <- db.bulkPutWithError(data) + newData = bulkInserted._1.map(doc => doc.updateData(data => data.copy(foo = data.foo + 1))) + bulkUpdated <- db.bulkPutWithError(newData) + bulkConflict <- db.bulkPutWithError(bulkInserted._1) + bulkGotten <- db.allDocs[Test](keys = data.map(_.id)) + } yield { + assert(bulkConflict._1 === Seq()) + bulkConflict._2.zip(bulkInserted._1).foreach { + case (error, document) => + assert(error.id === document.id) + assert(error.error === "conflict") + assert(error.reason === "Document update conflict.") + } + bulkGotten + } + }) + } + test("get the list of errors when a create conflict happens") { + withNewDb(db => { + val data = Seq(Test(0, "a"), Test(1, "b"), Test(2, "c")).map(d => new NewDocument(d.foo.toString, d)) + for { + bulkInserted <- db.bulkPutWithError(data) + bulkConflict <- db.bulkPutWithError(data) + } yield { + assert(bulkConflict._1 === Seq()) + bulkConflict._2.zip(bulkInserted._1).foreach { + case (error, document) => + assert(error.id === document.id) + assert(error.error === "conflict") + assert(error.reason === "Document update conflict.") + } + bulkInserted._1 + } + }) + } } \ No newline at end of file