postgresql-lo-stream (empty) → 0.1.0.0

raw patch · 5 files changed

+179/−0 lines, 5 filesdep +basedep +bytestringdep +io-streamssetup-changed

Dependencies added: base, bytestring, io-streams, lifted-base, monad-loops, mtl, postgresql-simple

Files

• LICENSE +30/−0
• README.md +3/−0
• Setup.hs +2/−0
• postgresql-lo-stream.cabal +28/−0
• src/Database/PostgreSQL/LargeObjects/Stream.hs +116/−0

@@ -0,0 +1,30 @@+Copyright (c) 2019, Obsidian Systems LLC++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 Obsidian Systems LLC 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.

@@ -0,0 +1,3 @@+# posgresql-lo-stream++This library provides some convenience functions for working with [Large Objects](https://www.postgresql.org/docs/current/largeobjects.html) in PostgreSQL. This is built on top of [postgresql-simple](https://hackage.haskell.org/package/postgresql-simple/docs/Database-PostgreSQL-Simple-LargeObjects.html), which provides the lower-level functions for interacting with Large Objects.

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

@@ -0,0 +1,28 @@+cabal-version:       >=1.10+name:                postgresql-lo-stream+version:             0.1.0.0+synopsis:            Utilities for streaming PostgreSQL LargeObjects+description:         Functions for streaming large objects to and from PostgreSQL+homepage:            https://github.com/obsidiansystems/posgresql-lo-stream+bug-reports:         https://github.com/obsidiansystems/postgresql-lo-stream/issues+license:             BSD3+license-file:        LICENSE+author:              Obsidian Systems LLC+maintainer:          maintainer@obsidian.systems+copyright:           2019 Obsidian Systems LLC+category:            Database+build-type:          Simple+extra-source-files:  README.md++library+  exposed-modules:     Database.PostgreSQL.LargeObjects.Stream+  other-extensions:    LambdaCase+  build-depends:       base >=4.11 && <4.12+                     , bytestring >=0.10 && <0.11+                     , io-streams >=1.5.0 && <1.6+                     , lifted-base >=0.2.2 && <0.2.4+                     , monad-loops >=0.4 && <0.5+                     , mtl >=2.2 && <2.3+                     , postgresql-simple >=0.5 && <0.6+  hs-source-dirs:      src+  default-language:    Haskell2010

@@ -0,0 +1,116 @@+{-# LANGUAGE LambdaCase #-}+module Database.PostgreSQL.LargeObjects.Stream where++import Control.Exception.Lifted (AssertionFailed (..), bracket, throwIO)+import Control.Monad.Loops (whileJust_)+import Control.Monad.State as State+import qualified Data.ByteString as BS+import Data.ByteString.Builder (byteString)+import qualified Data.ByteString.Builder as BS+import qualified Data.ByteString.Lazy as LBS+import Data.IORef (modifyIORef, newIORef, readIORef)+import Data.Semigroup ((<>))+import Database.PostgreSQL.Simple+import Database.PostgreSQL.Simple.LargeObjects (LoFd, Oid (..))+import qualified Database.PostgreSQL.Simple.LargeObjects as Sql+import System.IO (IOMode (ReadMode, WriteMode))+import System.IO.Streams (makeOutputStream)+import qualified System.IO.Streams as Streams++-- | Given a strict ByteString, create a postgres large object and fill it with those contents.+newLargeObjectBS+  :: Connection+  -> BS.ByteString+  -> IO Oid+newLargeObjectBS conn contents = do+  oid <- Sql.loCreat conn+  n <- withLargeObject conn oid WriteMode $ \lofd -> Sql.loWrite conn lofd contents+  let l = BS.length contents+  when (n /= l) . throwIO . AssertionFailed $+    "newLargeObjectBS: loWrite reported writing " <> show n <> " bytes, expected " <> show l <> "."+  return oid++-- | Given a lazy ByteString, create a postgres large object and fill it with those contents.+-- Also returns the total length of the data written.+newLargeObjectLBS+  :: Connection+  -> LBS.ByteString+  -> IO (Oid, Int)+newLargeObjectLBS conn = newLargeObjectStream conn <=< Streams.fromLazyByteString++-- | Create a new large object from an input stream, returning its object id and overall size.+newLargeObjectStream+  :: Connection+  -> Streams.InputStream BS.ByteString+  -> IO (Oid, Int)+newLargeObjectStream conn s = do+  oid <- Sql.loCreat conn+  t <- withLargeObject conn oid WriteMode $ \lofd -> do+    whileJust_ (Streams.read s) $ \chunk -> do+      n <- Sql.loWrite conn lofd chunk+      let l = BS.length chunk+      when (n /= l) . throwIO . AssertionFailed $+        "newLargeObjectStream: loWrite reported writing " <> show n <> " bytes, expected " <> show l <> "."+    Sql.loTell conn lofd+  return (oid, t)++-- | Act on a large object given by id, opening and closing the file descriptor appropriately.+withLargeObject+  :: Connection+  -> Oid+  -> IOMode+  -> (LoFd -> IO a)+  -> IO a+withLargeObject conn oid mode f =+  bracket (Sql.loOpen conn oid mode)+          (\lofd -> Sql.loClose conn lofd)+          f++-- | Stream the contents of a database large object to the given output stream. Useful with Snap's 'addToOutput'.+streamLargeObject+  :: Connection+  -> Oid+  -> Streams.OutputStream BS.Builder+  -> IO ()+streamLargeObject conn oid os =+  withLargeObject conn oid ReadMode $ \lofd ->+    fix $ \again -> do+      chunk <- Sql.loRead conn lofd 8192 -- somewhat arbitrary+      case BS.length chunk of+        0 -> return ()+        _ -> do+          Streams.write (Just $ byteString chunk) os+          again++-- | Stream the contents of a database large object to the given output stream. Useful with Snap's 'addToOutput'.+streamLargeObjectRange+  :: Connection+  -> Oid+  -> Int+  -> Int+  -> Streams.OutputStream BS.Builder+  -> IO ()+streamLargeObjectRange conn oid start end os =+  withLargeObject conn oid ReadMode $ \lofd -> do+    _ <- Sql.loSeek conn lofd Sql.AbsoluteSeek start+    let again n = do+          let nextChunkSize = min 8192 (end - n)+          chunk <- Sql.loRead conn lofd nextChunkSize+          case BS.length chunk of+            0 -> return ()+            k -> do+              Streams.write (Just $ byteString chunk) os+              again (n + k)+    again start++-- | Act on the contents of a LargeObject as a Lazy ByteString+withLargeObjectLBS :: Connection -> Oid -> (LBS.ByteString -> IO ()) -> IO ()+withLargeObjectLBS conn oid f = do+  lo <- newIORef mempty+  cb <- makeOutputStream $ \case+    Just chunk -> modifyIORef lo $ \chunks -> chunks <> chunk+    Nothing -> do+      payload <- readIORef lo+      f $ BS.toLazyByteString payload+  streamLargeObject conn oid cb+  Streams.write Nothing cb