pipes-s3 (empty) → 0.1.0.0

raw patch · 4 files changed

+235/−0 lines, 4 filesdep +awsdep +basedep +bytestringsetup-changed

Dependencies added: aws, base, bytestring, http-client, http-client-tls, pipes, pipes-bytestring, pipes-safe, resourcet, text, transformers

Files

LICENSE +30/−0
Setup.hs +2/−0
pipes-s3.cabal +36/−0
src/Pipes/Aws/S3.hs +167/−0

+ LICENSE view

@@ -0,0 +1,30 @@+Copyright (c) 2016, Ben Gamari++All rights reserved.++Redistribution and use in source and binary forms, with or without+modification, are permitted provided that the following conditions are met:++    * Redistributions of source code must retain the above copyright+      notice, this list of conditions and the following disclaimer.++    * Redistributions in binary form must reproduce the above+      copyright notice, this list of conditions and the following+      disclaimer in the documentation and/or other materials provided+      with the distribution.++    * Neither the name of Ben Gamari nor the names of other+      contributors may be used to endorse or promote products derived+      from this software without specific prior written permission.++THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS+"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT+LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR+A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT+OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,+SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT+LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,+DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY+THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

+ Setup.hs view

@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain

+ pipes-s3.cabal view

@@ -0,0 +1,36 @@+name:                pipes-s3+version:             0.1.0.0+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+    Amazon's S3 cloud storage service with the @pipes@ package.+homepage:            http://github.com/bgamari/pipes-s3+license:             BSD3+license-file:        LICENSE+author:              Ben Gamari+maintainer:          ben@smart-cactus.org+copyright:           (c) 2016 Ben Gamari+category:            Network+build-type:          Simple+cabal-version:       >=1.10++source-repository head+  type:      git+  location:  git://github.com/bgamari/pipes-s3++library+  exposed-modules:     Pipes.Aws.S3+  other-extensions:    OverloadedStrings, GeneralizedNewtypeDeriving, ScopedTypeVariables+  hs-source-dirs:      src+  default-language:    Haskell2010+  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,+                       resourcet        >=1.1  && <1.2,+                       aws              >=0.13 && <0.14

+ src/Pipes/Aws/S3.hs view

@@ -0,0 +1,167 @@+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE ScopedTypeVariables #-}++-- | A simple streaming interface to the AWS S3 storage service.+module Pipes.Aws.S3+   ( Bucket(..)+   , Object(..)+     -- * Downloading+   , fromS3+   , fromS3'+     -- ** Convenient re-exports+   , responseBody+     -- * Uploading+   , ChunkSize+   , toS3+   , toS3'+   ) 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)