diff --git a/amazonka-s3-streaming.cabal b/amazonka-s3-streaming.cabal --- a/amazonka-s3-streaming.cabal +++ b/amazonka-s3-streaming.cabal @@ -1,65 +1,76 @@ -name: amazonka-s3-streaming -version: 0.2.0.3 -cabal-version: >=1.10 -build-type: Simple -license: BSD3 -license-file: LICENSE -copyright: Copyright (c) 2016 Commonwealth Scientific and Industrial Research Organisation (CSIRO) -maintainer: Alex.Mason@data61.csiro.au -homepage: https://github.com/Axman6/amazonka-s3-streaming#readme -synopsis: Provides conduits to upload data to S3 using the Multipart API -description: - Please see README.md -category: Network, AWS, Cloud, Distributed Computing -author: Alex Mason -extra-source-files: - README.md - Changelog.md - -source-repository head - type: git - location: https://github.com/Axman6/amazonka-s3-streaming +name: amazonka-s3-streaming +version: 0.2.0.4 +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 + files or bytestrings of known size in parallel. Please see README.md. +homepage: https://github.com/Axman6/amazonka-s3-streaming#readme +license: BSD3 +license-file: LICENSE +author: Alex Mason +maintainer: Alex.Mason@data61.csiro.au +copyright: 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.* library - - if impl(ghc <7.11) - build-depends: - semigroups >=0.12 && <0.19, - transformers >=0.4 && <0.6 - exposed-modules: - Network.AWS.S3.StreamingUpload - build-depends: - base >=4.7 && <5, - amazonka >=1.3.7 && <1.5, - amazonka-core >=1.3.7 && <1.5, - amazonka-s3 >=1.3.7 && <1.5, - resourcet >=1.1.7.4 && <1.2, - conduit >=1.2.6.6 && <1.3, - bytestring >=0.10.6.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, - dlist ==0.8.*, - lifted-async ==0.9.*, - mmap ==0.5.*, - deepseq ==1.4.*, - http-client >=0.4 && <0.6 - default-language: Haskell2010 - hs-source-dirs: src + 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 + , 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 + , dlist >= 0.8 && < 0.9 + , lifted-async >= 0.9 && < 0.10 + , mmap >= 0.5 && < 0.6 + , deepseq >= 1.2 && < 1.5 + , http-client >= 0.4 && < 0.6 -executable s3upload - main-is: Main.hs + if impl(ghc < 7.11) build-depends: - base >=4.7 && <5, - amazonka >=1.4.3 && <1.5, - amazonka-core >=1.4.3 && <1.5, - amazonka-s3 >=1.4.3 && <1.5, - amazonka-s3-streaming >=0.2.0.3 && <0.3, - conduit-extra >=1.1.15 && <1.2, - conduit >=1.2.8 && <1.3, - bytestring >=0.10.8.1 && <0.11, - text >=1.2.2.1 && <1.3 - default-language: Haskell2010 - ghc-options: -threaded -rtsopts -with-rtsopts=-N -O2 + 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 + Description: Whether to build the s3upload executable for uploading files using this library. + Default: False +source-repository head + type: git + location: https://github.com/Axman6/amazonka-s3-streaming + +executable s3upload + main-is: Main.hs + ghc-options: -threaded -rtsopts -with-rtsopts=-N + default-language: Haskell2010 + buildable: False + build-depends: base + , amazonka + , amazonka-core + , amazonka-s3 + , amazonka-s3-streaming + , conduit-extra + , conduit + , bytestring + , text + if flag(s3upload-exe) + buildable: True diff --git a/src/Network/AWS/S3/StreamingUpload.hs b/src/Network/AWS/S3/StreamingUpload.hs --- a/src/Network/AWS/S3/StreamingUpload.hs +++ b/src/Network/AWS/S3/StreamingUpload.hs @@ -46,7 +46,7 @@ import Control.Monad.Morph (lift) import Control.Monad.Reader.Class (local) import Control.Monad.Trans.Resource (MonadBaseControl, - MonadResource, throwM) + MonadResource) import Data.Conduit (Sink, await) import Data.Conduit.List (sourceList) @@ -61,8 +61,9 @@ import Data.List (unfoldr) import Data.List.NonEmpty (nonEmpty) -import Control.Exception.Lens (catching, handling) -import Control.Lens +import Control.Lens (set, view) +import Control.Lens.Operators +import Control.Monad.Catch (onException) import Text.Printf (printf) @@ -168,8 +169,7 @@ when (upr ^. uprsResponseStatus /= 200) $ fail "Failed to upload piece" return upr - catching id (go D.empty 0 hashInit 1 D.empty) $ \e -> - lift (send (abortMultipartUpload bucket key upId)) >> throwM e + go D.empty 0 hashInit 1 D.empty `onException` lift (send (abortMultipartUpload bucket key upId)) -- Whatever happens, we abort the upload and rethrow @@ -219,8 +219,6 @@ let chunkSize' = maybe minimumChunkSize (max minimumChunkSize) mcs in if len `div` chunkSize' >= 10000 then len `div` 9999 else chunkSize' - -- hndlr :: SomeException -> m CompleteMultipartUploadResponse - hndlr e = send (abortMultipartUpload bucket key upId) >> throwM e mgr <- view envManager let mConnCount = mMaxConns mgr nThreads = maybe mConnCount (max 1) mnt @@ -229,18 +227,18 @@ mgr' <- liftIO $ newManager defaultManagerSettings{managerConnCount = nThreads} local (envManager .~ mgr') run else run - exec $ handling id hndlr $ do + 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) + 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) FP fp -> do fsize <- liftIO $ getFileSize fp