<NME> HttpStringStoreProperties.scala
<BEF> ADDFILE
<MSG> Merge pull request #241 from tresata/feature/storehaus-http

implementation of store that uses http protocol
<DFF> @@ -0,0 +1,109 @@
+/*
+ * Copyright 2014 Twitter Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License. You may obtain
+ * a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.twitter.storehaus.http
+
+import java.util.concurrent.ConcurrentHashMap
+import java.nio.charset.Charset
+import org.jboss.netty.buffer.ChannelBuffers
+import org.jboss.netty.handler.codec.http.{ HttpRequest, HttpResponse, DefaultHttpResponse, HttpResponseStatus, HttpMethod, HttpHeaders }
+import com.twitter.util.{ Await, Future }
+import com.twitter.finagle.{ Service, Http, ListeningServer }
+import com.twitter.storehaus.{ FutureOps, Store }
+import com.twitter.storehaus.testing.CloseableCleanup
+import com.twitter.storehaus.testing.generator.NonEmpty
+import org.scalacheck.{ Arbitrary, Gen, Properties }
+import org.scalacheck.Prop._
+
+object HttpStringStoreProperties extends Properties("HttpStringStore") with CloseableCleanup[ListeningServer] {
+  def validPairs: Gen[List[(String, Option[String])]] =
+    NonEmpty.Pairing.alphaStrs().map(_.map{ case (k, v) => ("/" + k, v) })
+
+  def baseTest[K: Arbitrary, V: Arbitrary : Equiv](store: Store[K, V], validPairs: Gen[List[(K, Option[V])]])
+                                                  (put: (Store[K, V], List[(K, Option[V])]) => Unit) =
+    forAll(validPairs) { (examples: List[(K, Option[V])]) =>
+      put(store, examples)
+      examples.toMap.forall { case (k, optV) =>
+        val res = Await.result(store.get(k))
+        Equiv[Option[V]].equiv(res, optV)
+      }
+    }
+
+  def putStoreTest[K: Arbitrary, V: Arbitrary : Equiv](store: Store[K, V], validPairs: Gen[List[(K, Option[V])]]) =
+    baseTest(store, validPairs) { (s, pairs) =>
+      pairs.foreach {
+        case (k, v) =>
+          Await.result(s.put((k, v)))
+      }
+    }
+
+  def multiPutStoreTest[K: Arbitrary, V: Arbitrary : Equiv](store: Store[K, V], validPairs: Gen[List[(K, Option[V])]]) =
+    baseTest(store, validPairs) { (s, pairs) => 
+      Await.result(FutureOps.mapCollect(s.multiPut(pairs.toMap)))
+    }
+
+  def storeTest(store: Store[String, String]) =
+    putStoreTest(store, validPairs) && multiPutStoreTest(store, validPairs)
+
+  val service = new Service[HttpRequest, HttpResponse] {
+    private val map = new ConcurrentHashMap[String, String]()
+    private val utf8 = Charset.forName("UTF-8")
+
+    def apply(request: HttpRequest): Future[HttpResponse] = {
+      val response = request.getMethod match {
+        case HttpMethod.GET =>
+          Option(map.get(request.getUri)).map{ v =>
+            val resp = new DefaultHttpResponse(request.getProtocolVersion, HttpResponseStatus.OK)
+            val content = ChannelBuffers.wrappedBuffer(v.getBytes(utf8))
+            resp.setContent(content)
+            resp.headers.set(HttpHeaders.Names.CONTENT_LENGTH, content.readableBytes.toString)
+            resp
+          }.getOrElse {
+            val resp = new DefaultHttpResponse(request.getProtocolVersion, HttpResponseStatus.NOT_FOUND)
+            resp.headers.set(HttpHeaders.Names.CONTENT_LENGTH, "0")
+            resp
+          }
+        case HttpMethod.DELETE =>
+          map.remove(request.getUri)
+          val resp = new DefaultHttpResponse(request.getProtocolVersion, HttpResponseStatus.NO_CONTENT)
+          resp.headers.set(HttpHeaders.Names.CONTENT_LENGTH, "0")
+          resp
+        case HttpMethod.PUT =>
+          val maybeOldV = Option(map.put(request.getUri, request.getContent.toString(utf8)))
+          val resp = new DefaultHttpResponse(request.getProtocolVersion, maybeOldV.map(_ => HttpResponseStatus.OK).getOrElse(HttpResponseStatus.CREATED))
+          resp.setContent(request.getContent)
+          resp.headers.set(HttpHeaders.Names.CONTENT_LENGTH, request.getContent.readableBytes.toString)
+          resp
+        case _ =>
+          new DefaultHttpResponse(request.getProtocolVersion, HttpResponseStatus.METHOD_NOT_ALLOWED)
+      }
+      Future.value(response)
+    }
+  }
+
+  val server = Http.serve("localhost:0", service)
+
+  val store = HttpStringStore(server.boundAddress.toString.substring(1)) // i dont know how else to convert boundAddress into something usable
+
+  property("HttpStringStore test") = storeTest(store)
+
+  override def closeable = server
+
+  override def cleanup() = {
+    println("closing server")
+    super.cleanup()
+  }
+}

{'commit': 'diff --git a/project/Build.scala b/project/Build.scala\nindex 3f6d271..4bb2e30 100644\n--- a/project/Build.scala\n+++ b/project/Build.scala\n@@ -145,6 +145,7 @@ object StorehausBuild extends Build {\n     storehausKafka08,\n     storehausMongoDB,\n     storehausElastic,\n+    storehausHttp,\n     storehausTesting\n   )\n \n@@ -292,4 +293,12 @@ object StorehausBuild extends Build {\n       javaOptions in run <++= (fullClasspath in Runtime) map { cp => Seq("-cp", sbt.Build.data(cp).mkString(":")) }\n   ).dependsOn(storehausCore, storehausAlgebra, storehausCache)\n \n+  lazy val storehausHttp = module("http").settings(\n+    libraryDependencies ++= Seq(\n+      Finagle.module("http"),\n+      "com.twitter" %% "bijection-netty" % bijectionVersion\n+    )\n+  ).dependsOn(storehausCore)\n+\n+\n }\ndiff --git a/storehaus-http/src/main/scala/com/twitter/storehaus/http/HttpStore.scala b/storehaus-http/src/main/scala/com/twitter/storehaus/http/HttpStore.scala\nnew file mode 100644\nindex 0000000..39dee3c\n--- /dev/null\n+++ b/storehaus-http/src/main/scala/com/twitter/storehaus/http/HttpStore.scala\n@@ -0,0 +1,81 @@\n+/*\n+ * Copyright 2014 Twitter Inc.\n+ *\n+ * Licensed under the Apache License, Version 2.0 (the "License"); you may\n+ * not use this file except in compliance with the License. You may obtain\n+ * a copy of the License at\n+ *\n+ *     http://www.apache.org/licenses/LICENSE-2.0\n+ *\n+ * Unless required by applicable law or agreed to in writing, software\n+ * distributed under the License is distributed on an "AS IS" BASIS,\n+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n+ * See the License for the specific language governing permissions and\n+ * limitations under the License.\n+ */\n+\n+package com.twitter.storehaus.http\n+\n+import java.nio.charset.Charset\n+import org.jboss.netty.buffer.{ ChannelBuffer, ChannelBuffers }\n+import org.jboss.netty.handler.codec.http.{ HttpRequest, HttpResponse, DefaultHttpRequest, HttpVersion, HttpMethod, HttpHeaders, HttpResponseStatus }\n+import com.twitter.util.Future\n+import com.twitter.bijection.StringCodec\n+import com.twitter.bijection.netty.ChannelBufferBijection\n+import com.twitter.finagle.{ Service, Http }\n+import com.twitter.storehaus.{ Store, ConvertedStore }\n+\n+object HttpException {\n+  def apply(response: HttpResponse): HttpException = \n+    new HttpException(response.getStatus.getCode, response.getStatus.getReasonPhrase, response.getContent.toString(Charset.forName("UTF-8")))\n+}\n+\n+case class HttpException(code: Int, reasonPhrase: String, content: String) extends Exception(reasonPhrase + Option(content).map("\\n" + _ ).getOrElse(""))\n+\n+object HttpStore {\n+  def apply(dest: String): HttpStore = new HttpStore(Http.newService(dest))\n+}\n+\n+class HttpStore(val client: Service[HttpRequest, HttpResponse]) extends Store[String, ChannelBuffer] {\n+  override def get(k: String): Future[Option[ChannelBuffer]] = {\n+    val request =  new DefaultHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.GET, k)\n+    request.headers.set(HttpHeaders.Names.CONTENT_LENGTH, "0")\n+    client(request).map{ response => \n+      response.getStatus match {\n+        case HttpResponseStatus.OK => Some(response.getContent)\n+        case HttpResponseStatus.NOT_FOUND => None\n+        case _ => throw HttpException(response)\n+      }\n+    }\n+  }\n+\n+  override def put(kv: (String, Option[ChannelBuffer])): Future[Unit] = {\n+    val request = kv match {\n+      case (k, Some(cb)) =>\n+        val req = new DefaultHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.PUT, k)\n+        req.setContent(cb)\n+        req.headers.set(HttpHeaders.Names.CONTENT_LENGTH, cb.readableBytes.toString)\n+        req\n+      case (k, None) =>\n+        val req = new DefaultHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.DELETE, k)\n+        req.headers.set(HttpHeaders.Names.CONTENT_LENGTH, "0")\n+        req\n+    }\n+    client(request).map{ response =>\n+      response.getStatus match {\n+        case HttpResponseStatus.OK => ()\n+        case HttpResponseStatus.CREATED => ()\n+        case HttpResponseStatus.NO_CONTENT => ()\n+        case _ => throw HttpException(response)\n+      }\n+    }\n+  }\n+}\n+\n+object HttpStringStore {\n+  def apply(dest: String): HttpStringStore = new HttpStringStore(Http.newService(dest))\n+}\n+\n+class HttpStringStore(val client: Service[HttpRequest, HttpResponse])\n+    extends ConvertedStore[String, String, ChannelBuffer, String](new HttpStore(client))(identity)(\n+      StringCodec.utf8 andThen ChannelBufferBijection.inverse)\ndiff --git a/storehaus-http/src/test/scala/com/twitter/storehaus/http/HttpStringStoreProperties.scala b/storehaus-http/src/test/scala/com/twitter/storehaus/http/HttpStringStoreProperties.scala\nnew file mode 100644\nindex 0000000..853897a\n--- /dev/null\n+++ b/storehaus-http/src/test/scala/com/twitter/storehaus/http/HttpStringStoreProperties.scala\n@@ -0,0 +1,109 @@\n+/*\n+ * Copyright 2014 Twitter Inc.\n+ *\n+ * Licensed under the Apache License, Version 2.0 (the "License"); you may\n+ * not use this file except in compliance with the License. You may obtain\n+ * a copy of the License at\n+ *\n+ *     http://www.apache.org/licenses/LICENSE-2.0\n+ *\n+ * Unless required by applicable law or agreed to in writing, software\n+ * distributed under the License is distributed on an "AS IS" BASIS,\n+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n+ * See the License for the specific language governing permissions and\n+ * limitations under the License.\n+ */\n+\n+package com.twitter.storehaus.http\n+\n+import java.util.concurrent.ConcurrentHashMap\n+import java.nio.charset.Charset\n+import org.jboss.netty.buffer.ChannelBuffers\n+import org.jboss.netty.handler.codec.http.{ HttpRequest, HttpResponse, DefaultHttpResponse, HttpResponseStatus, HttpMethod, HttpHeaders }\n+import com.twitter.util.{ Await, Future }\n+import com.twitter.finagle.{ Service, Http, ListeningServer }\n+import com.twitter.storehaus.{ FutureOps, Store }\n+import com.twitter.storehaus.testing.CloseableCleanup\n+import com.twitter.storehaus.testing.generator.NonEmpty\n+import org.scalacheck.{ Arbitrary, Gen, Properties }\n+import org.scalacheck.Prop._\n+\n+object HttpStringStoreProperties extends Properties("HttpStringStore") with CloseableCleanup[ListeningServer] {\n+  def validPairs: Gen[List[(String, Option[String])]] =\n+    NonEmpty.Pairing.alphaStrs().map(_.map{ case (k, v) => ("/" + k, v) })\n+\n+  def baseTest[K: Arbitrary, V: Arbitrary : Equiv](store: Store[K, V], validPairs: Gen[List[(K, Option[V])]])\n+                                                  (put: (Store[K, V], List[(K, Option[V])]) => Unit) =\n+    forAll(validPairs) { (examples: List[(K, Option[V])]) =>\n+      put(store, examples)\n+      examples.toMap.forall { case (k, optV) =>\n+        val res = Await.result(store.get(k))\n+        Equiv[Option[V]].equiv(res, optV)\n+      }\n+    }\n+\n+  def putStoreTest[K: Arbitrary, V: Arbitrary : Equiv](store: Store[K, V], validPairs: Gen[List[(K, Option[V])]]) =\n+    baseTest(store, validPairs) { (s, pairs) =>\n+      pairs.foreach {\n+        case (k, v) =>\n+          Await.result(s.put((k, v)))\n+      }\n+    }\n+\n+  def multiPutStoreTest[K: Arbitrary, V: Arbitrary : Equiv](store: Store[K, V], validPairs: Gen[List[(K, Option[V])]]) =\n+    baseTest(store, validPairs) { (s, pairs) => \n+      Await.result(FutureOps.mapCollect(s.multiPut(pairs.toMap)))\n+    }\n+\n+  def storeTest(store: Store[String, String]) =\n+    putStoreTest(store, validPairs) && multiPutStoreTest(store, validPairs)\n+\n+  val service = new Service[HttpRequest, HttpResponse] {\n+    private val map = new ConcurrentHashMap[String, String]()\n+    private val utf8 = Charset.forName("UTF-8")\n+\n+    def apply(request: HttpRequest): Future[HttpResponse] = {\n+      val response = request.getMethod match {\n+        case HttpMethod.GET =>\n+          Option(map.get(request.getUri)).map{ v =>\n+            val resp = new DefaultHttpResponse(request.getProtocolVersion, HttpResponseStatus.OK)\n+            val content = ChannelBuffers.wrappedBuffer(v.getBytes(utf8))\n+            resp.setContent(content)\n+            resp.headers.set(HttpHeaders.Names.CONTENT_LENGTH, content.readableBytes.toString)\n+            resp\n+          }.getOrElse {\n+            val resp = new DefaultHttpResponse(request.getProtocolVersion, HttpResponseStatus.NOT_FOUND)\n+            resp.headers.set(HttpHeaders.Names.CONTENT_LENGTH, "0")\n+            resp\n+          }\n+        case HttpMethod.DELETE =>\n+          map.remove(request.getUri)\n+          val resp = new DefaultHttpResponse(request.getProtocolVersion, HttpResponseStatus.NO_CONTENT)\n+          resp.headers.set(HttpHeaders.Names.CONTENT_LENGTH, "0")\n+          resp\n+        case HttpMethod.PUT =>\n+          val maybeOldV = Option(map.put(request.getUri, request.getContent.toString(utf8)))\n+          val resp = new DefaultHttpResponse(request.getProtocolVersion, maybeOldV.map(_ => HttpResponseStatus.OK).getOrElse(HttpResponseStatus.CREATED))\n+          resp.setContent(request.getContent)\n+          resp.headers.set(HttpHeaders.Names.CONTENT_LENGTH, request.getContent.readableBytes.toString)\n+          resp\n+        case _ =>\n+          new DefaultHttpResponse(request.getProtocolVersion, HttpResponseStatus.METHOD_NOT_ALLOWED)\n+      }\n+      Future.value(response)\n+    }\n+  }\n+\n+  val server = Http.serve("localhost:0", service)\n+\n+  val store = HttpStringStore(server.boundAddress.toString.substring(1)) // i dont know how else to convert boundAddress into something usable\n+\n+  property("HttpStringStore test") = storeTest(store)\n+\n+  override def closeable = server\n+\n+  override def cleanup() = {\n+    println("closing server")\n+    super.cleanup()\n+  }\n+}\n', 'commit_subject': 'Merge pull request #241 from tresata/feature/storehaus-http', 'repo_name': 'twitter/storehaus', 'committer': {'name': "Ian O'Connell", 'email': 'c25a7357a70d324f8ce9bb0d5e125e2f48d1e339@gmail.com', 'time_sec': '1413852425', 'tz_offset': '-420', 'date': {'seconds': '1413852425'}}, 'commit_tree': 'aece21e49f5317202ff1e2f4ef7a9c9caf14c83f', 'commit_message': 'Merge pull request #241 from tresata/feature/storehaus-http\n\nimplementation of store that uses http protocol', 'language_name': 'Shell', 'license': 'apache-2.0', 'hunks': '@@ -0,0 +1,109 @@\n+/*\n+ * Copyright 2014 Twitter Inc.\n+ *\n+ * Licensed under the Apache License, Version 2.0 (the "License"); you may\n+ * not use this file except in compliance with the License. You may obtain\n+ * a copy of the License at\n+ *\n+ *     http://www.apache.org/licenses/LICENSE-2.0\n+ *\n+ * Unless required by applicable law or agreed to in writing, software\n+ * distributed under the License is distributed on an "AS IS" BASIS,\n+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n+ * See the License for the specific language governing permissions and\n+ * limitations under the License.\n+ */\n+\n+package com.twitter.storehaus.http\n+\n+import java.util.concurrent.ConcurrentHashMap\n+import java.nio.charset.Charset\n+import org.jboss.netty.buffer.ChannelBuffers\n+import org.jboss.netty.handler.codec.http.{ HttpRequest, HttpResponse, DefaultHttpResponse, HttpResponseStatus, HttpMethod, HttpHeaders }\n+import com.twitter.util.{ Await, Future }\n+import com.twitter.finagle.{ Service, Http, ListeningServer }\n+import com.twitter.storehaus.{ FutureOps, Store }\n+import com.twitter.storehaus.testing.CloseableCleanup\n+import com.twitter.storehaus.testing.generator.NonEmpty\n+import org.scalacheck.{ Arbitrary, Gen, Properties }\n+import org.scalacheck.Prop._\n+\n+object HttpStringStoreProperties extends Properties("HttpStringStore") with CloseableCleanup[ListeningServer] {\n+  def validPairs: Gen[List[(String, Option[String])]] =\n+    NonEmpty.Pairing.alphaStrs().map(_.map{ case (k, v) => ("/" + k, v) })\n+\n+  def baseTest[K: Arbitrary, V: Arbitrary : Equiv](store: Store[K, V], validPairs: Gen[List[(K, Option[V])]])\n+                                                  (put: (Store[K, V], List[(K, Option[V])]) => Unit) =\n+    forAll(validPairs) { (examples: List[(K, Option[V])]) =>\n+      put(store, examples)\n+      examples.toMap.forall { case (k, optV) =>\n+        val res = Await.result(store.get(k))\n+        Equiv[Option[V]].equiv(res, optV)\n+      }\n+    }\n+\n+  def putStoreTest[K: Arbitrary, V: Arbitrary : Equiv](store: Store[K, V], validPairs: Gen[List[(K, Option[V])]]) =\n+    baseTest(store, validPairs) { (s, pairs) =>\n+      pairs.foreach {\n+        case (k, v) =>\n+          Await.result(s.put((k, v)))\n+      }\n+    }\n+\n+  def multiPutStoreTest[K: Arbitrary, V: Arbitrary : Equiv](store: Store[K, V], validPairs: Gen[List[(K, Option[V])]]) =\n+    baseTest(store, validPairs) { (s, pairs) => \n+      Await.result(FutureOps.mapCollect(s.multiPut(pairs.toMap)))\n+    }\n+\n+  def storeTest(store: Store[String, String]) =\n+    putStoreTest(store, validPairs) && multiPutStoreTest(store, validPairs)\n+\n+  val service = new Service[HttpRequest, HttpResponse] {\n+    private val map = new ConcurrentHashMap[String, String]()\n+    private val utf8 = Charset.forName("UTF-8")\n+\n+    def apply(request: HttpRequest): Future[HttpResponse] = {\n+      val response = request.getMethod match {\n+        case HttpMethod.GET =>\n+          Option(map.get(request.getUri)).map{ v =>\n+            val resp = new DefaultHttpResponse(request.getProtocolVersion, HttpResponseStatus.OK)\n+            val content = ChannelBuffers.wrappedBuffer(v.getBytes(utf8))\n+            resp.setContent(content)\n+            resp.headers.set(HttpHeaders.Names.CONTENT_LENGTH, content.readableBytes.toString)\n+            resp\n+          }.getOrElse {\n+            val resp = new DefaultHttpResponse(request.getProtocolVersion, HttpResponseStatus.NOT_FOUND)\n+            resp.headers.set(HttpHeaders.Names.CONTENT_LENGTH, "0")\n+            resp\n+          }\n+        case HttpMethod.DELETE =>\n+          map.remove(request.getUri)\n+          val resp = new DefaultHttpResponse(request.getProtocolVersion, HttpResponseStatus.NO_CONTENT)\n+          resp.headers.set(HttpHeaders.Names.CONTENT_LENGTH, "0")\n+          resp\n+        case HttpMethod.PUT =>\n+          val maybeOldV = Option(map.put(request.getUri, request.getContent.toString(utf8)))\n+          val resp = new DefaultHttpResponse(request.getProtocolVersion, maybeOldV.map(_ => HttpResponseStatus.OK).getOrElse(HttpResponseStatus.CREATED))\n+          resp.setContent(request.getContent)\n+          resp.headers.set(HttpHeaders.Names.CONTENT_LENGTH, request.getContent.readableBytes.toString)\n+          resp\n+        case _ =>\n+          new DefaultHttpResponse(request.getProtocolVersion, HttpResponseStatus.METHOD_NOT_ALLOWED)\n+      }\n+      Future.value(response)\n+    }\n+  }\n+\n+  val server = Http.serve("localhost:0", service)\n+\n+  val store = HttpStringStore(server.boundAddress.toString.substring(1)) // i dont know how else to convert boundAddress into something usable\n+\n+  property("HttpStringStore test") = storeTest(store)\n+\n+  override def closeable = server\n+\n+  override def cleanup() = {\n+    println("closing server")\n+    super.cleanup()\n+  }\n+}\n', 'addition_count': 109, 'deletion_count': 0, 'src_file': '/dev/null', 'tgt_file': 'b/storehaus-http/src/test/scala/com/twitter/storehaus/http/HttpStringStoreProperties.scala', 'file_extension': '.scala', 'before_file': 'ADDFILE', 'message': 'Merge pull request #241 from tresata/feature/storehaus-http\n\nimplementation of store that uses http protocol', 'author_name': "Ian O'Connell", 'lang': 'scala'}