diff --git a/Database/PostgreSQL/Typed/Array.hs b/Database/PostgreSQL/Typed/Array.hs
--- a/Database/PostgreSQL/Typed/Array.hs
+++ b/Database/PostgreSQL/Typed/Array.hs
@@ -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.)
diff --git a/Database/PostgreSQL/Typed/HDBC.hs b/Database/PostgreSQL/Typed/HDBC.hs
--- a/Database/PostgreSQL/Typed/HDBC.hs
+++ b/Database/PostgreSQL/Typed/HDBC.hs
@@ -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"
diff --git a/Database/PostgreSQL/Typed/Protocol.hs b/Database/PostgreSQL/Typed/Protocol.hs
--- a/Database/PostgreSQL/Typed/Protocol.hs
+++ b/Database/PostgreSQL/Typed/Protocol.hs
@@ -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
diff --git a/Database/PostgreSQL/Typed/Range.hs b/Database/PostgreSQL/Typed/Range.hs
--- a/Database/PostgreSQL/Typed/Range.hs
+++ b/Database/PostgreSQL/Typed/Range.hs
@@ -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
diff --git a/Database/PostgreSQL/Typed/Types.hs b/Database/PostgreSQL/Typed/Types.hs
--- a/Database/PostgreSQL/Typed/Types.hs
+++ b/Database/PostgreSQL/Typed/Types.hs
@@ -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
diff --git a/postgresql-typed.cabal b/postgresql-typed.cabal
--- a/postgresql-typed.cabal
+++ b/postgresql-typed.cabal
@@ -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
diff --git a/test/Connect.hs b/test/Connect.hs
--- a/test/Connect.hs
+++ b/test/Connect.hs
@@ -11,7 +11,7 @@
   , pgDBPort = UnixSocket "/tmp/.s.PGSQL.5432"
 #endif
   , pgDBUser = "templatepg"
-  , pgDBDebug = True
+  -- , pgDBDebug = True
   , pgDBParams = [("TimeZone", "UTC")]
   }
 
diff --git a/test/Main.hs b/test/Main.hs
--- a/test/Main.hs
+++ b/test/Main.hs
@@ -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