postgresql-typed 0.5.2 → 0.5.3.0

raw patch · 8 files changed

+331/−195 lines, 8 filesPVP: major bump suggested

API removals or changes: PVP suggests a major version bump

API changes (from Hackage documentation)

Files

• Database/PostgreSQL/Typed/Array.hs +1/−1
• Database/PostgreSQL/Typed/HDBC.hs +12/−12
• Database/PostgreSQL/Typed/Protocol.hs +256/−155
• Database/PostgreSQL/Typed/Range.hs +27/−16
• Database/PostgreSQL/Typed/Types.hs +14/−8
• postgresql-typed.cabal +1/−1
• test/Connect.hs +1/−1
• test/Main.hs +19/−1

@@ -54,7 +54,7 @@     (PGArrayType t, PGParameter (PGElemType t) a) => PGParameter t (PGArray a) where   pgEncode ta l = buildPGValue $ BSB.char7 '{' <> mconcat (intersperse (BSB.char7 $ pgArrayDelim ta) $ map el l) <> BSB.char7 '}' where     el Nothing = BSB.string7 "null"-    el (Just e) = pgDQuote (pgArrayDelim ta : "{}") $ pgEncode (pgArrayElementType ta) e+    el (Just e) = pgDQuoteFrom (pgArrayDelim ta : "{}") $ pgEncode (pgArrayElementType ta) e #if __GLASGOW_HASKELL__ >= 710 -- |Allow entirely non-null arrays as parameter inputs only. -- (Only supported on ghc >= 7.10 due to instance overlap.)

@@ -89,7 +89,7 @@   pgv <- takePGConnection pg   reloadTypes Connection     { connectionPG = pgv-    , connectionServerVer = maybe "" BSC.unpack $ pgServerVersion pg+    , connectionServerVer = maybe "" BSC.unpack $ pgServerVersion $ pgTypeEnv pg     , connectionTypes = mempty     , connectionFetchSize = 1     }@@ -139,16 +139,16 @@  getType :: Connection -> PGConnection -> Maybe Bool -> PGColDescription -> ColDesc getType c pg nul PGColDescription{..} = ColDesc-  { colDescName = BSC.unpack colName+  { colDescName = BSC.unpack pgColName   , colDesc = HDBC.SqlColDesc     { HDBC.colType = sqlTypeId t-    , HDBC.colSize = fromIntegral colModifier <$ guard (colModifier >= 0)-    , HDBC.colOctetLength = fromIntegral colSize <$ guard (colSize >= 0)+    , HDBC.colSize = fromIntegral pgColModifier <$ guard (pgColModifier >= 0)+    , HDBC.colOctetLength = fromIntegral pgColSize <$ guard (pgColSize >= 0)     , HDBC.colDecDigits = Nothing     , HDBC.colNullable = nul     }   , colDescDecode = sqlTypeDecode t-  } where t = IntMap.findWithDefault (sqlType (pgTypeEnv pg) $ show colType) (fromIntegral colType) (connectionTypes c)+  } where t = IntMap.findWithDefault (sqlType (pgTypeEnv pg) $ show pgColType) (fromIntegral pgColType) (connectionTypes c)  instance HDBC.IConnection Connection where   disconnect c = withPGConnection c@@ -232,13 +232,13 @@   describeTable c t = withPGConnection c $ \pg ->     map (\[attname, attrelid, attnum, atttypid, attlen, atttypmod, attnotnull] ->       colDescName &&& colDesc $ getType c pg (Just $ not $ pgDecodeRep attnotnull) PGColDescription-        { colName = pgDecodeRep attname-        , colTable = pgDecodeRep attrelid-        , colNumber = pgDecodeRep attnum-        , colType = pgDecodeRep atttypid-        , colSize = pgDecodeRep attlen-        , colModifier = pgDecodeRep atttypmod-        , colBinary = False+        { pgColName = pgDecodeRep attname+        , pgColTable = pgDecodeRep attrelid+        , pgColNumber = pgDecodeRep attnum+        , pgColType = pgDecodeRep atttypid+        , pgColSize = pgDecodeRep attlen+        , pgColModifier = pgDecodeRep atttypmod+        , pgColBinary = False         })       . snd <$> pgSimpleQuery pg (BSLC.fromChunks         [ "SELECT attname, attrelid, attnum, atttypid, attlen, atttypmod, attnotnull"

@@ -1,6 +1,13 @@-{-# LANGUAGE CPP, DeriveDataTypeable, PatternGuards, DataKinds #-}+{-# LANGUAGE CPP #-}+{-# LANGUAGE DataKinds #-}+{-# LANGUAGE DeriveDataTypeable #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE PatternGuards #-}+{-# LANGUAGE TupleSections #-}+{-# LANGUAGE ViewPatterns #-} -- Copyright 2010, 2011, 2012, 2013 Chris Forno--- Copyright 2014-2015 Dylan Simon+-- Copyright 2014-2018 Dylan Simon  -- |The Protocol module allows for direct, low-level communication with a --  PostgreSQL server over TCP/IP. You probably don't want to use this module@@ -14,7 +21,6 @@   , pgErrorCode   , pgConnectionDatabase   , pgTypeEnv-  , pgServerVersion   , pgConnect   , pgDisconnect   , pgReconnect@@ -42,14 +48,18 @@   , PGRowDescription   , pgBind   , pgFetch+  -- * Notifications+  , PGNotification(..)+  , pgGetNotifications+  , pgGetNotification   ) where  #if !MIN_VERSION_base(4,8,0)-import Control.Applicative ((<$>), (<$))+import           Control.Applicative ((<$>), (<$)) #endif-import Control.Arrow ((&&&), first, second)-import Control.Exception (Exception, throwIO, onException)-import Control.Monad (void, liftM2, replicateM, when, unless)+import           Control.Arrow ((&&&), first, second)+import           Control.Exception (Exception, throwIO, onException, finally)+import           Control.Monad (void, liftM2, replicateM, when, unless) #ifdef VERSION_cryptonite import qualified Crypto.Hash as Hash import qualified Data.ByteArray.Encoding as BA@@ -58,36 +68,37 @@ import qualified Data.ByteString as BS import qualified Data.ByteString.Builder as B import qualified Data.ByteString.Char8 as BSC-import Data.ByteString.Internal (w2c)+import           Data.ByteString.Internal (w2c) import qualified Data.ByteString.Lazy as BSL import qualified Data.ByteString.Lazy.Char8 as BSLC-import Data.ByteString.Lazy.Internal (smallChunkSize)+import           Data.ByteString.Lazy.Internal (smallChunkSize) import qualified Data.Foldable as Fold-import Data.IORef (IORef, newIORef, writeIORef, readIORef, atomicModifyIORef, atomicModifyIORef', modifyIORef, modifyIORef')-import Data.Int (Int32, Int16)+import           Data.IORef (IORef, newIORef, writeIORef, readIORef, atomicModifyIORef, atomicModifyIORef', modifyIORef')+import           Data.Int (Int32, Int16) import qualified Data.Map.Lazy as Map-import Data.Maybe (fromMaybe)-import Data.Monoid ((<>))+import           Data.Maybe (fromMaybe)+import           Data.Monoid ((<>)) #if !MIN_VERSION_base(4,8,0)-import Data.Monoid (mempty)+import           Data.Monoid (mempty) #endif-import Data.Tuple (swap)-import Data.Typeable (Typeable)+import           Data.Tuple (swap)+import           Data.Typeable (Typeable) #if !MIN_VERSION_base(4,8,0)-import Data.Word (Word)+import           Data.Word (Word) #endif-import Data.Word (Word32)-import Network (HostName, PortID(..), connectTo)-import System.IO (Handle, hFlush, hClose, stderr, hPutStrLn, hSetBuffering, BufferMode(BlockBuffering))-import System.IO.Unsafe (unsafeInterleaveIO)-import Text.Read (readMaybe)+import           Data.Word (Word32)+import           Network (HostName, PortID(..), connectTo)+import           System.IO (Handle, hFlush, hClose, stderr, hPutStrLn, hSetBuffering, BufferMode(BlockBuffering))+import           System.IO.Error (mkIOError, eofErrorType, ioError)+import           System.IO.Unsafe (unsafeInterleaveIO)+import           Text.Read (readMaybe)  import Database.PostgreSQL.Typed.Types import Database.PostgreSQL.Typed.Dynamic  data PGState   = StateUnsync -- no Sync-  | StatePending -- Sync sent+  | StatePending -- expecting ReadyForQuery   -- ReadyForQuery received:   | StateIdle   | StateTransaction@@ -124,28 +135,52 @@   , connDatabase :: !PGDatabase   , connPid :: !Word32 -- unused   , connKey :: !Word32 -- unused-  , connParameters :: Map.Map BS.ByteString BS.ByteString   , connTypeEnv :: PGTypeEnv+  , connParameters :: IORef (Map.Map BS.ByteString BS.ByteString)   , connPreparedStatementCount :: IORef Integer   , connPreparedStatementMap :: IORef (Map.Map (BS.ByteString, [OID]) PGPreparedStatement)   , connState :: IORef PGState   , connInput :: IORef (G.Decoder PGBackendMessage)   , connTransaction :: IORef Word+  , connNotifications :: IORef (Queue PGNotification)   }  data PGColDescription = PGColDescription-  { colName :: BS.ByteString-  , colTable :: !OID-  , colNumber :: !Int16-  , colType :: !OID-  , colSize :: !Int16-  , colModifier :: !Int32-  , colBinary :: !Bool+  { pgColName :: BS.ByteString+  , pgColTable :: !OID+  , pgColNumber :: !Int16+  , pgColType :: !OID+  , pgColSize :: !Int16+  , pgColModifier :: !Int32+  , pgColBinary :: !Bool   } deriving (Show) type PGRowDescription = [PGColDescription]  type MessageFields = Map.Map Char BS.ByteString +data PGNotification = PGNotification+  { pgNotificationPid :: !Word32+  , pgNotificationChannel :: !BS.ByteString+  , pgNotificationPayload :: BSL.ByteString+  } deriving (Show)++-- |Simple amortized fifo+data Queue a = Queue [a] [a]++emptyQueue :: Queue a+emptyQueue = Queue [] []++enQueue :: a -> Queue a -> Queue a+enQueue a (Queue e d) = Queue (a:e) d++deQueue :: Queue a -> (Queue a, Maybe a)+deQueue (Queue e (x:d)) = (Queue e d, Just x)+deQueue (Queue (reverse -> x:d) []) = (Queue [] d, Just x)+deQueue q = (q, Nothing)++queueToList :: Queue a -> [a]+queueToList (Queue e d) = d ++ reverse e+ -- |PGFrontendMessage represents a PostgreSQL protocol message that we'll send. -- See <http://www.postgresql.org/docs/current/interactive/protocol-message-formats.html>. data PGFrontendMessage@@ -190,6 +225,7 @@   | ErrorResponse { messageFields :: MessageFields }   | NoData   | NoticeResponse { messageFields :: MessageFields }+  | NotificationResponse PGNotification   -- |A ParameterDescription describes the type of a given SQL   --  query/statement parameter ($1, $2, etc.). Unfortunately,   --  PostgreSQL does not give us nullability information for the@@ -207,7 +243,7 @@  -- |PGException is thrown upon encountering an 'ErrorResponse' with severity of --  ERROR, FATAL, or PANIC. It holds the message of the error.-data PGError = PGError { pgErrorFields :: MessageFields }+newtype PGError = PGError { pgErrorFields :: MessageFields }   deriving (Typeable)  instance Show PGError where@@ -237,7 +273,16 @@ -- |A database connection with sane defaults: -- localhost:5432:postgres defaultPGDatabase :: PGDatabase-defaultPGDatabase = PGDatabase "localhost" (PortNumber 5432) (BSC.pack "postgres") (BSC.pack "postgres") BS.empty [] False defaultLogMessage+defaultPGDatabase = PGDatabase+  { pgDBHost = "localhost"+  , pgDBPort = PortNumber 5432+  , pgDBName = "postgres"+  , pgDBUser = "postgres"+  , pgDBPass = BS.empty+  , pgDBParams = []+  , pgDBDebug = False+  , pgDBLogMessage = defaultLogMessage+  }  connDebug :: PGConnection -> Bool connDebug = pgDBDebug . connDatabase@@ -253,10 +298,6 @@ pgTypeEnv :: PGConnection -> PGTypeEnv pgTypeEnv = connTypeEnv --- |Retrieve the \"server_version\" parameter from the connection, if any.-pgServerVersion :: PGConnection -> Maybe BS.ByteString-pgServerVersion PGConnection{ connParameters = p } = Map.lookup (BSC.pack "server_version") p- #ifdef VERSION_cryptonite md5 :: BS.ByteString -> BS.ByteString md5 = BA.convertToBase BA.Base16 . (Hash.hash :: BS.ByteString -> Hash.Digest Hash.MD5)@@ -272,7 +313,7 @@ lazyByteStringNul :: BSL.ByteString -> B.Builder lazyByteStringNul s = B.lazyByteString s <> nul --- |Given a message, determin the (optional) type ID and the body+-- |Given a message, determine the (optional) type ID and the body messageBody :: PGFrontendMessage -> (Maybe Char, B.Builder) messageBody (StartupMessage kv) = (Nothing, B.word32BE 0x30000   <> Fold.foldMap (\(k, v) -> byteStringNul k <> byteStringNul v) kv <> nul)@@ -326,6 +367,7 @@   (t, b) = second (BSL.toStrict . B.toLazyByteString) $ messageBody msg   state _ StateClosed = StateClosed   state Sync _ = StatePending+  state SimpleQuery{} _ = StatePending   state Terminate _ = StateClosed   state _ _ = StateUnsync @@ -363,13 +405,13 @@     tmod <- G.getWord32be -- type modifier     fmt <- G.getWord16be -- format code     return $ PGColDescription-      { colName = name-      , colTable = oid-      , colNumber = fromIntegral col-      , colType = typ'-      , colSize = fromIntegral siz-      , colModifier = fromIntegral tmod-      , colBinary = toEnum (fromIntegral fmt)+      { pgColName = name+      , pgColTable = oid+      , pgColNumber = fromIntegral col+      , pgColType = typ'+      , pgColSize = fromIntegral siz+      , pgColModifier = fromIntegral tmod+      , pgColBinary = toEnum (fromIntegral fmt)       } getMessageBody 'Z' = ReadyForQuery <$> (rs . w2c =<< G.getWord8) where   rs 'I' = return StateIdle@@ -393,65 +435,123 @@ getMessageBody 'n' = return NoData getMessageBody 's' = return PortalSuspended getMessageBody 'N' = NoticeResponse <$> getMessageFields+getMessageBody 'A' = NotificationResponse <$> do+  PGNotification+    <$> G.getWord32be+    <*> getByteStringNul+    <*> G.getLazyByteStringNul getMessageBody t = fail $ "pgGetMessage: unknown message type: " ++ show t  getMessage :: G.Decoder PGBackendMessage getMessage = G.runGetIncremental $ do   typ <- G.getWord8-  s <- G.bytesRead   len <- G.getWord32be-  msg <- getMessageBody (w2c typ)-  e <- G.bytesRead-  let r = fromIntegral len - fromIntegral (e - s)-  when (r > 0) $ G.skip r-  when (r < 0) $ fail "pgReceive: decoder overran message"-  return msg+  G.isolate (fromIntegral len - 4) $ getMessageBody (w2c typ) -pgRecv :: Bool -> PGConnection -> IO (Maybe PGBackendMessage)-pgRecv block c@PGConnection{ connHandle = h, connInput = dr, connState = sr } =-  go =<< readIORef dr where+class Show m => RecvMsg m where+  -- |Read from connection, returning immediate value or non-empty data+  recvMsgData :: PGConnection -> IO (Either m BS.ByteString)+  recvMsgData c = do+    r <- BS.hGetSome (connHandle c) smallChunkSize+    if BS.null r+      then do+        writeIORef (connState c) StateClosed+        hClose (connHandle c)+        -- Should this instead be a special PGError?+        ioError $ mkIOError eofErrorType "PGConnection" (Just (connHandle c)) Nothing+      else+        return (Right r)+  -- |Expected ReadyForQuery message+  recvMsgSync :: Maybe m+  recvMsgSync = Nothing+  -- |NotificationResponse message+  recvMsgNotif :: PGConnection -> PGNotification -> IO (Maybe m)+  recvMsgNotif c n = Nothing <$+    modifyIORef' (connNotifications c) (enQueue n)+  -- |ErrorResponse message+  recvMsgErr :: PGConnection -> MessageFields -> IO (Maybe m)+  recvMsgErr c m = Nothing <$+    connLogMessage c m+  -- |Any other unhandled message+  recvMsg :: PGConnection -> PGBackendMessage -> IO (Maybe m)+  recvMsg c m = Nothing <$ +    connLogMessage c (makeMessage (BSC.pack $ "Unexpected server message: " ++ show m) "Each statement should only contain a single query")++-- |Process all pending messages+data RecvNonBlock = RecvNonBlock deriving (Show)+instance RecvMsg RecvNonBlock where+  recvMsgData c = do+    r <- BS.hGetNonBlocking (connHandle c) smallChunkSize+    if BS.null r+      then return (Left RecvNonBlock)+      else return (Right r)++-- |Wait for ReadyForQuery+data RecvSync = RecvSync deriving (Show)+instance RecvMsg RecvSync where+  recvMsgSync = Just RecvSync++-- |Wait for NotificationResponse+instance RecvMsg PGNotification where+  recvMsgNotif _ = return . Just++-- |Return any message (throwing errors)+instance RecvMsg PGBackendMessage where+  recvMsgErr _ = throwIO . PGError+  recvMsg _ = return . Just++-- |Return any message or ReadyForQuery+instance RecvMsg (Either PGBackendMessage RecvSync) where+  recvMsgSync = Just $ Right RecvSync+  recvMsgErr _ = throwIO . PGError+  recvMsg _ = return . Just . Left++-- |Receive the next message from PostgreSQL (low-level).+pgRecv :: RecvMsg m => PGConnection -> IO m+pgRecv c@PGConnection{ connInput = dr, connState = sr } =+  rcv =<< readIORef dr where   next = writeIORef dr   new = G.pushChunk getMessage-  go (G.Done b _ m) = do++  -- read and parse+  rcv (G.Done b _ m) = do     when (connDebug c) $ putStrLn $ "< " ++ show m     got (new b) m-  go (G.Fail _ _ r) = next (new BS.empty) >> fail r -- not clear how can recover-  go d@(G.Partial r) = do-    b <- (if block then BS.hGetSome else BS.hGetNonBlocking) h smallChunkSize-    if BS.null b-      then Nothing <$ next d-      else go $ r (Just b)-  got :: G.Decoder PGBackendMessage -> PGBackendMessage -> IO (Maybe PGBackendMessage)-  got d (NoticeResponse m) = connLogMessage c m >> go d-  got d m@(ReadyForQuery s) = do-    s' <- atomicModifyIORef' sr ((,) s)-    if s == s'-      then go d-      else done d m-  got d m@(ErrorResponse _) = writeIORef sr StateUnsync >> done d m-  got d m = done d m-  done d m = Just m <$ next d+  rcv (G.Fail _ _ r) = next (new BS.empty) >> fail r -- not clear how can recover+  rcv d@(G.Partial r) = recvMsgData c `onException` next d >>=+    either (<$ next d) (rcv . r . Just) --- |Receive the next message from PostgreSQL (low-level). Note that this will--- block until it gets a message.-pgReceive :: PGConnection -> IO PGBackendMessage-pgReceive c = do-  r <- pgRecv True c-  case r of-    Nothing -> do-      writeIORef (connState c) StateClosed-      fail $ "pgReceive: connection closed"-    Just ErrorResponse{ messageFields = m } -> throwIO (PGError m)-    Just m -> return m+  -- process message+  msg (ParameterStatus k v) = Nothing <$+    modifyIORef' (connParameters c) (Map.insert k v)+  msg (NoticeResponse m) = Nothing <$+    connLogMessage c m+  msg (ErrorResponse m) =+    recvMsgErr c m+  msg m@(ReadyForQuery s) = do+    s' <- atomicModifyIORef' sr (s, )+    if s' == StatePending+      then return recvMsgSync -- expected+      else recvMsg c m -- unexpected+  msg (NotificationResponse n) =+    recvMsgNotif c n+  msg m@AuthenticationOk = do+    writeIORef sr StatePending+    recvMsg c m+  msg m = recvMsg c m+  got d m = msg m `onException` next d >>=+    maybe (rcv d) (<$ next d)  -- |Connect to a PostgreSQL server. pgConnect :: PGDatabase -> IO PGConnection pgConnect db = do+  param <- newIORef Map.empty   state <- newIORef StateUnsync   prepc <- newIORef 0   prepm <- newIORef Map.empty   input <- newIORef getMessage   tr <- newIORef 0+  notif <- newIORef emptyQueue   h <- connectTo (pgDBHost db) (pgDBPort db)   hSetBuffering h (BlockBuffering Nothing)   let c = PGConnection@@ -459,53 +559,55 @@         , connDatabase = db         , connPid = 0         , connKey = 0-        , connParameters = Map.empty+        , connParameters = param         , connPreparedStatementCount = prepc         , connPreparedStatementMap = prepm         , connState = state         , connTypeEnv = unknownPGTypeEnv         , connInput = input         , connTransaction = tr+        , connNotifications = notif         }   pgSend c $ StartupMessage $-    [ (BSC.pack "user", pgDBUser db)-    , (BSC.pack "database", pgDBName db)-    , (BSC.pack "client_encoding", BSC.pack "UTF8")-    , (BSC.pack "standard_conforming_strings", BSC.pack "on")-    , (BSC.pack "bytea_output", BSC.pack "hex")-    , (BSC.pack "DateStyle", BSC.pack "ISO, YMD")-    , (BSC.pack "IntervalStyle", BSC.pack "iso_8601")+    [ ("user", pgDBUser db)+    , ("database", pgDBName db)+    , ("client_encoding", "UTF8")+    , ("standard_conforming_strings", "on")+    , ("bytea_output", "hex")+    , ("DateStyle", "ISO, YMD")+    , ("IntervalStyle", "iso_8601")     ] ++ pgDBParams db   pgFlush c   conn c   where-  conn c = pgReceive c >>= msg c-  msg c (ReadyForQuery _) = return c-    { connTypeEnv = PGTypeEnv-      { pgIntegerDatetimes = fmap (BSC.pack "on" ==) $ Map.lookup (BSC.pack "integer_datetimes") (connParameters c)+  conn c = pgRecv c >>= msg c+  msg c (Right RecvSync) = do+    cp <- readIORef (connParameters c)+    return c+      { connTypeEnv = PGTypeEnv+        { pgIntegerDatetimes = fmap ("on" ==) $ Map.lookup "integer_datetimes" cp+        , pgServerVersion = Map.lookup "server_version" cp+        }       }-    }-  msg c (BackendKeyData p k) = conn c{ connPid = p, connKey = k }-  msg c (ParameterStatus k v) = conn c{ connParameters = Map.insert k v $ connParameters c }-  msg c AuthenticationOk = conn c-  msg c AuthenticationCleartextPassword = do+  msg c (Left (BackendKeyData p k)) = conn c{ connPid = p, connKey = k }+  msg c (Left AuthenticationOk) = conn c+  msg c (Left AuthenticationCleartextPassword) = do     pgSend c $ PasswordMessage $ pgDBPass db     pgFlush c     conn c #ifdef VERSION_cryptonite-  msg c (AuthenticationMD5Password salt) = do-    pgSend c $ PasswordMessage $ BSC.pack "md5" `BS.append` md5 (md5 (pgDBPass db <> pgDBUser db) `BS.append` salt)+  msg c (Left (AuthenticationMD5Password salt)) = do+    pgSend c $ PasswordMessage $ "md5" `BS.append` md5 (md5 (pgDBPass db <> pgDBUser db) `BS.append` salt)     pgFlush c     conn c #endif-  msg _ m = fail $ "pgConnect: unexpected response: " ++ show m+  msg _ (Left m) = fail $ "pgConnect: unexpected response: " ++ show m  -- |Disconnect cleanly from the PostgreSQL server. pgDisconnect :: PGConnection -- ^ a handle from 'pgConnect'              -> IO ()-pgDisconnect c@PGConnection{ connHandle = h } = do-  pgSend c Terminate-  hClose h+pgDisconnect c@PGConnection{ connHandle = h } =+  pgSend c Terminate `finally` hClose h  -- |Disconnect cleanly from the PostgreSQL server, but only if it's still connected. pgDisconnectOnce :: PGConnection -- ^ a handle from 'pgConnect'@@ -523,7 +625,7 @@   if cd == d && s /= StateClosed     then return c{ connDatabase = d }     else do-      when (s /= StateClosed) $ pgDisconnect c+      pgDisconnectOnce c       pgConnect d  pgSync :: PGConnection -> IO ()@@ -531,28 +633,16 @@   s <- readIORef sr   case s of     StateClosed -> fail "pgSync: operation on closed connection"-    StatePending -> wait True-    StateUnsync -> wait False+    StatePending -> wait+    StateUnsync -> do+      pgSend c Sync+      pgFlush c+      wait     _ -> return ()   where-  wait s = do-    r <- pgRecv s c-    case r of-      Nothing-        | s -> do-          writeIORef sr StateClosed-          fail $ "pgReceive: connection closed"-        | otherwise -> do-          pgSend c Sync-          pgFlush c-          wait True-      (Just (ErrorResponse{ messageFields = m })) -> do-        connLogMessage c m-        wait s-      (Just (ReadyForQuery _)) -> return ()-      (Just m) -> do-        connLogMessage c $ makeMessage (BSC.pack $ "Unexpected server message: " ++ show m) $ BSC.pack "Each statement should only contain a single query"-        wait s+  wait = do+    RecvSync <- pgRecv c+    return ()      rowDescription :: PGBackendMessage -> PGRowDescription rowDescription (RowDescription d) = d@@ -573,11 +663,11 @@   pgSend h DescribeStatement{ statementName = BS.empty }   pgSend h Sync   pgFlush h-  ParseComplete <- pgReceive h-  ParameterDescription ps <- pgReceive h-  (,) ps <$> (mapM desc . rowDescription =<< pgReceive h)+  ParseComplete <- pgRecv h+  ParameterDescription ps <- pgRecv h+  (,) ps <$> (mapM desc . rowDescription =<< pgRecv h)   where-  desc (PGColDescription{ colName = name, colTable = tab, colNumber = col, colType = typ }) = do+  desc (PGColDescription{ pgColName = name, pgColTable = tab, pgColNumber = col, pgColType = typ }) = do     n <- nullable tab col     return (name, typ, n)   -- We don't get nullability indication from PostgreSQL, at least not directly.@@ -587,7 +677,7 @@     | nulls && oid /= 0 = do       -- In cases where the resulting field is tracable to the column of a       -- table, we can check there.-      (_, r) <- pgPreparedQuery h (BSC.pack "SELECT attnotnull FROM pg_catalog.pg_attribute WHERE attrelid = $1 AND attnum = $2") [26, 21] [pgEncodeRep (oid :: OID), pgEncodeRep (col :: Int16)] []+      (_, r) <- pgPreparedQuery h "SELECT attnotnull FROM pg_catalog.pg_attribute WHERE attrelid = $1 AND attnum = $2" [26, 21] [pgEncodeRep (oid :: OID), pgEncodeRep (col :: Int16)] []       case r of         [[s]] -> return $ not $ pgDecodeRep s         [] -> return True@@ -617,8 +707,8 @@   pgSend h $ SimpleQuery sql   pgFlush h   go start where -  go = (pgReceive h >>=)-  start (RowDescription rd) = go $ row (map colBinary rd) id+  go = (pgRecv h >>=)+  start (RowDescription rd) = go $ row (map pgColBinary rd) id   start (CommandComplete c) = got c []   start EmptyQueryResponse = return (0, [])   start m = fail $ "pgSimpleQuery: unexpected response: " ++ show m@@ -636,13 +726,12 @@   pgSend h $ SimpleQuery sql   pgFlush h   go where-  go = pgReceive h >>= res-  res (RowDescription _) = go-  res (CommandComplete _) = go-  res EmptyQueryResponse = go-  res (DataRow _) = go-  res (ParameterStatus _ _) = go-  res (ReadyForQuery _) = return ()+  go = pgRecv h >>= res+  res (Left (RowDescription _)) = go+  res (Left (CommandComplete _)) = go+  res (Left EmptyQueryResponse) = go+  res (Left (DataRow _)) = go+  res (Right RecvSync) = return ()   res m = fail $ "pgSimpleQueries_: unexpected response: " ++ show m  pgPreparedBind :: PGConnection -> BS.ByteString -> [OID] -> PGValues -> [Bool] -> IO (IO ())@@ -656,9 +745,9 @@     pgSend c Parse{ queryString = BSL.fromStrict sql, statementName = preparedStatementName n, parseTypes = types }   pgSend c Bind{ portalName = BS.empty, statementName = preparedStatementName n, bindParameters = bind, binaryColumns = bc }   let-    go = pgReceive c >>= start+    go = pgRecv c >>= start     start ParseComplete = do-      modifyIORef (connPreparedStatementMap c) $+      modifyIORef' (connPreparedStatementMap c) $         Map.insert key n       go     start BindComplete = return ()@@ -681,7 +770,7 @@   start   go id   where-  go r = pgReceive c >>= row r+  go r = pgRecv c >>= row r   row r (DataRow fs) = go (r . (fixBinary bc fs :))   row r (CommandComplete d) = return (rowsAffected d, r [])   row r EmptyQueryResponse = return (0, r [])@@ -702,7 +791,7 @@     pgSend c Execute{ portalName = BS.empty, executeRows = count }     pgSend c Flush     pgFlush c-  go r = pgReceive c >>= row r+  go r = pgRecv c >>= row r   row r (DataRow fs) = go (r . (fixBinary bc fs :))   row r PortalSuspended = r <$> unsafeInterleaveIO (execute >> go id)   row r (CommandComplete _) = return (r [])@@ -770,7 +859,7 @@   pgSend c Sync   pgFlush c   go where-  go = pgReceive c >>= res+  go = pgRecv c >>= res   res ParseComplete = go   res BindComplete = go   res (DataRow _) = go@@ -787,7 +876,7 @@   pgSend c Parse{ queryString = sql, statementName = preparedStatementName n, parseTypes = types }   pgSend c Sync   pgFlush c-  ParseComplete <- pgReceive c+  ParseComplete <- pgRecv c   return n  -- |Close a previously prepared query.@@ -798,8 +887,8 @@   pgSend c CloseStatement{ statementName = preparedStatementName n }   pgSend c Sync   pgFlush c-  CloseComplete <- pgReceive c-  CloseComplete <- pgReceive c+  CloseComplete <- pgRecv c+  CloseComplete <- pgRecv c   return ()  -- |Bind a prepared statement, and return the row description.@@ -812,12 +901,12 @@   pgSend c DescribePortal{ portalName = sn }   pgSend c Sync   pgFlush c-  CloseComplete <- pgReceive c-  BindComplete <- pgReceive c-  rowDescription <$> pgReceive c+  CloseComplete <- pgRecv c+  BindComplete <- pgRecv c+  rowDescription <$> pgRecv c   where sn = preparedStatementName n --- |Fetch a single row from an executed prepared statement, returning the next N result rows (if any) and number of affected rows when complete.+-- |Fetch some rows from an executed prepared statement, returning the next N result rows (if any) and number of affected rows when complete. pgFetch :: PGConnection -> PGPreparedStatement -> Word32 -- ^Maximum number of rows to return, or 0 for all   -> IO ([PGValues], Maybe Integer) pgFetch c n count = do@@ -826,7 +915,7 @@   pgSend c Sync   pgFlush c   go where-  go = pgReceive c >>= res+  go = pgRecv c >>= res   res (DataRow v) = first (v :) <$> go   res PortalSuspended = return ([], Nothing)   res (CommandComplete d) = do@@ -834,7 +923,19 @@     pgSend c ClosePortal{ portalName = preparedStatementName n }     pgSend c Sync     pgFlush c-    CloseComplete <- pgReceive c+    CloseComplete <- pgRecv c     return ([], Just $ rowsAffected d)   res EmptyQueryResponse = return ([], Just 0)   res m = fail $ "pgFetch: unexpected response: " ++ show m++-- |Retrieve any pending notifications.  Non-blocking.+pgGetNotifications :: PGConnection -> IO [PGNotification]+pgGetNotifications c = do+  RecvNonBlock <- pgRecv c+  queueToList <$> atomicModifyIORef' (connNotifications c) (emptyQueue, )++-- |Retrieve a notifications, blocking if necessary.+pgGetNotification :: PGConnection -> IO PGNotification+pgGetNotification c =+  maybe (pgRecv c) return+   =<< atomicModifyIORef' (connNotifications c) deQueue

@@ -14,17 +14,21 @@ module Database.PostgreSQL.Typed.Range where  #if !MIN_VERSION_base(4,8,0)-import Control.Applicative ((<$>), (<$))+import           Control.Applicative ((<$>), (<$)) #endif-import Control.Monad (guard)+import           Control.Monad (guard) import qualified Data.Attoparsec.ByteString.Char8 as P import qualified Data.ByteString.Builder as BSB import qualified Data.ByteString.Char8 as BSC-import Data.Monoid ((<>))+#if MIN_VERSION_base(4,9,0)+import           Data.Semigroup (Semigroup(..))+#else+import           Data.Monoid ((<>)) #if !MIN_VERSION_base(4,8,0)-import Data.Monoid (Monoid(..))+import           Data.Monoid (Monoid(..)) #endif-import GHC.TypeLits (Symbol)+#endif+import           GHC.TypeLits (Symbol)  import Database.PostgreSQL.Typed.Types @@ -205,18 +209,25 @@ intersect (Range la ua) (Range lb ub) = normalize $ Range (max la lb) (min ua ub) intersect _ _ = Empty +-- |Union ranges.  Fails if ranges are disjoint.+union :: Ord a => Range a -> Range a -> Range a+union Empty r = r+union r Empty = r+union _ra@(Range la ua) _rb@(Range lb ub)+  -- isEmpty _ra = _rb+  -- isEmpty _rb = _ra+  | Bounded False False <- compareBounds lb ua = error "union: disjoint Ranges"+  | Bounded False False <- compareBounds la ub = error "union: disjoint Ranges"+  | otherwise = Range (min la lb) (max ua ub)++#if MIN_VERSION_base(4,9,0)+instance Ord a => Semigroup (Range a) where+  (<>) = union+#endif+ instance Ord a => Monoid (Range a) where   mempty = Empty-  -- |Union ranges.  Fails if ranges are disjoint.-  mappend Empty r = r-  mappend r Empty = r-  mappend _ra@(Range la ua) _rb@(Range lb ub)-    -- isEmpty _ra = _rb-    -- isEmpty _rb = _ra-    | Bounded False False <- compareBounds lb ua = error "mappend: disjoint Ranges"-    | Bounded False False <- compareBounds la ub = error "mappend: disjoint Ranges"-    | otherwise = Range (min la lb) (max ua ub)-+  mappend = union  -- |Class indicating that the first PostgreSQL type is a range of the second. -- This implies 'PGParameter' and 'PGColumn' instances that will work for any type.@@ -235,7 +246,7 @@       <> pc ']' ')' u     where     pb Nothing = mempty-    pb (Just b) = pgDQuote "(),[]" $ pgEncode (pgRangeElementType tr) b+    pb (Just b) = pgDQuoteFrom "(),[]" $ pgEncode (pgRangeElementType tr) b     pc c o b = BSB.char7 $ if boundClosed b then c else o instance (PGRangeType t, PGColumn (PGSubType t) a) => PGColumn t (Range a) where   pgDecode tr a = either (error . ("pgDecode range (" ++) . (++ ("): " ++ BSC.unpack a))) id $ P.parseOnly per a where

@@ -38,6 +38,7 @@   -- * Conversion utilities   , pgQuote   , pgDQuote+  , pgDQuoteFrom   , parsePGDQuote   , buildPGValue   ) where@@ -119,11 +120,13 @@ -- Nothing values represent unknown. data PGTypeEnv = PGTypeEnv   { pgIntegerDatetimes :: Maybe Bool -- ^ If @integer_datetimes@ is @on@; only relevant for binary encoding.+  , pgServerVersion :: Maybe BS.ByteString -- ^ The @server_version@ parameter   } deriving (Show)  unknownPGTypeEnv :: PGTypeEnv unknownPGTypeEnv = PGTypeEnv   { pgIntegerDatetimes = Nothing+  , pgServerVersion = Nothing   }  -- |A PostgreSQL literal identifier, generally corresponding to the \"name\" type (63-byte strings), but as it would be entered in a query, so may include double-quoting for special characters or schema-qualification.@@ -235,17 +238,20 @@ buildPGValue :: BSB.Builder -> BS.ByteString buildPGValue = BSL.toStrict . BSB.toLazyByteString --- |Double-quote a value if it's \"\", \"null\", or contains any whitespace, \'\"\', \'\\\', or the characters given in the first argument.--- Checking all these things may not be worth it.  We could just double-quote everything.-pgDQuote :: [Char] -> BS.ByteString -> BSB.Builder-pgDQuote unsafe s-  | BS.null s || BSC.any (\c -> isSpace c || c == '"' || c == '\\' || c `elem` unsafe) s || BSC.map toLower s == BSC.pack "null" =-    dq <> BSBP.primMapByteStringBounded ec s <> dq-  | otherwise = BSB.byteString s where+-- |Double-quote a value (e.g., as an identifier).+-- Does not properly handle unicode escaping (yet).+pgDQuote :: BS.ByteString -> BSB.Builder+pgDQuote s = dq <> BSBP.primMapByteStringBounded ec s <> dq where   dq = BSB.char7 '"'   ec = BSBP.condB (\c -> c == c2w '"' || c == c2w '\\') bs (BSBP.liftFixedToBounded BSBP.word8)   bs = BSBP.liftFixedToBounded $ ((,) '\\') BSBP.>$< (BSBP.char7 BSBP.>*< BSBP.word8) +-- |Double-quote a value if it's \"\", \"null\", or contains any whitespace, \'\"\', \'\\\', or the characters given in the first argument.+pgDQuoteFrom :: [Char] -> BS.ByteString -> BSB.Builder+pgDQuoteFrom unsafe s+  | BS.null s || BSC.any (\c -> isSpace c || c == '"' || c == '\\' || c `elem` unsafe) s || BSC.map toLower s == BSC.pack "null" = pgDQuote s+  | otherwise = BSB.byteString s+ -- |Parse double-quoted values ala 'pgDQuote'. parsePGDQuote :: Bool -> [Char] -> (BS.ByteString -> Bool) -> P.Parser (Maybe BS.ByteString) parsePGDQuote blank unsafe isnul = (Just <$> q) <> (mnul <$> uq) where@@ -770,7 +776,7 @@ class PGType t => PGRecordType t instance PGRecordType t => PGParameter t PGRecord where   pgEncode _ (PGRecord l) =-    buildPGValue $ BSB.char7 '(' <> mconcat (intersperse (BSB.char7 ',') $ map (maybe mempty (pgDQuote "(),")) l) <> BSB.char7 ')'+    buildPGValue $ BSB.char7 '(' <> mconcat (intersperse (BSB.char7 ',') $ map (maybe mempty (pgDQuoteFrom "(),")) l) <> BSB.char7 ')'   pgLiteral _ (PGRecord l) =     BSC.pack "ROW(" <> BS.intercalate (BSC.singleton ',') (map (maybe (BSC.pack "NULL") pgQuote) l) `BSC.snoc` ')' instance PGRecordType t => PGColumn t PGRecord where

@@ -1,5 +1,5 @@ Name:          postgresql-typed-Version:       0.5.2+Version:       0.5.3.0 Cabal-Version: >= 1.8 License:       BSD3 License-File:  COPYING

@@ -11,7 +11,7 @@   , pgDBPort = UnixSocket "/tmp/.s.PGSQL.5432" #endif   , pgDBUser = "templatepg"-  , pgDBDebug = True+  -- , pgDBDebug = True   , pgDBParams = [("TimeZone", "UTC")]   } 

@@ -3,6 +3,7 @@ --{-# OPTIONS_GHC -ddump-splices #-} module Main (main) where +import Control.Exception (try) import qualified Data.ByteString as BS import qualified Data.ByteString.Char8 as BSC import Data.Char (isDigit, toUpper)@@ -14,13 +15,15 @@ import Test.QuickCheck.Test (isSuccess)  import Database.PostgreSQL.Typed-import Database.PostgreSQL.Typed.Types (OID)+import Database.PostgreSQL.Typed.Types+import Database.PostgreSQL.Typed.Protocol import Database.PostgreSQL.Typed.Array () import qualified Database.PostgreSQL.Typed.Range as Range import Database.PostgreSQL.Typed.Enum import Database.PostgreSQL.Typed.Inet import Database.PostgreSQL.Typed.SQLToken import Database.PostgreSQL.Typed.Relation+import qualified Database.PostgreSQL.Typed.ErrCodes as PGErr  import Connect @@ -155,6 +158,21 @@   ["box"] <- preparedApply c 603   [Just "line"] <- prepared c 628 "line"   ["line"] <- preparedApply c 628++  pgSimpleQueries_ c "LISTEN channame; NOTIFY channame, 'oh hello'; SELECT pg_notify('channame', 'there')"+  PGNotification _ "channame" "oh hello" <- pgGetNotification c+  (-1, []) <- pgSimpleQuery c "NOTIFY channame"++  pgTransaction c $ do+    (1, [[PGTextValue "1"]]) <- pgSimpleQuery c "SELECT 1"+    (-1, []) <- pgSimpleQuery c "NOTIFY channame, 'nope'"+    Left e1 <- try $ pgSimpleQuery c "SYNTAX_ERROR"+    assert $ pgErrorCode e1 == PGErr.syntax_error+    Left e2 <- try $ pgSimpleQuery c "SELECT 1"+    assert $ pgErrorCode e2 == PGErr.in_failed_sql_transaction++  [PGNotification _ "channame" "there", PGNotification _ "channame" ""] <- pgGetNotifications c+  [] <- pgGetNotifications c    pgDisconnect c   exitSuccess