pipes-s3 0.1.0.0 → 0.3.0.2
raw patch · 7 files changed
+533/−163 lines, 7 filesdep +QuickCheckdep +exceptionsdep +http-typesdep ~awsdep ~basedep ~bytestring
Dependencies added: QuickCheck, exceptions, http-types, pipes-s3, tasty, tasty-quickcheck
Dependency ranges changed: aws, base, bytestring, http-client, http-client-tls, pipes, pipes-bytestring, pipes-safe, text
Files
- Test.hs +76/−0
- pipes-s3.cabal +28/−5
- src/Pipes/Aws/S3.hs +12/−158
- src/Pipes/Aws/S3/Download.hs +133/−0
- src/Pipes/Aws/S3/Download/Retry.hs +84/−0
- src/Pipes/Aws/S3/Types.hs +18/−0
- src/Pipes/Aws/S3/Upload.hs +182/−0
+ Test.hs view
@@ -0,0 +1,76 @@+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}++import Pipes+import Pipes.Safe+import qualified Pipes.Aws.S3 as S3+import Pipes.ByteString as PBS+import qualified Data.ByteString as BS+import qualified Data.ByteString.Lazy as BSL+import Test.Tasty+import Test.Tasty.QuickCheck+import Test.QuickCheck+import Test.QuickCheck.Monadic++main :: IO ()+main = defaultMain $ testGroup "tests"+ [ testProperty "round-trip" $ propRoundTrip bucket object+ , testProperty "failures fail" $ propFailure bucket object+ , testProperty "empty uploads fail" $ propEmptyFails bucket object+ ]+ where+ bucket = S3.Bucket "bgamari-test"+ object = S3.Object "test"++newtype ChunkSize = ChunkSize Int+ deriving (Show, Enum)++megabyte = 1024*1024++instance Bounded ChunkSize where+ minBound = ChunkSize $ 5*megabyte+ maxBound = ChunkSize $ 2*1024*megabyte++instance Arbitrary ChunkSize where+ arbitrary = arbitraryBoundedEnum++data Outcome = Succeeds | Fails+ deriving (Enum, Bounded)++instance Arbitrary Outcome where+ arbitrary = arbitraryBoundedEnum++data FailureException = FailureException+ deriving (Show)++instance Exception FailureException++instance Arbitrary BS.ByteString where+ arbitrary = BS.pack . getNonEmpty <$> arbitrary++instance Arbitrary BSL.ByteString where+ arbitrary = BSL.fromChunks . getNonEmpty <$> arbitrary++propEmptyFails :: S3.Bucket -> S3.Object -> ChunkSize -> Property+propEmptyFails bucket object (ChunkSize chunkSize) = monadicIO $ do+ run $ handle checkException $ do+ S3.toS3 chunkSize bucket object (each $ replicate 5 BS.empty)+ fail "empty uploads should fail"+ where+ checkException (S3.EmptyS3UploadError _ _) = return ()++propRoundTrip :: S3.Bucket -> S3.Object -> ChunkSize -> BSL.ByteString -> Property+propRoundTrip bucket object (ChunkSize chunkSize) content = monadicIO $ do+ run $ S3.toS3 chunkSize bucket object (each $ BSL.toChunks content)+ content' <- run $ runSafeT $ PBS.toLazyM $ S3.fromS3 bucket object Nothing+ return $ content == content'++propFailure :: S3.Bucket -> S3.Object -> ChunkSize -> BSL.ByteString -> Property+propFailure bucket object (ChunkSize chunkSize) content = monadicIO $ do+ run $ handle handleFailure $ do+ S3.toS3 chunkSize bucket object (each (BSL.toChunks content) >> throwM FailureException)+ fail "unexpectedly succeeded"+ where+ handleFailure (S3.FailedUploadError {S3.failedUploadException = exc})+ | Just FailureException <- fromException exc = return ()+ | otherwise = fail $ "failed with exception: "++show exc
pipes-s3.cabal view
@@ -1,5 +1,5 @@ name: pipes-s3-version: 0.1.0.0+version: 0.3.0.2 synopsis: A simple interface for streaming data to and from Amazon S3 description: This package provides a simple interface for streaming data to and from@@ -13,6 +13,7 @@ category: Network build-type: Simple cabal-version: >=1.10+tested-with: GHC ==7.8.4, GHC ==7.10.3, GHC ==8.0.1 source-repository head type: git@@ -20,17 +21,39 @@ library exposed-modules: Pipes.Aws.S3+ Pipes.Aws.S3.Download+ Pipes.Aws.S3.Download.Retry+ Pipes.Aws.S3.Upload+ Pipes.Aws.S3.Types other-extensions: OverloadedStrings, GeneralizedNewtypeDeriving, ScopedTypeVariables hs-source-dirs: src default-language: Haskell2010+ ghc-options: -Wall build-depends: base >=4.7 && <4.10, transformers >=0.4 && <0.6, bytestring >=0.10 && <0.11, text >=1.2 && <1.3, pipes-bytestring >=2.1 && <2.2, pipes-safe >=2.2 && <2.3,- pipes >=4.1 && <4.3,- http-client >=0.4 && <0.5,- http-client-tls >=0.2 && <0.3,+ pipes >=4.1 && <4.4,+ http-types >=0.9 && <0.10,+ http-client >=0.4 && <0.6,+ http-client-tls >=0.2 && <0.4, resourcet >=1.1 && <1.2,- aws >=0.13 && <0.14+ aws >=0.13 && <0.16++test-suite pipes-s3-tests+ type: exitcode-stdio-1.0+ main-is: Test.hs+ build-depends: base,+ bytestring,+ text,+ exceptions,+ pipes,+ pipes-safe,+ pipes-bytestring,+ pipes-s3,+ tasty,+ tasty-quickcheck,+ QuickCheck+ default-language: Haskell2010
src/Pipes/Aws/S3.hs view
@@ -1,167 +1,21 @@-{-# LANGUAGE GeneralizedNewtypeDeriving #-} {-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE GADTs #-} -- | A simple streaming interface to the AWS S3 storage service. module Pipes.Aws.S3- ( Bucket(..)+ ( -- * Basic types+ Bucket(..) , Object(..) -- * Downloading- , fromS3- , fromS3'- -- ** Convenient re-exports- , responseBody+ , module Pipes.Aws.S3.Download+ -- ** With retries+ , module Pipes.Aws.S3.Download.Retry+ -- * Uploading- , ChunkSize- , toS3- , toS3'+ , module Pipes.Aws.S3.Upload ) where -import Control.Monad (unless)-import Data.String (IsString)--import qualified Data.ByteString as BS-import Data.ByteString (ByteString)-import qualified Data.Text as T--import Pipes-import Pipes.Safe-import qualified Pipes.Prelude as PP-import qualified Pipes.ByteString as PBS-import Control.Monad.Trans.Resource-import Control.Monad.IO.Class-import Network.HTTP.Client-import Network.HTTP.Client.TLS-import qualified Aws-import qualified Aws.Core as Aws-import qualified Aws.S3 as S3---- | An AWS S3 bucket name-newtype Bucket = Bucket T.Text- deriving (Eq, Ord, Show, Read, IsString)---- | An AWS S3 object name-newtype Object = Object T.Text- deriving (Eq, Ord, Show, Read, IsString)---- | Download an object from S3------ This initiates an S3 download, requiring that the caller provide a way to--- construct a 'Producer' from the initial 'Response' to the request (allowing--- the caller to, e.g., handle failure).------ For instance to merely produced the content of the response,------ @--- 'fromS3' bucket object responseBody--- @----fromS3 :: MonadSafe m- => Bucket -> Object- -> (Response (Producer BS.ByteString m ()) -> Producer BS.ByteString m a)- -> Producer BS.ByteString m a-fromS3 bucket object handler = do- cfg <- liftIO Aws.baseConfiguration- fromS3' cfg bucket object handler---- | Download an object from S3 explicitly specifying an @aws@ 'Aws.Configuration',--- which provides credentials and logging configuration.-fromS3' :: MonadSafe m- => Aws.Configuration -> Bucket -> Object- -> (Response (Producer BS.ByteString m ()) -> Producer BS.ByteString m a)- -> Producer BS.ByteString m a-fromS3' cfg (Bucket bucket) (Object object) handler = do- let s3cfg = Aws.defServiceConfig :: S3.S3Configuration Aws.NormalQuery- mgr <- liftIO $ newManager tlsManagerSettings- req <- liftIO $ buildRequest cfg s3cfg $ S3.getObject bucket object- Pipes.Safe.bracket (liftIO $ responseOpen req mgr) (liftIO . responseClose) $ \resp ->- handler $ resp { responseBody = from $ brRead $ responseBody resp }---- Stolen from pipes-http-withHTTP :: MonadSafe m- => Request- -> Manager- -> (Response (Producer ByteString m ()) -> m a)- -> m a-withHTTP req mgr k =- Pipes.Safe.bracket (liftIO $ responseOpen req mgr) (liftIO . responseClose) k'- where- k' resp = do- let p = (from . brRead . responseBody) resp- k (resp { responseBody = p})--from :: MonadIO m => IO ByteString -> Producer ByteString m ()-from io = go- where- go = do- bs <- liftIO io- unless (BS.null bs) $ do- yield bs- go---buildRequest :: (MonadIO m, Aws.Transaction r a)- => Aws.Configuration- -> Aws.ServiceConfiguration r Aws.NormalQuery- -> r- -> m Request-buildRequest cfg scfg req = do- Just cred <- Aws.loadCredentialsDefault- sigData <- liftIO $ Aws.signatureData Aws.Timestamp cred- let signed = Aws.signQuery req scfg sigData- liftIO $ Aws.queryToHttpRequest signed---- | To maintain healthy streaming uploads are performed in a chunked manner.--- This is the size of the upload chunk size. Due to S3 interface restrictions--- this must be at least five megabytes.-type ChunkSize = Int--type ETag = T.Text-type PartN = Integer---- | Upload content to an S3 object explicitly specifying an @aws@--- 'Aws.Configuration', which provides credentials and logging configuration.-toS3 :: forall m a. MonadIO m- => ChunkSize -> Bucket -> Object- -> Producer BS.ByteString m a- -> m a-toS3 chunkSize bucket object consumer = do- cfg <- Aws.baseConfiguration- toS3' cfg chunkSize bucket object consumer---- | Upload content to an S3 object.------ This internally uses the S3 multi-part upload interface to achieve streaming--- upload behavior.-toS3' :: forall m a. MonadIO m- => Aws.Configuration -> ChunkSize -> Bucket -> Object- -> Producer BS.ByteString m a- -> m a-toS3' cfg chunkSize (Bucket bucket) (Object object) consumer = do- let s3cfg = Aws.defServiceConfig :: S3.S3Configuration Aws.NormalQuery- mgr <- liftIO $ newManager tlsManagerSettings-- resp1 <- liftIO $ runResourceT- $ Aws.pureAws cfg s3cfg mgr- $ S3.postInitiateMultipartUpload bucket object- let uploadId = S3.imurUploadId resp1-- let uploadPart :: (PartN, BS.ByteString) -> m (PartN, ETag)- uploadPart (partN, content) = do- resp <- liftIO $ runResourceT- $ Aws.pureAws cfg s3cfg mgr- $ S3.uploadPart bucket object partN uploadId (RequestBodyBS content)- return (partN, S3.uprETag resp)-- (parts, res) <- PP.toListM' $ consumer- >-> enumFromP 1- >-> PP.mapM uploadPart-- resp2 <- liftIO $ runResourceT- $ Aws.pureAws cfg s3cfg mgr- $ S3.postCompleteMultipartUpload bucket object uploadId parts- return res--enumFromP :: (Monad m, Enum i) => i -> Pipe a (i, a) m r-enumFromP = go- where- go i = await >>= \x -> yield (i, x) >> go (succ i)+import Pipes.Aws.S3.Download+import Pipes.Aws.S3.Download.Retry+import Pipes.Aws.S3.Upload+import Pipes.Aws.S3.Types
+ src/Pipes/Aws/S3/Download.hs view
@@ -0,0 +1,133 @@+-- | @pipes@ 'Producer's for downloading data from AWS S3 objects.+module Pipes.Aws.S3.Download+ ( -- | These may fail with either an 'S3DownloadError' or a 'Aws.S3Error'.+ fromS3+ , fromS3'+ , fromS3WithManager+ , ContentRange(..)+ -- * Error handling+ , S3DownloadError(..)+ ) where++import Control.Monad (unless)+import Control.Exception (Exception)++import qualified Data.ByteString as BS+import Data.ByteString (ByteString)++import Pipes+import Pipes.Safe+import Network.HTTP.Types+import Network.HTTP.Client+import Network.HTTP.Client.TLS+import qualified Aws+import qualified Aws.Core as Aws+import qualified Aws.S3 as S3++import Pipes.Aws.S3.Types++-- | A byte range within an object.+data ContentRange = ContentRange { firstBytePos, lastBytePos :: Int }+ deriving (Eq, Ord, Show)++-- | Thrown when an unknown status code is returned from an S3 download request.+data S3DownloadError = S3DownloadError Bucket Object Status+ deriving (Show)++instance Exception S3DownloadError++-- | Download an object from S3+--+-- Note that this makes no attempt at reusing a 'Manager' and therefore may not+-- be very efficient for many small requests. See 'fromS3WithManager' for more+-- control over the 'Manager' used.+fromS3 :: MonadSafe m+ => Bucket -> Object+ -> Maybe ContentRange+ -- ^ The requested 'ContentRange'. 'Nothing' implies entire object.+ -> Producer BS.ByteString m ()+fromS3 bucket object range = do+ cfg <- liftIO Aws.baseConfiguration+ fromS3' cfg Aws.defServiceConfig bucket object range++-- | Download an object from S3 explicitly specifying an @aws@ 'Aws.Configuration',+-- which provides credentials and logging configuration.+--+-- Note that this makes no attempt at reusing a 'Manager' and therefore may not+-- be very efficient for many small requests. See 'fromS3WithManager' for more+-- control over the 'Manager' used.+--+-- @+-- import qualified Aws.Core as Aws+--+-- getWholeObject :: MonadSafe m => Bucket -> Object -> Producer BS.ByteString m ()+-- getWholeObject bucket object = do+-- cfg <- liftIO 'Aws.baseConfiguration'+-- 'fromS3'' cfg 'Aws.defServiceConfig' bucket object Nothing+-- @+fromS3' :: MonadSafe m+ => Aws.Configuration -- ^ e.g. from 'Aws.baseConfiguration'+ -> S3.S3Configuration Aws.NormalQuery -- ^ e.g. 'Aws.defServiceConfig'+ -> Bucket -> Object+ -> Maybe ContentRange+ -- ^ The requested 'ContentRange'. 'Nothing' implies entire object.+ -> Producer BS.ByteString m ()+fromS3' cfg s3cfg bucket object range = do+ mgr <- liftIO $ newManager tlsManagerSettings+ fromS3WithManager mgr cfg s3cfg bucket object range++-- | Download an object from S3 explicitly specifying an @http-client@ 'Manager'+-- and @aws@ 'Aws.Configuration' (which provides credentials and logging+-- configuration).+--+-- This can be more efficient when submitting many small requests as it allows+-- re-use of the 'Manager' across requests. Note that the 'Manager' provided+-- must support TLS; such a manager can be created with+--+-- @+-- import qualified Aws.Core as Aws+-- import qualified Network.HTTP.Client as HTTP.Client+-- import qualified Network.HTTP.Client.TLS as HTTP.Client.TLS+--+-- getWholeObject :: MonadSafe m => Bucket -> Object -> Producer BS.ByteString m ()+-- getWholeObject bucket object = do+-- cfg <- liftIO 'Aws.baseConfiguration'+-- mgr <- liftIO $ 'newManager' 'HTTP.Client.TLS.tlsManagerSettings'+-- 'fromS3WithManager' mgr cfg 'Aws.defServiceConfig' bucket object Nothing+-- @+fromS3WithManager+ :: MonadSafe m+ => Manager+ -> Aws.Configuration -- ^ e.g. from 'Aws.baseConfiguration'+ -> S3.S3Configuration Aws.NormalQuery -- ^ e.g. 'Aws.defServiceConfig'+ -> Bucket -> Object+ -> Maybe ContentRange+ -- ^ The requested 'ContentRange'. 'Nothing' implies entire object.+ -> Producer BS.ByteString m ()+fromS3WithManager mgr cfg s3cfg (Bucket bucket) (Object object) range = do+ let getObj = (S3.getObject bucket object) { S3.goResponseContentRange = fmap (\(ContentRange a b) -> (a,b)) range }+ req <- liftIO $ buildRequest cfg s3cfg getObj+ Pipes.Safe.bracket (liftIO $ responseOpen req mgr) (liftIO . responseClose) $ \resp ->+ if statusIsSuccessful (responseStatus resp)+ then from $ brRead $ responseBody resp+ else throwM $ S3DownloadError (Bucket bucket) (Object object) (responseStatus resp)++from :: MonadIO m => IO ByteString -> Producer ByteString m ()+from io = go+ where+ go = do+ bs <- liftIO io+ unless (BS.null bs) $ do+ yield bs+ go++buildRequest :: (MonadIO m, Aws.Transaction r a)+ => Aws.Configuration+ -> Aws.ServiceConfiguration r Aws.NormalQuery+ -> r+ -> m Request+buildRequest cfg scfg req = do+ let cred = Aws.credentials cfg+ sigData <- liftIO $ Aws.signatureData Aws.Timestamp cred+ let signed = Aws.signQuery req scfg sigData+ liftIO $ Aws.queryToHttpRequest signed
+ src/Pipes/Aws/S3/Download/Retry.hs view
@@ -0,0 +1,84 @@+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE GADTs #-}+{-# LANGUAGE BangPatterns #-}++-- | @pipes@ 'Producer's from downloading data from AWS S3 objects, with+-- retry-based failure handling.+module Pipes.Aws.S3.Download.Retry+ ( fromS3WithRetries+ -- * Deciding when to retry+ , RetryPolicy(..)+ , retryNTimes+ , retryIfException+ -- * Diagnostics+ , warnOnRetry+ ) where++import Control.Monad (when)+import Data.IORef+import System.IO++import qualified Data.ByteString as BS++import Pipes+import Pipes.Safe+import qualified Pipes.Prelude as PP++import Pipes.Aws.S3.Types+import Pipes.Aws.S3.Download++-- | How many times to attempt an object download before giving up.+data RetryPolicy m = forall s. RetryIf s (Bucket -> Object -> s -> SomeException -> m (s, Bool))++-- | @mempty@ will always retry. @mappend@ will retry only if both+-- 'RetryPolicy's say to retry.+instance Applicative m => Monoid (RetryPolicy m) where+ mempty = RetryIf () (\_ _ s _ -> pure (s, True))+ RetryIf sx0 px `mappend` RetryIf sy0 py =+ RetryIf (sx0, sy0) (\bucket object (sx, sy) exc -> merge <$> px bucket object sx exc <*> py bucket object sy exc)+ where+ merge (sx1, againX) (sy1, againY) = ((sx1, sy1), againX && againY)++-- | Retry a download no more than @n@ times.+retryNTimes :: Applicative m => Int -> RetryPolicy m+retryNTimes n0 = RetryIf n0 shouldRetry+ where+ shouldRetry _ _ !n _ = pure (n-1, n > 0)++-- | Retry if the exception thrown satisfies the given predicate.+retryIfException :: Applicative m => (SomeException -> Bool) -> RetryPolicy m+retryIfException predicate = RetryIf () (\_ _ s exc -> pure (s, predicate exc))++-- | Modify a 'RetryPolicy' to print a warning on stderr when a retry is+-- attempted.+warnOnRetry :: MonadIO m => RetryPolicy m -> RetryPolicy m+warnOnRetry (RetryIf s0 action) =+ RetryIf (0::Int, s0) $ \bucket object (!n,s) exc -> do+ (s', shouldRetry) <- action bucket object s exc+ let msg = show bucket ++ "/" ++ show object ++ ": Retry attempt " ++ show n ++ " failed: " ++ show exc+ when shouldRetry $ liftIO $ hPutStrLn stderr msg+ return ((n+1, s'), shouldRetry)++-- | Download an object from S3, retrying a finite number of times on failure.+fromS3WithRetries :: forall m. (MonadSafe m)+ => RetryPolicy m -> Bucket -> Object+ -> Producer BS.ByteString m ()+fromS3WithRetries (RetryIf retryAcc0 shouldRetry) bucket object = do+ offsetVar <- liftIO $ newIORef 0+ go retryAcc0 offsetVar+ where+ --go :: s -> IORef Int -> Producer BS.ByteString m ()+ go retryAcc offsetVar = do+ offset <- liftIO $ readIORef offsetVar+ handleAll retry $+ fromS3 bucket object (Just $ ContentRange offset maxBound) >-> PP.mapM trackOffset+ where+ retry exc = do+ (retryAcc', again) <- lift $ shouldRetry bucket object retryAcc exc+ if again+ then go retryAcc' offsetVar+ else fail $ "Failed to download "++show bucket++" "++show object++ trackOffset bs = do+ liftIO $ modifyIORef' offsetVar (+ BS.length bs)+ return bs
+ src/Pipes/Aws/S3/Types.hs view
@@ -0,0 +1,18 @@+{-# LANGUAGE GeneralizedNewtypeDeriving #-}++-- | Basic types used throughout @pipes-s3@.+module Pipes.Aws.S3.Types+ ( Bucket(..)+ , Object(..)+ ) where++import Data.String (IsString)+import qualified Data.Text as T++-- | An AWS S3 bucket name+newtype Bucket = Bucket T.Text+ deriving (Eq, Ord, Show, Read, IsString)++-- | An AWS S3 object name+newtype Object = Object T.Text+ deriving (Eq, Ord, Show, Read, IsString)
+ src/Pipes/Aws/S3/Upload.hs view
@@ -0,0 +1,182 @@+{-# LANGUAGE ScopedTypeVariables #-}++-- | @pipes@ utilities for uploading data to AWS S3 objects.+module Pipes.Aws.S3.Upload+ ( -- | These internally use the S3 multi-part upload interface to achieve+ -- streaming upload behavior.+ --+ -- In the case of failure one of two exceptions will be thrown,+ --+ -- - 'EmptyS3UploadError': In the event that the 'Producer' fails to+ -- produce any content to upload+ --+ -- - 'FailedUploadError': In any other case.+ --+ -- The 'FailedUploadError' exception carries the 'UploadId' of the failed+ -- upload as well as the inner exception. Note that while the library makes+ -- an attempt to clean up the parts of the partial upload, there may still+ -- be remnants due to limitations in the @aws@ library.+ --+ toS3+ , toS3'+ , toS3WithManager++ -- * Chunk size+ , ChunkSize+ , defaultChunkSize++ -- * Error handling+ , EmptyS3UploadError(..)+ , FailedUploadError(..)+ , UploadId(..)+ ) where++import Control.Monad (when)+import Control.Exception (Exception)++import qualified Data.ByteString as BS+import qualified Data.Text as T++import Pipes+import Pipes.Safe+import qualified Pipes.Prelude as PP+import qualified Pipes.ByteString as PBS+import Control.Monad.Trans.Resource+import Network.HTTP.Client+import Network.HTTP.Client.TLS+import qualified Aws+import qualified Aws.S3 as S3++import Pipes.Aws.S3.Types++-- | Thrown when an upload with no data is attempted.+data EmptyS3UploadError = EmptyS3UploadError Bucket Object+ deriving (Show)++instance Exception EmptyS3UploadError++-- | An identifier representing an active upload.+newtype UploadId = UploadId T.Text+ deriving (Show, Eq, Ord)++-- | Thrown when an error occurs during an upload.+data FailedUploadError = FailedUploadError { failedUploadBucket :: Bucket+ , failedUploadObject :: Object+ , failedUploadException :: SomeException+ , failedUploadId :: UploadId+ }+ deriving (Show)++instance Exception FailedUploadError++-- | To maintain healthy streaming uploads are performed in a chunked manner.+-- This is the size of the upload chunk size in bytes. Due to S3 interface+-- restrictions this must be at least five megabytes.+type ChunkSize = Int++-- | A reasonable chunk size of 10 megabytes.+defaultChunkSize :: ChunkSize+defaultChunkSize = 10 * 1024 * 1024++type ETag = T.Text+type PartN = Integer++-- | Upload content to an S3 object.+--+-- May throw a 'EmptyS3UploadError' if the producer fails to provide any content.+toS3 :: forall m a. (MonadIO m, MonadCatch m)+ => ChunkSize -> Bucket -> Object+ -> Producer BS.ByteString m a+ -> m a+toS3 chunkSize bucket object consumer = do+ cfg <- Aws.baseConfiguration+ toS3' cfg Aws.defServiceConfig chunkSize bucket object consumer++-- | Upload content to an S3 object, explicitly specifying an+-- 'Aws.Configuration', which provides credentials and logging configuration.+--+-- May throw a 'EmptyS3UploadError' if the producer fails to provide any content.+toS3' :: forall m a. (MonadIO m, MonadCatch m)+ => Aws.Configuration -- ^ e.g. from 'Aws.baseConfiguration'+ -> S3.S3Configuration Aws.NormalQuery -- ^ e.g. 'Aws.defServiceConfig'+ -> ChunkSize -> Bucket -> Object+ -> Producer BS.ByteString m a+ -> m a+toS3' cfg s3cfg chunkSize bucket object consumer = do+ mgr <- liftIO $ newManager tlsManagerSettings+ toS3WithManager mgr cfg s3cfg chunkSize bucket object consumer++-- | Download an object from S3 explicitly specifying an @http-client@ 'Manager'+-- and @aws@ 'Aws.Configuration' (which provides credentials and logging+-- configuration).+--+-- This can be more efficient when submitting many small requests as it allows+-- re-use of the 'Manager' across requests. Note that the 'Manager' provided+-- must support TLS; such a manager can be created with+--+-- @+-- import qualified Aws.Core as Aws+-- import qualified Network.HTTP.Client as HTTP.Client+-- import qualified Network.HTTP.Client.TLS as HTTP.Client.TLS+--+-- putObject :: MonadSafe m => Bucket -> Object -> Producer BS.ByteString m () -> m ()+-- putObject bucket object prod = do+-- cfg <- liftIO 'Aws.baseConfiguration'+-- mgr <- liftIO $ 'newManager' 'HTTP.Client.TLS.tlsManagerSettings'+-- 'toS3WithManager' mgr cfg 'Aws.defServiceConfig' defaultChunkSize bucket object prod+-- @+--+-- May throw a 'EmptyS3UploadError' if the producer fails to provide any content.+toS3WithManager :: forall m a. (MonadIO m, MonadCatch m)+ => Manager+ -> Aws.Configuration -- ^ e.g. from 'Aws.baseConfiguration'+ -> S3.S3Configuration Aws.NormalQuery -- ^ e.g. 'Aws.defServiceConfig'+ -> ChunkSize -> Bucket -> Object+ -> Producer BS.ByteString m a+ -> m a+toS3WithManager mgr cfg s3cfg chunkSize bucket object consumer = do+ let Bucket bucketName = bucket+ Object objectName = object+ resp1 <- liftIO $ runResourceT+ $ Aws.pureAws cfg s3cfg mgr+ $ S3.postInitiateMultipartUpload bucketName objectName+ let uploadId = S3.imurUploadId resp1+ abortUpload err+ -- Otherwise we apparently get a 'Missing root element' error+ -- when aborting.+ | Just (EmptyS3UploadError _ _) <- fromException err = throwM err+ | otherwise = do+ resp <- liftIO $ runResourceT $ Aws.aws cfg s3cfg mgr+ $ S3.postAbortMultipartUpload bucketName objectName uploadId+ case Aws.responseResult resp of+ Left err' -> throwM err'+ Right _ -> throwM $ FailedUploadError bucket object err (UploadId uploadId)++ handleAll abortUpload $ do+ let uploadPart :: (PartN, BS.ByteString) -> m (PartN, ETag)+ uploadPart (partN, content) = do+ resp <- liftIO $ runResourceT+ $ Aws.pureAws cfg s3cfg mgr+ $ S3.uploadPart bucketName objectName+ partN uploadId (RequestBodyBS content)+ return (partN, S3.uprETag resp)++ (parts, res) <- PP.toListM' $ PBS.chunksOf' chunkSize consumer+ >-> PP.filter (not . BS.null)+ >-> enumFromP 1+ >-> PP.mapM uploadPart++ -- We handle this specifically to provide a more sensible error than+ -- "Missing root element"+ when (null parts)+ $ throwM (EmptyS3UploadError bucket object)++ _ <- liftIO $ runResourceT+ $ Aws.pureAws cfg s3cfg mgr+ $ S3.postCompleteMultipartUpload bucketName objectName uploadId parts+ return res++enumFromP :: (Monad m, Enum i) => i -> Pipe a (i, a) m r+enumFromP = go+ where+ go i = await >>= \x -> yield (i, x) >> go (succ i)