amazonka-s3-streaming 0.2.0.5 → 1.0.0.0

raw patch · 3 files changed

+174/−216 lines, 3 filesdep +asyncdep −deepseqdep −foundationdep −lifted-asyncdep ~amazonkadep ~amazonka-coredep ~amazonka-s3PVP ok

version bump matches the API change (PVP)

Dependencies added: async

Dependencies removed: deepseq, foundation, lifted-async, mmap, resourcet, semigroups, transformers

Dependency ranges changed: amazonka, amazonka-core, amazonka-s3, bytestring, conduit, exceptions

API changes (from Hackage documentation)

Files

• Main.hs +17/−24
• amazonka-s3-streaming.cabal +17/−31
• src/Network/AWS/S3/StreamingUpload.hs +140/−161

@@ -1,23 +1,23 @@-{-# LANGUAGE CPP               #-}+{-# LANGUAGE CPP #-} {-# LANGUAGE OverloadedStrings #-} module Main where  -import           Data.Conduit                         (($$))-import           Data.Conduit.Binary                  (sourceHandle)-import           Data.Text                            (pack)+import Data.Conduit        ( runConduit, (.|) )+import Data.Conduit.Binary ( sourceHandle )+import Data.Text           ( pack ) -import           Network.AWS-import           Network.AWS.Data.Text                (fromText)-import           Network.AWS.S3.CreateMultipartUpload-import           Network.AWS.S3.StreamingUpload+import Network.AWS+import Network.AWS.Data.Text                ( fromText )+import Network.AWS.S3.CreateMultipartUpload+import Network.AWS.S3.StreamingUpload -import           System.Environment-import           System.IO                            (BufferMode(BlockBuffering),-                                                       hSetBuffering, stdin)+import Control.Monad.IO.Class ( liftIO )+import System.Environment+import System.IO              ( BufferMode(BlockBuffering), hSetBuffering, stdin )  #if !MIN_VERSION_base(4,8,0)-import           Control.Applicative                  (pure, (<$>), (<*>))+import Control.Applicative ( pure, (<$>), (<*>) ) #endif  main :: IO ()@@ -31,17 +31,14 @@                  <*> fromText (pack key)       of         Right (creds,_reg,buck,ky) -> do-#if !MIN_VERSION_amazonka(1,4,4)-          env <- newEnv _reg creds-#else           env <- newEnv creds-#endif           hSetBuffering stdin (BlockBuffering Nothing)           res <- runResourceT . runAWS env $ case file of-                  -- Stream data from stdin-                  "-" -> sourceHandle stdin $$ streamUpload Nothing (createMultipartUpload buck ky)-                  -- Read from a file-                  _   -> concurrentUpload Nothing Nothing (FP file) $ createMultipartUpload buck ky+                  "-" -> runConduit (sourceHandle stdin .| streamUpload Nothing (createMultipartUpload buck ky))+                          >>= liftIO . either print print+                  _   -> concurrentUpload Nothing Nothing (FP file) (createMultipartUpload buck ky)+                          >>= liftIO . print+           print res         Left err -> print err >> usage     ("abort":region:profile:credfile:bucket:_) ->@@ -50,11 +47,7 @@                 <*> fromText (pack bucket)       of         Right (creds,_reg,buck) -> do-#if !MIN_VERSION_amazonka(1,4,4)-          env <- newEnv _reg creds-#else           env <- newEnv creds-#endif           res <- runResourceT . runAWS env . abortAllUploads $ buck           print res         Left err -> print err >> usage

@@ -1,5 +1,5 @@ name:                amazonka-s3-streaming-version:             0.2.0.5+version:             1.0.0.0 synopsis:            Provides conduits to upload data to S3 using the Multipart API description:         Provides a conduit based streaming interface and a concurrent interface to                      uploading data to S3 using the Multipart API. Also provides method to upload@@ -9,50 +9,35 @@ license-file:        LICENSE author:              Alex Mason maintainer:          Alex.Mason@data61.csiro.au-copyright:           Copyright (c) 2016 Commonwealth Scientific and Industrial Research Organisation (CSIRO)+copyright:           Alex Mason, Copyright (c) 2016 Commonwealth Scientific and Industrial Research Organisation (CSIRO) category:            Network, AWS, Cloud, Distributed Computing build-type:          Simple extra-source-files:  README.md, Changelog.md cabal-version:       >=1.10-tested-with:         GHC == 7.8.4, GHC == 7.10.2, GHC == 8.0.2, GHC == 8.2.2, GHC == 8.4.*+tested-with:         GHC == 8.0.* || == 8.2.2 || == 8.4.* || == 8.6.*  library   hs-source-dirs:      src   exposed-modules:     Network.AWS.S3.StreamingUpload   default-language:    Haskell2010   build-depends:       base >= 4.6 && < 5-                       , amazonka         >= 1.3        && < 1.6-                       , amazonka-core    >= 1.3        && < 1.6-                       , amazonka-s3      >= 1.3        && < 1.6-                       , resourcet        >= 1.1.7.4    && < 1.2-                       , conduit          >= 1.2.6.6    && < 1.3+                       , amazonka         >= 1.6        && < 1.7+                       , amazonka-core    >= 1.6        && < 1.7+                       , amazonka-s3      >= 1.6        && < 1.7+                       , conduit          >= 1.3        && < 1.4                        , bytestring       >= 0.10.8.0   && < 0.11                        , mmorph           >= 1.0.6      && < 1.2                        , lens             >= 4.13       && < 5.0                        , mtl              >= 2.2.1      && < 2.3-                       , exceptions       >= 0.8.2.1    && < 0.9+                       , exceptions       >= 0.8.2.1    && < 0.11                        , dlist            >= 0.8        && < 0.9-                       , lifted-async     >= 0.9        && < 0.10-                       , mmap             >= 0.5        && < 0.6-                       , deepseq          >= 1.2        && < 1.5+                       , async            >= 2          && < 2.3                        , http-client      >= 0.4        && < 0.6 -  if impl(ghc < 7.11)-    build-depends:-      semigroups >= 0.12-      , transformers >=0.4-  if impl(ghc == 7.8.*)-    build-depends:-      foundation == 0.0.15-  -  -- -- ini 0.3.4 fails to build because it doesn't have <*-  -- if impl(ghc < 7.9)-  --   build-depends:-  --     ini >= 0.3.5-  -Flag s3upload-exe+flag s3upload-exe   Description: Whether to build the s3upload executable for uploading files using this library.-  Default: False+  default: False+  manual: True  source-repository head   type:     git@@ -62,7 +47,10 @@   main-is: Main.hs   ghc-options: -threaded -rtsopts -with-rtsopts=-N   default-language: Haskell2010-  buildable: False+  if flag(s3upload-exe)+     buildable: True+  else+     buildable: False   build-depends: base                  , amazonka                  , amazonka-core@@ -70,7 +58,5 @@                  , amazonka-s3-streaming                  , conduit-extra                  , conduit-                 , bytestring                  , text-  if flag(s3upload-exe)-    buildable: True+

@@ -1,80 +1,66 @@-{-# LANGUAGE BangPatterns          #-}-{-# LANGUAGE CPP                   #-}-{-# LANGUAGE ConstraintKinds       #-}-{-# LANGUAGE FlexibleContexts      #-}-{-# LANGUAGE MultiParamTypeClasses #-}-{-# LANGUAGE OverloadedStrings     #-}-{-# LANGUAGE ParallelListComp      #-}-{-# LANGUAGE PatternGuards         #-}-{-# LANGUAGE ScopedTypeVariables   #-}+{-# LANGUAGE BangPatterns #-}+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE ParallelListComp #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TupleSections #-}  module Network.AWS.S3.StreamingUpload-(-    streamUpload-    , UploadLocation(..)-    , concurrentUpload-    , abortAllUploads-    , module Network.AWS.S3.CreateMultipartUpload-    , module Network.AWS.S3.CompleteMultipartUpload-    , minimumChunkSize-) where+  ( streamUpload+  , ChunkSize+  , minimumChunkSize+  , NumThreads+  , concurrentUpload+  , UploadLocation(..)+  , abortAllUploads+  , module Network.AWS.S3.CreateMultipartUpload+  , module Network.AWS.S3.CompleteMultipartUpload+  ) where -import           Network.AWS                            (HasEnv(..),-                                                         LogLevel(..), MonadAWS,-                                                         getFileSize,-                                                         hashedBody, send,-                                                         toBody)+import Network.AWS+       ( AWS, HasEnv(..), LogLevel(..), MonadAWS, getFileSize, hashedBody, hashedFileRange,+       liftAWS, runAWS, runResourceT, send, toBody ) -import           Control.Monad.Trans.AWS                (AWSConstraint)-import           Network.AWS.Data.Crypto                (Digest, SHA256,-                                                         hashFinalize, hashInit,-                                                         hashUpdate)+import Network.AWS.Data.Crypto ( Digest, SHA256, hashFinalize, hashInit, hashUpdate ) -import           Network.AWS.S3.AbortMultipartUpload-import           Network.AWS.S3.CompleteMultipartUpload-import           Network.AWS.S3.CreateMultipartUpload-import           Network.AWS.S3.ListMultipartUploads-import           Network.AWS.S3.Types                   (BucketName, cmuParts, completedMultipartUpload,-                                                         completedPart, muKey,-                                                         muUploadId)-import           Network.AWS.S3.UploadPart+import Network.AWS.S3.AbortMultipartUpload+import Network.AWS.S3.CompleteMultipartUpload+import Network.AWS.S3.CreateMultipartUpload+import Network.AWS.S3.ListMultipartUploads+import Network.AWS.S3.Types+       ( BucketName, cmuParts, completedMultipartUpload, completedPart, muKey, muUploadId )+import Network.AWS.S3.UploadPart -import           Control.Applicative-import           Control.Category                       ((>>>))-import           Control.Monad                          (forM_, when, (>=>))-import           Control.Monad.IO.Class                 (MonadIO, liftIO)-import           Control.Monad.Morph                    (lift)-import           Control.Monad.Reader.Class             (local)-import           Control.Monad.Trans.Resource           (MonadBaseControl,-                                                         MonadResource)+import Control.Applicative+import Control.Category           ( (>>>) )+import Control.Monad              ( forM_, when, (>=>) )+import Control.Monad.IO.Class     ( MonadIO, liftIO )+import Control.Monad.Morph        ( lift )+import Control.Monad.Reader.Class ( local ) -import           Data.Conduit                           (Sink, await)-import           Data.Conduit.List                      (sourceList)+import Conduit           ( MonadUnliftIO(..) )+import Data.Conduit      ( ConduitT, Void, await, catchC )+import Data.Conduit.List ( sourceList ) -import           Data.ByteString                        (ByteString)-import qualified Data.ByteString                        as BS-import           Data.ByteString.Builder                (stringUtf8)-import           System.IO.MMap                         (mmapFileByteString)+import           Data.ByteString         ( ByteString )+import qualified Data.ByteString         as BS+import           Data.ByteString.Builder ( stringUtf8 ) -import           Control.DeepSeq                        (rnf)-import qualified Data.DList                             as D-import           Data.List                              (unfoldr)-import           Data.List.NonEmpty                     (nonEmpty)+import qualified Data.DList         as D+import           Data.List          ( unfoldr )+import           Data.List.NonEmpty ( nonEmpty ) -import           Control.Lens                           (set, view)-import           Control.Lens.Operators-import           Control.Monad.Catch                    (onException)+import Control.Lens           ( set, view )+import Control.Lens.Operators -import           Text.Printf                            (printf)+import Text.Printf ( printf ) -import           Control.Concurrent                     (newQSem, signalQSem,-                                                         waitQSem)-import           Control.Concurrent.Async.Lifted        (forConcurrently)-import           System.Mem                             (performGC)+import Control.Concurrent       ( newQSem, signalQSem, waitQSem )+import Control.Concurrent.Async ( forConcurrently )+import Control.Exception.Base   ( SomeException, bracket_ )+import Control.Monad.Catch      ( onException )+import System.Mem               ( performGC ) -import           Network.HTTP.Client                    (defaultManagerSettings,-                                                         managerConnCount,-                                                         newManager)+import Network.HTTP.Client ( defaultManagerSettings, managerConnCount, newManager )  type ChunkSize = Int type NumThreads = Int@@ -86,11 +72,11 @@  {- | Given a 'CreateMultipartUpload', creates a 'Sink' which will sequentially-upload the data streamed in in chunks of at least 'chunkSize' and return-the 'CompleteMultipartUploadResponse'.+upload the data streamed in in chunks of at least 'minimumChunkSize' and return either+the 'CompleteMultipartUploadResponse', or if an exception is thrown,+`AbortMultipartUploadResponse` and the exception as `SomeException`. If aborting+the upload also fails then the exception caused by the call to abort will be thrown. -If uploading of any parts fails, an attempt is made to abort the Multipart-upload, but it is likely that if an upload fails this abort may also fail. 'Network.AWS.S3.ListMultipartUploads' can be used to list any pending uploads - it is important to abort multipart uploads because you will be charged for storage of the parts until it is completed or aborted.@@ -98,51 +84,45 @@  May throw 'Network.AWS.Error' -}-streamUpload :: (MonadResource m, AWSConstraint r m, MonadAWS m)+streamUpload :: (MonadUnliftIO m, MonadAWS m)              => Maybe ChunkSize -- ^ Optional chunk size              -> CreateMultipartUpload -- ^ Upload location-             -> Sink ByteString m CompleteMultipartUploadResponse-streamUpload mcs cmu = do-  logger <- lift $ view envLogger+             -> ConduitT ByteString Void m (Either (AbortMultipartUploadResponse, SomeException) CompleteMultipartUploadResponse)+streamUpload mChunkSize multiPartUploadDesc = do+  logger <- lift $ liftAWS $ view envLogger   let logStr :: MonadIO m => String -> m ()       logStr = liftIO . logger Debug . stringUtf8-      chunkSize = maybe minimumChunkSize (max minimumChunkSize) mcs+      chunkSize = maybe minimumChunkSize (max minimumChunkSize) mChunkSize -  cmur <- lift $ send cmu-  when (cmur ^. cmursResponseStatus /= 200) $+  multiPartUpload <- lift $ send multiPartUploadDesc+  when (multiPartUpload ^. cmursResponseStatus /= 200) $     fail "Failed to create upload"    logStr "\n**** Created upload\n" -  let Just upId = cmur ^. cmursUploadId-      bucket    = cmu  ^. cmuBucket-      key       = cmu  ^. cmuKey-      -- go :: Text -> Builder -> Int -> Int -> Sink ByteString m ()-      go !bss !bufsize !ctx !partnum !completed = Data.Conduit.await >>= \mbs -> case mbs of-        Just bs | l <- BS.length bs-                , bufsize + l <= chunkSize ->-                    go (D.snoc bss bs) (bufsize + l) (hashUpdate ctx bs) partnum completed--                | otherwise -> do-                    rs <- lift $ partUploader partnum (bufsize + BS.length bs)-                                              (hashFinalize $ hashUpdate ctx bs)-                                              (D.snoc bss bs)+  let Just upId = multiPartUpload ^. cmursUploadId+      bucket    = multiPartUploadDesc  ^. cmuBucket+      key       = multiPartUploadDesc  ^. cmuKey+      -- go :: DList ByteString -> Int -> Context SHA256 -> Int -> DList (Maybe CompletedPart) -> Sink ByteString m ()+      go !bss !bufsize !ctx !partnum !completed = await >>= \case+        Just bs+            | l <- BS.length bs, bufsize + l <= chunkSize+                -> go (D.snoc bss bs) (bufsize + l) (hashUpdate ctx bs) partnum completed+            | otherwise -> do+                rs <- lift $ performUpload partnum (bufsize + BS.length bs)+                                            (hashFinalize $ hashUpdate ctx bs)+                                            (D.snoc bss bs) -                    logStr $ printf "\n**** Uploaded part %d size %d\n" partnum bufsize-                    let part = completedPart partnum <$> (rs ^. uprsETag)-#if MIN_VERSION_amazonka_s3(1,4,1)-                        !_ = rnf part-#endif-                    liftIO performGC-                    go empty 0 hashInit (partnum+1) . D.snoc completed $! part+                logStr $ printf "\n**** Uploaded part %d size %d\n" partnum bufsize+                let !part = completedPart partnum <$> (rs ^. uprsETag)+                liftIO performGC+                go empty 0 hashInit (partnum+1) . D.snoc completed $! part          Nothing -> lift $ do             prts <- if bufsize > 0                 then do-                    rs <- partUploader partnum bufsize (hashFinalize ctx) bss-+                    rs <- performUpload partnum bufsize (hashFinalize ctx) bss                     logStr $ printf "\n**** Uploaded (final) part %d size %d\n" partnum bufsize-                     let allParts = D.toList $ D.snoc completed $ completedPart partnum <$> (rs ^. uprsETag)                     pure $ nonEmpty =<< sequence allParts                 else do@@ -153,8 +133,8 @@                     & cMultipartUpload ?~ set cmuParts prts completedMultipartUpload  -      partUploader :: MonadAWS m => Int -> Int -> Digest SHA256 -> D.DList ByteString -> m UploadPartResponse-      partUploader pnum size digest =+      performUpload :: MonadAWS m => Int -> Int -> Digest SHA256 -> D.DList ByteString -> m UploadPartResponse+      performUpload pnum size digest =         D.toList         >>> sourceList         >>> hashedBody digest (fromIntegral size)@@ -168,8 +148,9 @@         when (upr ^. uprsResponseStatus /= 200) $ fail "Failed to upload piece"         return upr -  go D.empty 0 hashInit 1 D.empty `onException` lift (send (abortMultipartUpload bucket key upId))-      -- Whatever happens, we abort the upload and rethrow+  (Right <$> go D.empty 0 hashInit 1 D.empty) `catchC` \(except :: SomeException) ->+    Left . (,except) <$> lift (send (abortMultipartUpload bucket key upId))+      -- Whatever happens, we abort the upload and return the exception   -- | Specifies whether to upload a file or 'ByteString@@ -177,10 +158,6 @@     = FP FilePath -- ^ A file to be uploaded     | BS ByteString -- ^ A strict 'ByteString' - --  IO (Int -> IO (Maybe ByteString)) (Either (IO ()) (IO ()))-        -- part number as input, may be called many times until Nothing is returned,-        -- and a function to close either this part or all parts- {-| Allows a file or 'ByteString' to be uploaded concurrently, using the async library.  The chunk size may optionally be specified, but will be at least@@ -196,78 +173,80 @@ multipart upload on any error, but this may also fail if, for example, the network connection has been broken. See `abortAllUploads` for a crude cleanup method. -}-concurrentUpload :: (AWSConstraint r m, MonadAWS m, MonadBaseControl IO m)+concurrentUpload :: (MonadAWS m)                  => Maybe ChunkSize -- ^ Optional chunk size                  -> Maybe NumThreads -- ^ Optional number of threads to upload with                  -> UploadLocation -- ^ Whether to upload a file on disk or a `ByteString` that's already in memory.                  -> CreateMultipartUpload -- ^ Description of where to upload.                  -> m CompleteMultipartUploadResponse-concurrentUpload mcs mnt ud cmu = do-    cmur <- send cmu-    when (cmur ^. cmursResponseStatus /= 200) $-        fail "Failed to create upload"-    logger <- view envLogger-    let logStr :: MonadIO m => String -> m ()-        logStr = liftIO . logger Info . stringUtf8-    let Just upId = cmur ^. cmursUploadId-        bucket    = cmu  ^. cmuBucket-        key       = cmu  ^. cmuKey+concurrentUpload mChunkSize mNumThreads uploadLoc multiPartUploadDesc = do+  env <- liftAWS $ view environment+  cmur <- send multiPartUploadDesc+  when (cmur ^. cmursResponseStatus /= 200) $+      fail "Failed to create upload" -        calcChunkSize :: Int -> Int-        calcChunkSize len =-            let chunkSize' = maybe minimumChunkSize (max minimumChunkSize) mcs-            in if len `div` chunkSize' >= 10000 then len `div` 9999 else chunkSize'+  let logStr :: MonadIO m => String -> m ()+      logStr    = liftIO . (env ^. envLogger) Info . stringUtf8+      bucket    = multiPartUploadDesc  ^. cmuBucket+      key       = multiPartUploadDesc  ^. cmuKey+      Just upId = cmur ^. cmursUploadId -    let mConnCount = managerConnCount defaultManagerSettings-        nThreads = maybe mConnCount (max 1) mnt-        exec run = if maybe False (> mConnCount) mnt-                then do-                    mgr' <- liftIO $ newManager  defaultManagerSettings{managerConnCount = nThreads}-                    local (envManager .~ mgr') run-                else run-    exec $ flip onException (send (abortMultipartUpload bucket key upId)) $ do-        sem <- liftIO $ newQSem nThreads-        umrs <- case ud of-            BS bs ->-                let chunkSize = calcChunkSize $ BS.length bs-                in forConcurrently (zip [1..] $ chunksOf chunkSize bs) $ \(partnum, b) -> do-                    liftIO $ waitQSem sem-                    logStr $ "Starting part: " ++ show partnum-                    umr <- send . uploadPart bucket key partnum upId . toBody $ b-                    logStr $ "Finished part: " ++ show partnum-                    liftIO $ signalQSem sem-                    pure $ completedPart partnum <$> (umr ^. uprsETag)+      calculateChunkSize :: Int -> Int+      calculateChunkSize len =+          let chunkSize' = maybe minimumChunkSize (max minimumChunkSize) mChunkSize+          in if len `div` chunkSize' >= 10000 then len `div` 9999 else chunkSize' -            FP fp -> do-                fsize <- liftIO $ getFileSize fp-                let chunkSize = calcChunkSize $ fromIntegral fsize-                    (count,lst) = divMod (fromIntegral fsize) chunkSize-                    params = [(partnum, chunkSize*offset, size)-                            | partnum <- [1..]-                            | offset  <- [0..count]-                            | size    <- (chunkSize <$ [0..count-1]) ++ [lst]-                            ]+      mConnCount = managerConnCount defaultManagerSettings+      nThreads   = maybe mConnCount (max 1) mNumThreads -                forConcurrently params $ \(partnum,off,size) -> do-                    liftIO $ waitQSem sem-                    b <- liftIO $ mmapFileByteString fp (Just (fromIntegral off,size))-                    umr <- send . uploadPart bucket key partnum upId . toBody $ b-                    liftIO $ signalQSem sem-                    pure $ completedPart partnum <$> (umr ^. uprsETag)+      exec :: MonadAWS m => AWS a -> m a+      exec act = if maybe False (> mConnCount) mNumThreads+              then do+                  mgr' <- liftIO $ newManager  defaultManagerSettings{managerConnCount = nThreads}+                  liftAWS $ local (envManager .~ mgr') act+              else liftAWS act+  exec $ flip onException (send (abortMultipartUpload bucket key upId)) $ do+      sem <- liftIO $ newQSem nThreads+      uploadResponses <- case uploadLoc of+          BS bytes ->+            let chunkSize = calculateChunkSize $ BS.length bytes+            in liftIO $ forConcurrently (zip [1..] $ chunksOf chunkSize bytes) $ \(partnum, chunk) ->+                bracket_ (waitQSem sem) (signalQSem sem) $ do+                  logStr $ "Starting part: " ++ show partnum+                  umr <- runResourceT $ runAWS env $ send . uploadPart bucket key partnum upId . toBody $ chunk+                  logStr $ "Finished part: " ++ show partnum+                  pure $ completedPart partnum <$> (umr ^. uprsETag) -        let prts = nonEmpty =<< sequence umrs-        send $ completeMultipartUpload bucket key upId-                & cMultipartUpload ?~ set cmuParts prts completedMultipartUpload+          FP filePath -> do+            fsize <- liftIO $ getFileSize filePath+            let chunkSize = calculateChunkSize $ fromIntegral fsize+                (count,lst) = fromIntegral fsize `divMod` chunkSize+                params = [(partnum, chunkSize*offset, size)+                          | partnum <- [1..]+                          | offset  <- [0..count]+                          | size    <- (chunkSize <$ [0..count-1]) ++ [lst]+                          ] +            liftIO $ forConcurrently params $ \(partnum,off,size) ->+              bracket_ (waitQSem sem) (signalQSem sem) $ do+                logStr $ "Starting file part: " ++ show partnum+                chunkStream <- hashedFileRange filePath (fromIntegral off) (fromIntegral size)+                uploadResp <- runResourceT $ runAWS env $+                  send . uploadPart bucket key partnum upId . toBody $ chunkStream+                logStr $ "Finished file part: " ++ show partnum+                pure $ completedPart partnum <$> (uploadResp ^. uprsETag)++      let parts = nonEmpty =<< sequence uploadResponses+      send $ completeMultipartUpload bucket key upId+              & cMultipartUpload ?~ set cmuParts parts completedMultipartUpload+ -- | Aborts all uploads in a given bucket - useful for cleaning up. abortAllUploads :: (MonadAWS m) => BucketName -> m () abortAllUploads bucket = do   rs <- send (listMultipartUploads bucket)   forM_ (rs ^. lmursUploads) $ \mu -> do     let mki = (,) <$> mu ^. muKey <*> mu ^. muUploadId-    case mki of-      Nothing        -> pure ()-      Just (key,uid) -> send (abortMultipartUpload bucket key uid) >> pure ()+    forM_ mki $ \(key,uid) -> send (abortMultipartUpload bucket key uid)