WIP: Test stream-asset

Author: supersven

services/cargohold/src/CargoHold/API/Federation.hs

@@ -36,6 +36,7 @@ import Wire.API.Federation.API
 import qualified Wire.API.Federation.API.Cargohold as F
 import qualified CargoHold.S3 as S3
 import Wire.API.Routes.AssetBody
+import Debug.Trace
 
 type FederationAPI = "federation" :> ToServantApi (FedApi 'Cargohold)
 
@@ -54,7 +55,9 @@ checkAsset ga =
 
 streamAsset :: F.GetAsset -> Handler AssetSource
 streamAsset ga = do
+  traceM "streamAsset"
   available <- checkAsset ga
+  traceM $ "available : " ++ show available
   unless available (throwE assetNotFound)
   AssetSource <$> S3.downloadV3 (F.gaKey ga)
 

services/cargohold/src/CargoHold/AWS.hs

@@ -34,6 +34,7 @@ module CargoHold.AWS
     send,
     sendCatch,
     exec,
+    execStream,
     execCatch,
   )
 where
@@ -157,7 +158,7 @@ instance Exception Error
 --------------------------------------------------------------------------------
 -- Utilities
 
-sendCatch :: AWSRequest r => r -> Amazon (Either AWS.Error (Rs r))
+sendCatch :: (MonadCatch m, AWS.MonadAWS m, AWSRequest r) => r -> m (Either AWS.Error (Rs r))
 sendCatch = AWST.trying AWS._Error . AWS.send
 
 send :: AWSRequest r => r -> Amazon (Rs r)
@@ -173,10 +174,10 @@ exec ::
   m (Rs r)
 exec env request = do
   let req = request (_s3Bucket env)
-  resp <- execute env (sendCatch req)
+  resp <- execute env (sendCatch req) 
   case resp of
     Left err -> do
-      Log.info $
+      Logger.info (view logger env) $
         Log.field "remote" (Log.val "S3")
           ~~ Log.msg (show err)
           ~~ Log.msg (show req)
@@ -185,6 +186,25 @@ exec env request = do
       throwM (GeneralError err)
     Right r -> return r
 
+execStream :: 
+  (AWSRequest r, Show r) =>
+  Env ->
+  (Text -> r) ->
+  ResourceT IO (Either Error (Rs r)) 
+execStream env request = do
+  let req = request (_s3Bucket env)
+  resp <- AWS.runAWS (view amazonkaEnv env) (sendCatch req)
+  case resp of
+    Left err -> do
+      Logger.info (view logger env) $
+        Log.field "remote" (Log.val "S3")
+          ~~ Log.msg (show err)
+          ~~ Log.msg (show req)
+      -- We just re-throw the error, but logging it here also gives us the request
+      -- that caused it.
+      pure $ Left (GeneralError err)
+    Right r -> pure $ Right r
+
 execCatch ::
   (AWSRequest r, Show r, MonadLogger m, MonadIO m) =>
   Env ->
@@ -211,3 +231,4 @@ canRetry (Left e) = case e of
 
 retry5x :: (Monad m) => RetryPolicyM m
 retry5x = limitRetries 5 <> exponentialBackoff 100000
+ 

services/cargohold/src/CargoHold/S3.hs

@@ -133,7 +133,8 @@ downloadV3 ::
   V3.AssetKey ->
   ExceptT Error App (ConduitM () ByteString (ResourceT IO) ())
 downloadV3 (s3Key . mkKey -> key) = do
-   _streamBody . view gorsBody <$> exec req 
+   env <- view aws
+   _streamBody . view gorsBody <$> AWS.execStream env req 
   where
     req :: Text -> GetObject
     req b = getObject (BucketName b) (ObjectKey key) 

services/cargohold/test/integration/API/Federation.hs

@@ -30,7 +30,9 @@ tests s =
         ],
       testGroup
         "stream-asset"
-        [test s "streaming large asset" testLargeAsset]
+        [ test s "streaming large asset" testLargeAsset,
+          test s "stream an asset" testStreamAsset
+        ]
     ]
 
 testGetAssetAvailable :: Bool -> TestSignature ()
@@ -140,3 +142,32 @@ testLargeAsset c = do
     -- check that the first chunk is received
     chunk <- responseBody resp
     print chunk
+
+testStreamAsset :: TestSignature ()
+testStreamAsset c = do
+  -- Initial upload
+  let bdy = (applicationOctetStream, "Hello World")
+      settings =
+        defAssetSettings
+          & set setAssetRetention (Just AssetVolatile)
+  uid <- liftIO $ Id <$> nextRandom
+  ast :: Asset <-
+    responseJsonError
+      =<< uploadSimple (c . path "/assets/v3") uid settings bdy
+      <!! const 201 === statusCode
+
+  -- Call get-asset federation API
+  let tok = view assetToken ast
+  let key = view assetKey ast
+  let ga =
+        GetAsset
+          { gaUser = uid,
+            gaToken = tok,
+            gaKey = qUnqualified key
+          }
+  respBody <- fmap responseBody $
+       post (c . path "/federation/stream-asset" . json ga)
+       <!! const 200 === statusCode
+
+
+  liftIO $ respBody @?= Just "Hello World"