diff --git a/Test.hs b/Test.hs new file mode 100644 --- /dev/null +++ b/Test.hs @@ -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 diff --git a/pipes-s3.cabal b/pipes-s3.cabal --- a/pipes-s3.cabal +++ b/pipes-s3.cabal @@ -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 diff --git a/src/Pipes/Aws/S3.hs b/src/Pipes/Aws/S3.hs --- a/src/Pipes/Aws/S3.hs +++ b/src/Pipes/Aws/S3.hs @@ -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 diff --git a/src/Pipes/Aws/S3/Download.hs b/src/Pipes/Aws/S3/Download.hs new file mode 100644 --- /dev/null +++ b/src/Pipes/Aws/S3/Download.hs @@ -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 diff --git a/src/Pipes/Aws/S3/Download/Retry.hs b/src/Pipes/Aws/S3/Download/Retry.hs new file mode 100644 --- /dev/null +++ b/src/Pipes/Aws/S3/Download/Retry.hs @@ -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 diff --git a/src/Pipes/Aws/S3/Types.hs b/src/Pipes/Aws/S3/Types.hs new file mode 100644 --- /dev/null +++ b/src/Pipes/Aws/S3/Types.hs @@ -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) diff --git a/src/Pipes/Aws/S3/Upload.hs b/src/Pipes/Aws/S3/Upload.hs new file mode 100644 --- /dev/null +++ b/src/Pipes/Aws/S3/Upload.hs @@ -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)