Use custom modelled header

Author: simao

src/main/scala/com/advancedtelematic/treehub/http/ObjectResource.scala

@@ -3,7 +3,7 @@ package com.advancedtelematic.treehub.http
 import java.io.File
 
 import akka.http.scaladsl.model.StatusCodes
-import akka.http.scaladsl.server.{Directive0, Directive1, PathMatcher1}
+import akka.http.scaladsl.server._
 import akka.stream.Materializer
 import com.advancedtelematic.data.DataType.ObjectId
 import com.advancedtelematic.libats.data.DataType.Namespace
@@ -36,6 +36,10 @@ class ObjectResource(namespace: Directive1[Namespace],
     usageHandler ! UpdateBandwidth(namespace, usageBytes, objectId)
   }
 
+  import OutOfBandStorageHeader._
+
+  private val outOfBandStorageEnabled = validate(objectStore.outOfBandStorageEnabled, "Out of band storage not enabled")
+
   val route = namespace { ns =>
     path("objects" / PrefixedObjectId) { objectId =>
       head {
@@ -61,11 +65,7 @@ class ObjectResource(namespace: Directive1[Namespace],
         complete(objectStore.completeClientUpload(ns, objectId).map(_ => StatusCodes.NoContent))
       } ~
       (post & hintNamespaceStorage(ns)) {
-        // TODO: abstract header name/header
-        // TODO: Refactor all that repeated objectstore code
-        // TODO: Issue Reject if redirect is disabled, this way will bubble down to accept file anyway
-        // TODO: ^^^ Write test for that
-        (headerValueByName("x-ats-accept-redirect") & parameter("size".as[Long])) { (_, size) =>
+        (outOfBandStorageEnabled & headerValueByType[OutOfBandStorageHeader](()) & parameter("size".as[Long])) { (_, size) =>
           onSuccess(objectStore.storeOutOfBand(ns, objectId, size)) { case UploadAt(uri) =>
             redirect(uri, StatusCodes.Found)
           }

src/main/scala/com/advancedtelematic/treehub/http/OutOfBandStorageHeader.scala

@@ -0,0 +1,26 @@
+package com.advancedtelematic.treehub.http
+
+import akka.http.scaladsl.model.headers._
+
+import scala.util.{Failure, Success, Try}
+
+
+object OutOfBandStorageHeader extends ModeledCustomHeaderCompanion[OutOfBandStorageHeader] {
+  override def name: String = "x-ats-accept-redirect"
+
+  override def parse(value: String): Try[OutOfBandStorageHeader] =
+    if(value == "true")
+      Success(new OutOfBandStorageHeader)
+    else
+      Failure(new IllegalArgumentException(s"invalid value for header $name, valid values: true"))
+}
+
+class OutOfBandStorageHeader extends ModeledCustomHeader[OutOfBandStorageHeader] {
+  override def companion: ModeledCustomHeaderCompanion[OutOfBandStorageHeader] = OutOfBandStorageHeader
+
+  override def value(): String = "true"
+
+  override def renderInRequests(): Boolean = true
+
+  override def renderInResponses(): Boolean = true
+}

src/main/scala/com/advancedtelematic/treehub/object_store/BlobStore.scala

@@ -18,6 +18,8 @@ object BlobStore {
 trait BlobStore {
   def storeStream(namespace: Namespace, id: ObjectId, size: Long, blob: Source[ByteString, _]): Future[Long]
 
+  val supportsOutOfBandStorage: Boolean
+
   def storeOutOfBand(namespace: Namespace, id: ObjectId): Future[OutOfBandStoreResult]
 
   def buildResponse(namespace: Namespace, id: ObjectId): Future[HttpResponse]

src/main/scala/com/advancedtelematic/treehub/object_store/LocalFsBlobStore.scala

@@ -90,4 +90,6 @@ class LocalFsBlobStore(root: Path)(implicit ec: ExecutionContext, mat: Materiali
       path
     }
   }
+
+  override val supportsOutOfBandStorage: Boolean = false
 }

src/main/scala/com/advancedtelematic/treehub/object_store/ObjectStore.scala

@@ -15,14 +15,15 @@ import slick.jdbc.MySQLProfile.api._
 
 import scala.concurrent.{ExecutionContext, Future}
 
-// TODO: Use shapeless union type ?
 class ObjectStore(blobStore: BlobStore)(implicit ec: ExecutionContext, db: Database) extends ObjectRepositorySupport {
 
   import scala.async.Async._
 
   def completeClientUpload(namespace: Namespace, id: ObjectId): Future[Unit] =
     objectRepository.setCompleted(namespace, id).map(_ => ())
 
+  def outOfBandStorageEnabled: Boolean = blobStore.supportsOutOfBandStorage
+
   def storeOutOfBand(namespace: Namespace, id: ObjectId, size: Long): Future[OutOfBandStoreResult] = {
     val obj = TObject(namespace, id, size, ObjectStatus.CLIENT_UPLOADING)
     lazy val createF = objectRepository.create(obj)

src/main/scala/com/advancedtelematic/treehub/object_store/S3BlobStore.scala

@@ -141,6 +141,7 @@ class S3BlobStore(s3Credentials: S3Credentials, allowRedirects: Boolean)
   private def objectFilename(namespace: Namespace, objectId: ObjectId): String =
     objectId.path(Paths.get(namespaceDir(namespace))).toString
 
+  override val supportsOutOfBandStorage: Boolean = true
 }
 
 class S3Credentials(accessKey: String, secretKey: String,

src/test/scala/com/advancedtelematic/treehub/http/ObjectResourceIntegrationSpec.scala

@@ -33,7 +33,7 @@ class ObjectResourceIntegrationSpec extends TreeHubSpec with ScalatestRouteTest
     val obj = new ClientTObject()
 
     Post(s"/objects/${obj.prefixedObjectId}?size=${obj.blob.length}", obj.blob)
-      .addHeader(RawHeader("x-ats-accept-redirect", "true")) ~> routes ~> check {
+      .addHeader(new OutOfBandStorageHeader) ~> routes ~> check {
       status shouldBe StatusCodes.Found
       response.header[Location].get.uri.toString() should include("amazonaws.com")
     }
@@ -47,7 +47,7 @@ class ObjectResourceIntegrationSpec extends TreeHubSpec with ScalatestRouteTest
     val obj = new ClientTObject()
 
     val url = Post(s"/objects/${obj.prefixedObjectId}?size=${obj.blob.length}", obj.blob)
-      .addHeader(RawHeader("x-ats-accept-redirect", "true")) ~> routes ~> check {
+      .addHeader(new OutOfBandStorageHeader) ~> routes ~> check {
       status shouldBe StatusCodes.Found
       response.header[Location].get.uri
     }
@@ -71,7 +71,7 @@ class ObjectResourceIntegrationSpec extends TreeHubSpec with ScalatestRouteTest
     val obj = new ClientTObject()
 
     Post(s"/objects/${obj.prefixedObjectId}?size=${obj.blob.length}", obj.blob)
-      .addHeader(RawHeader("x-ats-accept-redirect", "true")) ~> routes ~> check {
+      .addHeader(new OutOfBandStorageHeader) ~> routes ~> check {
       status shouldBe StatusCodes.Found
     }
 

src/test/scala/com/advancedtelematic/treehub/http/ObjectResourceSpec.scala

@@ -5,16 +5,12 @@
 package com.advancedtelematic.treehub.http
 
 import akka.http.scaladsl.model._
-import akka.http.scaladsl.model.headers.RawHeader
+import akka.http.scaladsl.unmarshalling.PredefinedFromEntityUnmarshallers.byteArrayUnmarshaller
 import akka.pattern.ask
 import com.advancedtelematic.treehub.db.ObjectRepositorySupport
 import com.advancedtelematic.util.FakeUsageUpdate.{CurrentBandwith, CurrentStorage}
 import com.advancedtelematic.util.ResourceSpec.ClientTObject
 import com.advancedtelematic.util.{ResourceSpec, TreeHubSpec}
-import akka.http.scaladsl.unmarshalling.PredefinedFromEntityUnmarshallers.byteArrayUnmarshaller
-import de.heikoseeberger.akkahttpcirce.FailFastCirceSupport.unmarshaller
-import com.advancedtelematic.libats.data.ErrorRepresentation
-import com.advancedtelematic.libats.data.ErrorRepresentation._
 
 import scala.concurrent.duration._
 
@@ -59,13 +55,17 @@ class ObjectResourceSpec extends TreeHubSpec with ResourceSpec with ObjectReposi
     }
   }
 
-  test("cannot use redirects when storage method is local storage") {
+  test("request is still accepted and valid when client supports out of band storage but server does not") {
     val obj = new ClientTObject()
 
-    Post(apiUri(s"objects/${obj.prefixedObjectId}?size=${obj.blob.length}"))
-      .addHeader(RawHeader("x-ats-accept-redirect", "true")) ~> routes ~> check {
-      status shouldBe StatusCodes.BadRequest
-      responseAs[ErrorRepresentation].code shouldBe ErrorCodes.OutOfBandStorageNotSupported
+    Post(apiUri(s"objects/${obj.prefixedObjectId}?size=${obj.blob.length}"), obj.blob)
+      .addHeader(new OutOfBandStorageHeader) ~> routes ~> check {
+      status shouldBe StatusCodes.NoContent
+    }
+
+    Get(apiUri(s"objects/${obj.prefixedObjectId}")) ~> routes ~> check {
+      status shouldBe StatusCodes.OK
+      responseAs[Array[Byte]] shouldBe obj.blob
     }
   }