pqi-native 0.1.0.0 → 0.1.0.1
raw patch · 9 files changed
+179/−176 lines, 9 filesPVP ok
version bump matches the API change (PVP)
API changes (from Hackage documentation)
Files
- CHANGELOG.md +6/−0
- pqi-native.cabal +1/−6
- src/library/Pqi/Native.hs +43/−43
- src/library/Pqi/Native/Auth.hs +4/−4
- src/library/Pqi/Native/Connection.hs +29/−29
- src/library/Pqi/Native/LargeObject.hs +1/−1
- src/library/Pqi/Native/Query.hs +64/−62
- src/library/Pqi/Native/Types.hs +23/−23
- src/transport/Pqi/Native/Transport.hs +8/−8
CHANGELOG.md view
@@ -1,3 +1,9 @@+# v0.1.0.1++## Fixes++- Adapted to GHC 8.10: pruned unsupported default-extensions (`ApplicativeDo`, `DuplicateRecordFields`, `NoFieldSelectors`, `OverloadedRecordDot`, `TemplateHaskell`) and rewrote the source accordingly+ # v0.1.0.0 ## Breaking
pqi-native.cabal view
@@ -1,6 +1,6 @@ cabal-version: 3.0 name: pqi-native-version: 0.1.0.0+version: 0.1.0.1 category: Database, PostgreSQL synopsis: Native (pure-Haskell) adapter for pqi description:@@ -30,7 +30,6 @@ common base default-language: Haskell2010 default-extensions:- ApplicativeDo BangPatterns BinaryLiterals BlockArguments@@ -43,7 +42,6 @@ DeriveTraversable DerivingStrategies DerivingVia- DuplicateRecordFields EmptyDataDecls ExistentialQuantification FlexibleContexts@@ -57,11 +55,9 @@ MultiParamTypeClasses MultiWayIf NamedFieldPuns- NoFieldSelectors NoImplicitPrelude NoMonomorphismRestriction NumericUnderscores- OverloadedRecordDot OverloadedStrings ParallelListComp PatternGuards@@ -71,7 +67,6 @@ ScopedTypeVariables StandaloneDeriving StrictData- TemplateHaskell TupleSections TypeApplications TypeFamilies
src/library/Pqi/Native.hs view
@@ -48,31 +48,31 @@ mkConnection connection = Pqi.Connection { Pqi.connectPoll = pure Pqi.PollingOk,- Pqi.isNullConnection = connection.isNull,- Pqi.finish = readIORef connection.transport >>= Transport.close,+ Pqi.isNullConnection = (Connection.isNull connection),+ Pqi.finish = readIORef (Connection.transport connection) >>= Transport.close, Pqi.reset = Connection.reconnect connection, Pqi.resetStart = Connection.reconnect connection $> True, Pqi.resetPoll = pure Pqi.PollingOk,- Pqi.db = pure (Just connection.info.database),- Pqi.user = pure (Just connection.info.user),- Pqi.pass = pure (Just connection.info.password),- Pqi.host = pure (Just connection.info.host),- Pqi.port = pure (Just (ByteString.Char8.pack (show connection.info.port))),+ Pqi.db = pure (Just (Connection.database (Connection.info connection))),+ Pqi.user = pure (Just (Connection.user (Connection.info connection))),+ Pqi.pass = pure (Just (Connection.password (Connection.info connection))),+ Pqi.host = pure (Just (Connection.host (Connection.info connection))),+ Pqi.port = pure (Just (ByteString.Char8.pack (show (Connection.port (Connection.info connection))))), Pqi.options = pure (Just ""),- Pqi.status = readIORef connection.connStatus,- Pqi.transactionStatus = transactionStatusOf <$> readIORef connection.txStatus,- Pqi.parameterStatus = \name -> Map.lookup name <$> readIORef connection.parameters,+ Pqi.status = readIORef (Connection.connStatus connection),+ Pqi.transactionStatus = transactionStatusOf <$> readIORef (Connection.txStatus connection),+ Pqi.parameterStatus = \name -> Map.lookup name <$> readIORef (Connection.parameters connection), Pqi.protocolVersion = pure 3, Pqi.serverVersion =- maybe 0 parseServerVersion . Map.lookup "server_version" <$> readIORef connection.parameters,- Pqi.errorMessage = readIORef connection.lastError,+ maybe 0 parseServerVersion . Map.lookup "server_version" <$> readIORef (Connection.parameters connection),+ Pqi.errorMessage = readIORef (Connection.lastError connection), Pqi.socket = do- transport <- readIORef connection.transport+ transport <- readIORef (Connection.transport connection) fd <- Transport.socketFd transport pure (Just (fromIntegral fd :: Fd)),- Pqi.backendPID = maybe 0 fst <$> readIORef connection.backendKey,+ Pqi.backendPID = maybe 0 fst <$> readIORef (Connection.backendKey connection), Pqi.connectionNeedsPassword = pure False,- Pqi.connectionUsedPassword = pure (not (ByteString.null connection.info.password)),+ Pqi.connectionUsedPassword = pure (not (ByteString.null (Connection.password (Connection.info connection)))), Pqi.exec = \sql -> fmap mkResult <$> Query.exec connection sql, Pqi.execParams = \sql params resultFormat -> fmap mkResult <$> Query.execParams connection sql params resultFormat,@@ -100,52 +100,52 @@ Pqi.getResult = fmap mkResult <$> Query.getNextResult connection, Pqi.consumeInput = pure True, Pqi.isBusy = pure False,- Pqi.setnonblocking = \flag -> writeIORef connection.nonblocking flag $> True,- Pqi.isnonblocking = readIORef connection.nonblocking,+ Pqi.setnonblocking = \flag -> writeIORef (Connection.nonblocking connection) flag $> True,+ Pqi.isnonblocking = readIORef (Connection.nonblocking connection), Pqi.setSingleRowMode = do- pending <- readIORef connection.asyncPending+ pending <- readIORef (Connection.asyncPending connection) if pending- then writeIORef connection.singleRowMode True $> True+ then writeIORef (Connection.singleRowMode connection) True $> True else pure False, Pqi.flush = pure Pqi.FlushOk,- Pqi.pipelineStatus = readIORef connection.pipelineStatus,- Pqi.enterPipelineMode = writeIORef connection.pipelineStatus Pqi.PipelineOn $> True,+ Pqi.pipelineStatus = readIORef (Connection.pipelineStatus connection),+ Pqi.enterPipelineMode = writeIORef (Connection.pipelineStatus connection) Pqi.PipelineOn $> True, Pqi.exitPipelineMode = do- pending <- readIORef connection.asyncPending+ pending <- readIORef (Connection.asyncPending connection) if pending then pure False- else writeIORef connection.pipelineStatus Pqi.PipelineOff $> True,+ else writeIORef (Connection.pipelineStatus connection) Pqi.PipelineOff $> True, Pqi.pipelineSync = do Connection.sendMessage connection syncMessage- modifyIORef' connection.pendingSyncs (+ 1)- writeIORef connection.asyncPending True+ modifyIORef' (Connection.pendingSyncs connection) (+ 1)+ writeIORef (Connection.asyncPending connection) True pure True, Pqi.sendFlushRequest = Connection.sendMessage connection flushMessage $> True, Pqi.getCancel = do- key <- readIORef connection.backendKey+ key <- readIORef (Connection.backendKey connection) pure $ fmap ( \(pid, secret) -> mkCancel NativeCancel- { host = connection.info.host,- port = connection.info.port,+ { host = Connection.host (Connection.info connection),+ port = Connection.port (Connection.info connection), pid, secret,- asyncPendingRef = connection.asyncPending,- pipelineStatusRef = connection.pipelineStatus,- pendingCommandsRef = connection.pendingCommands+ asyncPendingRef = (Connection.asyncPending connection),+ pipelineStatusRef = (Connection.pipelineStatus connection),+ pendingCommandsRef = (Connection.pendingCommands connection) } ) key,- Pqi.notifies = popFirst connection.pendingNotifications,- Pqi.disableNoticeReporting = writeIORef connection.noticeReporting False,- Pqi.enableNoticeReporting = writeIORef connection.noticeReporting True,- Pqi.getNotice = popFirst connection.notices,+ Pqi.notifies = popFirst (Connection.pendingNotifications connection),+ Pqi.disableNoticeReporting = writeIORef (Connection.noticeReporting connection) False,+ Pqi.enableNoticeReporting = writeIORef (Connection.noticeReporting connection) True,+ Pqi.getNotice = popFirst (Connection.notices connection), Pqi.putCopyData = \payload -> Connection.sendMessage connection (copyDataMessage payload) $> Pqi.CopyInOk, Pqi.putCopyEnd = \reason -> do Connection.sendMessage connection (maybe copyDoneMessage copyFailMessage reason)- writeIORef connection.asyncPending True+ writeIORef (Connection.asyncPending connection) True pure Pqi.CopyInOk, Pqi.getCopyData = getCopyData connection, Pqi.loCreat = LargeObject.loCreat connection,@@ -162,13 +162,13 @@ Pqi.loClose = LargeObject.loClose connection, Pqi.loUnlink = LargeObject.loUnlink connection, Pqi.clientEncoding =- fromMaybe "SQL_ASCII" . Map.lookup "client_encoding" <$> readIORef connection.parameters,+ fromMaybe "SQL_ASCII" . Map.lookup "client_encoding" <$> readIORef (Connection.parameters connection), Pqi.setClientEncoding = \encoding -> do result <- Query.exec connection ("SET client_encoding TO '" <> encoding <> "'")- pure (maybe False (\value -> value.status /= Pqi.FatalError) result),+ pure (maybe False (\value -> status value /= Pqi.FatalError) result), Pqi.setErrorVerbosity = \verbosity -> do- previous <- readIORef connection.errorVerbosity- writeIORef connection.errorVerbosity verbosity+ previous <- readIORef (Connection.errorVerbosity connection)+ writeIORef (Connection.errorVerbosity connection) verbosity pure previous } @@ -182,11 +182,11 @@ case message of CopyData payload -> pure (Pqi.CopyOutRow payload) CopyDone -> do- writeIORef connection.asyncPending True+ writeIORef (Connection.asyncPending connection) True pure Pqi.CopyOutDone CommandComplete _ -> drainToReady connection $> Pqi.CopyOutDone ErrorResponse _ -> drainToReady connection $> Pqi.CopyOutError- ReadyForQuery txState -> writeIORef connection.txStatus txState $> Pqi.CopyOutDone+ ReadyForQuery txState -> writeIORef (Connection.txStatus connection) txState $> Pqi.CopyOutDone _ -> getCopyData connection nonBlocking -- | Read messages until @ReadyForQuery@, recording the transaction status.@@ -194,7 +194,7 @@ drainToReady connection = do message <- Connection.nextMessage connection case message of- ReadyForQuery txState -> writeIORef connection.txStatus txState+ ReadyForQuery txState -> writeIORef (Connection.txStatus connection) txState _ -> drainToReady connection -- | Pop the oldest element of a list stored newest-first.
src/library/Pqi/Native/Auth.hs view
@@ -58,8 +58,8 @@ | otherwise = do clientNonce <- Base64.encode <$> getRandomBytes 18 let clientFirstBare = "n=,r=" <> clientNonce- step.sendInitial mechanismName ("n,," <> clientFirstBare)- step.receive >>= \case+ (sendInitial step) mechanismName ("n,," <> clientFirstBare)+ (receive step) >>= \case SaslError problem -> pure (Left problem) SaslContinue serverFirst -> case parseServerFirst serverFirst of@@ -74,8 +74,8 @@ clientSignature = hmacSha256 storedKey authMessage clientProof = xorBytes clientKey clientSignature clientFinal = clientFinalWithoutProof <> ",p=" <> Base64.encode clientProof- step.sendResponse clientFinal- step.receive >>= \case+ (sendResponse step) clientFinal+ (receive step) >>= \case SaslFinal _ -> pure (Right ()) SaslOk -> pure (Right ()) SaslError problem -> pure (Left problem)
src/library/Pqi/Native/Connection.hs view
@@ -186,7 +186,7 @@ -- | Send a serialized frontend message. sendMessage :: Connection -> Poker.Write -> IO () sendMessage connection write = do- transport <- readIORef connection.transport+ transport <- readIORef (transport connection) Transport.send transport write -- | Receive the next /protocol-relevant/ backend message, transparently@@ -195,23 +195,23 @@ -- (collected when notice reporting is on), and @NotificationResponse@ (queued). nextMessage :: Connection -> IO BackendMessage nextMessage connection = do- transport <- readIORef connection.transport+ transport <- readIORef (transport connection) (typeByte, body) <- Transport.receiveFrame transport case decodeBackendMessage typeByte body of Left err -> ioError (userError ("pqi-native: protocol decode error: " <> show err)) Right message -> case message of ParameterStatus key value -> do- modifyIORef' connection.parameters (Map.insert key value)+ modifyIORef' (parameters connection) (Map.insert key value) nextMessage connection NoticeResponse fields -> do- reporting <- readIORef connection.noticeReporting+ reporting <- readIORef (noticeReporting connection) when reporting $ do let noticeText = formatErrorFields (Map.fromList fields) unless (ByteString.null noticeText)- $ modifyIORef' connection.notices (noticeText :)+ $ modifyIORef' (notices connection) (noticeText :) nextMessage connection NotificationResponse pid channel payload -> do- modifyIORef' connection.pendingNotifications (Notify channel pid payload :)+ modifyIORef' (pendingNotifications connection) (Notify channel pid payload :) nextMessage connection other -> pure other @@ -222,8 +222,8 @@ -- | Record a flat error message and mark the connection bad. setError :: Connection -> ByteString -> IO () setError connection message = do- writeIORef connection.lastError (Just message)- writeIORef connection.connStatus ConnectionBad+ writeIORef (lastError connection) (Just message)+ writeIORef (connStatus connection) ConnectionBad -- | Open a connection: resolve and connect the socket, send the startup -- message, and run the authentication\/startup handshake. Like libpq, a failed@@ -232,7 +232,7 @@ establish :: ByteString -> IO Connection establish conninfo = do info <- parseConnInfo conninfo- transportResult <- try @IOException (Transport.connect info.host info.port)+ transportResult <- try @IOException (Transport.connect (host info) (port info)) case transportResult of Left err -> do transport <- Transport.unconnected@@ -241,7 +241,7 @@ pure connection Right transport -> do connection <- newConnection False transport info- sendMessage connection (startupMessage [("user", info.user), ("database", info.database)])+ sendMessage connection (startupMessage [("user", user info), ("database", database info)]) handshake connection pure connection @@ -252,23 +252,23 @@ transport <- Transport.unconnected info <- parseConnInfo "" conn <- newConnection True transport info- writeIORef conn.lastError (Just "connection pointer is NULL\n")+ writeIORef (lastError conn) (Just "connection pointer is NULL\n") pure conn -- | Close the current socket and run the startup handshake again on a fresh -- one, reusing the stored conninfo (the analogue of @PQreset@). reconnect :: Connection -> IO () reconnect connection = do- oldTransport <- readIORef connection.transport+ oldTransport <- readIORef (transport connection) Transport.close oldTransport- newTransport <- Transport.connect connection.info.host connection.info.port- writeIORef connection.transport newTransport- writeIORef connection.parameters Map.empty- writeIORef connection.backendKey Nothing- writeIORef connection.txStatus 0x49- writeIORef connection.connStatus ConnectionBad- writeIORef connection.lastError (Just "")- sendMessage connection (startupMessage [("user", connection.info.user), ("database", connection.info.database)])+ newTransport <- Transport.connect (host (info connection)) (port (info connection))+ writeIORef (transport connection) newTransport+ writeIORef (parameters connection) Map.empty+ writeIORef (backendKey connection) Nothing+ writeIORef (txStatus connection) 0x49+ writeIORef (connStatus connection) ConnectionBad+ writeIORef (lastError connection) (Just "")+ sendMessage connection (startupMessage [("user", user (info connection)), ("database", database (info connection))]) handshake connection newConnection :: Bool -> Transport -> ConnInfo -> IO Connection@@ -305,14 +305,14 @@ case message of AuthenticationOk -> startingUp AuthenticationCleartextPassword -> do- sendMessage connection (passwordMessage connection.info.password)+ sendMessage connection (passwordMessage (password (info connection))) authenticating AuthenticationMD5Password salt -> do- let response = Auth.md5Password connection.info.user connection.info.password salt+ let response = Auth.md5Password (user (info connection)) (password (info connection)) salt sendMessage connection (passwordMessage response) authenticating AuthenticationSASL mechanisms ->- Auth.scram connection.info.user connection.info.password mechanisms (saslExchange connection) >>= \case+ Auth.scram (user (info connection)) (password (info connection)) mechanisms (saslExchange connection) >>= \case Left problem -> setError connection problem Right () -> startingUp ErrorResponse fields -> failWith fields@@ -321,26 +321,26 @@ message <- nextMessage connection case message of BackendKeyData pid secret -> do- writeIORef connection.backendKey (Just (pid, secret))+ writeIORef (backendKey connection) (Just (pid, secret)) startingUp ReadyForQuery txState -> do- writeIORef connection.txStatus txState- writeIORef connection.connStatus ConnectionOk+ writeIORef (txStatus connection) txState+ writeIORef (connStatus connection) ConnectionOk ErrorResponse fields -> failWith fields _ -> startingUp failWith fields = do let fmtFields = formatErrorFields (Map.fromList fields)- transport <- readIORef connection.transport+ transport <- readIORef (transport connection) mIp <- catch (Just <$> Transport.peerIp transport) (\(_ :: SomeException) -> pure Nothing) setError connection $ case mIp of Nothing -> fmtFields Just ip -> "connection to server at \""- <> connection.info.host+ <> host (info connection) <> "\" (" <> ip <> "), port "- <> ByteString.Char8.pack (show connection.info.port)+ <> ByteString.Char8.pack (show (port (info connection))) <> " failed: " <> fmtFields
src/library/Pqi/Native/LargeObject.hs view
@@ -92,7 +92,7 @@ callBinary connection sql params = (>>= firstValue) <$> Query.execParams connection sql params Binary firstValue :: NativeResult -> Maybe ByteString-firstValue result = case result.rows of+firstValue result = case rows result of (cell : _) : _ -> cell _ -> Nothing
src/library/Pqi/Native/Query.hs view
@@ -47,13 +47,13 @@ asyncParamsWrite :: ByteString -> [Maybe (Word32, ByteString, Format)] -> Format -> Poker.Write asyncParamsWrite sql params resultFormat = parseMessage "" sql (fmap paramOid params)- <> bindMessage "" "" (fmap paramFormat params) (fmap paramValue params) [formatCode resultFormat]+ <> bindMessage "" "" (fmap paramFormat params) (fmap paramValue params) [formatCodeOf resultFormat] <> describePortalMessage "" <> executeMessage "" 0 asyncPreparedWrite :: ByteString -> [Maybe (ByteString, Format)] -> Format -> Poker.Write asyncPreparedWrite name params resultFormat =- bindMessage "" name (fmap boundFormat params) (fmap boundValue params) [formatCode resultFormat]+ bindMessage "" name (fmap boundFormat params) (fmap boundValue params) [formatCodeOf resultFormat] <> describePortalMessage "" <> executeMessage "" 0 @@ -88,20 +88,20 @@ -- | Send a write in async mode, tracking pending commands for pipeline abort. sendAsync :: Connection -> ByteString -> Poker.Write -> IO Bool sendAsync connection sql write = do- status <- readIORef connection.connStatus+ status <- readIORef (connStatus connection) case status of ConnectionOk -> do sendMessage connection write- writeIORef connection.currentQuery sql- writeIORef connection.asyncPending True- pipeStatus <- readIORef connection.pipelineStatus- when (pipeStatus /= PipelineOff) $ modifyIORef' connection.pendingCommands (+ 1)+ writeIORef (currentQuery connection) sql+ writeIORef (asyncPending connection) True+ pipeStatus <- readIORef (pipelineStatus connection)+ when (pipeStatus /= PipelineOff) $ modifyIORef' (pendingCommands connection) (+ 1) pure True _ -> pure False -- | Whether the connection is in pipeline mode. inPipeline :: Connection -> IO Bool-inPipeline connection = (/= PipelineOff) <$> readIORef connection.pipelineStatus+inPipeline connection = (/= PipelineOff) <$> readIORef (pipelineStatus connection) -- Simple query protocol: no Sync needed (server sends ReadyForQuery on its own). sendQuery :: Connection -> ByteString -> IO Bool@@ -125,7 +125,7 @@ $ if pipeline then parseMessage name sql (fromMaybe [] parameterTypes) else prepareWrite name sql parameterTypes- when (ok && pipeline) $ modifyIORef' connection.pendingParses (+ 1)+ when (ok && pipeline) $ modifyIORef' (pendingParses connection) (+ 1) pure ok sendQueryPrepared :: Connection -> ByteString -> [Maybe (ByteString, Format)] -> Format -> IO Bool@@ -162,18 +162,18 @@ -- 'SingleTuple' result followed by a final 'TuplesOk' with no rows. getNextResult :: Connection -> IO (Maybe NativeResult) getNextResult connection = do- pending <- readIORef connection.asyncPending+ pending <- readIORef (asyncPending connection) if not pending then pure Nothing else do- sepPending <- readIORef connection.pipelineSeparatorPending+ sepPending <- readIORef (pipelineSeparatorPending connection) if sepPending then do- writeIORef connection.pipelineSeparatorPending False+ writeIORef (pipelineSeparatorPending connection) False pure Nothing else do- singleRow <- readIORef connection.singleRowMode- cachedFields <- readIORef connection.singleRowFields+ singleRow <- readIORef (singleRowMode connection)+ cachedFields <- readIORef (singleRowFields connection) let initBuilder = if singleRow && not (null cachedFields) then emptyBuilder {accFields = cachedFields, accSawRowDescription = True}@@ -184,20 +184,20 @@ -- pipeline mode. Called when a "terminal" result is about to be returned. finishCommand pipeStatus = do when (pipeStatus /= PipelineOff) $ do- modifyIORef' connection.pendingCommands (subtract 1)- writeIORef connection.pipelineSeparatorPending True+ modifyIORef' (pendingCommands connection) (subtract 1)+ writeIORef (pipelineSeparatorPending connection) True go singleRow builder = do- pipeStatus <- readIORef connection.pipelineStatus+ pipeStatus <- readIORef (pipelineStatus connection) -- In aborted pipeline mode, if the server has already sent nothing for -- the remaining commands (it discards them after the first error), we -- generate synthetic PipelineAbort results for each outstanding command -- rather than blocking on a wire read that will never come.- pending <- readIORef connection.pendingCommands+ pending <- readIORef (pendingCommands connection) if pipeStatus == PipelineAborted && pending > 0 then do- modifyIORef' connection.pendingCommands (subtract 1)- writeIORef connection.pipelineSeparatorPending True+ modifyIORef' (pendingCommands connection) (subtract 1)+ writeIORef (pipelineSeparatorPending connection) True pure (Just (NativeResult PipelineAbort [] [] Nothing Map.empty [] "")) else readAndProcess singleRow builder pipeStatus @@ -213,14 +213,14 @@ DataRow values -> if singleRow then do- writeIORef connection.singleRowFields builder.accFields- pure (Just (NativeResult SingleTuple builder.accFields [values] Nothing Map.empty [] ""))- else go singleRow builder {accRevRows = values : builder.accRevRows}+ writeIORef (singleRowFields connection) (accFields builder)+ pure (Just (NativeResult SingleTuple (accFields builder) [values] Nothing Map.empty [] ""))+ else go singleRow builder {accRevRows = values : (accRevRows builder)} ParseComplete -> do- parses <- readIORef connection.pendingParses+ parses <- readIORef (pendingParses connection) if parses > 0 && pipeStatus /= PipelineOff then do- modifyIORef' connection.pendingParses (subtract 1)+ modifyIORef' (pendingParses connection) (subtract 1) finishCommand pipeStatus pure (Just (NativeResult CommandOk [] [] Nothing Map.empty [] "")) else go singleRow builder {accHadResponse = True}@@ -231,16 +231,16 @@ CommandComplete tag -> do if singleRow then do- writeIORef connection.singleRowMode False- writeIORef connection.singleRowFields []- writeIORef connection.lastError (Just "")- pure (Just (NativeResult TuplesOk builder.accFields [] (Just tag) Map.empty [] ""))+ writeIORef (singleRowMode connection) False+ writeIORef (singleRowFields connection) []+ writeIORef (lastError connection) (Just "")+ pure (Just (NativeResult TuplesOk (accFields builder) [] (Just tag) Map.empty [] "")) else do- writeIORef connection.lastError (Just "")+ writeIORef (lastError connection) (Just "") finishCommand pipeStatus pure (Just (commandResult builder (Just tag))) EmptyQueryResponse -> do- writeIORef connection.lastError (Just "")+ writeIORef (lastError connection) (Just "") finishCommand pipeStatus pure (Just (NativeResult EmptyQuery [] [] Nothing Map.empty [] "")) ErrorResponse fs -> do@@ -252,33 +252,33 @@ finishCommand pipeStatus pure (Just (NativeResult PipelineAbort [] [] Nothing Map.empty [] "")) PipelineOn -> do- writeIORef connection.pipelineStatus PipelineAborted+ writeIORef (pipelineStatus connection) PipelineAborted finishCommand PipelineOn pure (Just (NativeResult FatalError [] [] Nothing errMap [] "")) PipelineOff -> do- sql <- readIORef connection.currentQuery- writeIORef connection.lastError (Just (formatResultError sql errMap))+ sql <- readIORef (currentQuery connection)+ writeIORef (lastError connection) (Just (formatResultError sql errMap)) pure (Just (NativeResult FatalError [] [] Nothing errMap [] sql)) PortalSuspended -> pure (Just (commandResult builder Nothing)) ReadyForQuery txState -> do- writeIORef connection.txStatus txState+ writeIORef (txStatus connection) txState case pipeStatus of PipelineOff -> do- writeIORef connection.asyncPending False- if builder.accHadResponse+ writeIORef (asyncPending connection) False+ if (accHadResponse builder) then pure (Just (describeResult builder)) else pure Nothing _ -> do- writeIORef connection.pipelineStatus PipelineOn+ writeIORef (pipelineStatus connection) PipelineOn -- A PipelineSync result is its own command boundary: unlike a -- normal command result, libpq does not emit a separating NULL -- after it, so consecutive syncs are reported back-to-back. We -- therefore never set 'pipelineSeparatorPending' here. Only the -- final sync clears 'asyncPending'; an earlier one leaves it set -- so the next 'getNextResult' reads straight on to the next sync.- remaining <- atomicModifyIORef' connection.pendingSyncs (\n -> (n - 1, n - 1))- when (remaining == 0) $ writeIORef connection.asyncPending False+ remaining <- atomicModifyIORef' (pendingSyncs connection) (\n -> (n - 1, n - 1))+ when (remaining == 0) $ writeIORef (asyncPending connection) False pure (Just (NativeResult PipelineSync [] [] Nothing Map.empty [] "")) _ -> go singleRow builder @@ -300,19 +300,21 @@ paramOid = maybe 0 (\(oid, _, _) -> oid) paramFormat :: Maybe (Word32, ByteString, Format) -> Int16-paramFormat = maybe 0 (\(_, _, format) -> formatCode format)+paramFormat = maybe 0 (\(_, _, format) -> formatCodeOf format) paramValue :: Maybe (Word32, ByteString, Format) -> Maybe ByteString paramValue = fmap (\(_, value, _) -> value) boundFormat :: Maybe (ByteString, Format) -> Int16-boundFormat = maybe 0 (formatCode . snd)+boundFormat = maybe 0 (formatCodeOf . snd) boundValue :: Maybe (ByteString, Format) -> Maybe ByteString boundValue = fmap fst -formatCode :: Format -> Int16-formatCode = \case+-- | Renamed from @formatCode@ to avoid clashing with 'FieldDescription's+-- @formatCode@ field now that 'DuplicateRecordFields' is no longer enabled.+formatCodeOf :: Format -> Int16+formatCodeOf = \case Text -> 0 Binary -> 1 @@ -322,7 +324,7 @@ -- when the connection is not usable. withReady :: Connection -> IO (Maybe a) -> IO (Maybe a) withReady connection action = do- status <- readIORef connection.connStatus+ status <- readIORef (connStatus connection) case status of ConnectionOk -> action _ -> pure Nothing@@ -350,16 +352,16 @@ message <- nextMessage connection case message of RowDescription fs -> go builder {accFields = fs, accSawRowDescription = True} acc- DataRow values -> go builder {accRevRows = values : builder.accRevRows} acc+ DataRow values -> go builder {accRevRows = values : (accRevRows builder)} acc CommandComplete tag -> do- writeIORef connection.lastError (Just "")+ writeIORef (lastError connection) (Just "") go emptyBuilder (commandResult builder (Just tag) : acc) EmptyQueryResponse -> do- writeIORef connection.lastError (Just "")+ writeIORef (lastError connection) (Just "") go emptyBuilder (NativeResult EmptyQuery [] [] Nothing Map.empty [] "" : acc) ErrorResponse fs -> do let errMap = Map.fromList fs- writeIORef connection.lastError (Just (formatResultError sql errMap))+ writeIORef (lastError connection) (Just (formatResultError sql errMap)) go emptyBuilder (NativeResult FatalError [] [] Nothing errMap [] sql : acc) CopyInResponse _ formats -> let fields = map copyField formats@@ -368,7 +370,7 @@ let fields = map copyField formats in pure (reverse (NativeResult CopyOut fields [] Nothing Map.empty [] "" : acc)) ReadyForQuery txState -> do- writeIORef connection.txStatus txState+ writeIORef (txStatus connection) txState pure (reverse acc) _ -> go builder acc @@ -382,23 +384,23 @@ RowDescription fs -> go builder {accFields = fs, accSawRowDescription = True} finished ParameterDescription oids -> go builder {accParamOids = oids} finished NoData -> go builder finished- DataRow values -> go builder {accRevRows = values : builder.accRevRows} finished+ DataRow values -> go builder {accRevRows = values : (accRevRows builder)} finished ParseComplete -> go builder finished BindComplete -> go builder finished CloseComplete -> go builder finished PortalSuspended -> go emptyBuilder (finished <|> Just (commandResult builder Nothing)) CommandComplete tag -> do- writeIORef connection.lastError (Just "")+ writeIORef (lastError connection) (Just "") go emptyBuilder (Just (commandResult builder (Just tag))) EmptyQueryResponse -> do- writeIORef connection.lastError (Just "")+ writeIORef (lastError connection) (Just "") go emptyBuilder (Just (NativeResult EmptyQuery [] [] Nothing Map.empty [] "")) ErrorResponse fs -> do let errMap = Map.fromList fs- writeIORef connection.lastError (Just (formatResultError sql errMap))+ writeIORef (lastError connection) (Just (formatResultError sql errMap)) go emptyBuilder (Just (NativeResult FatalError [] [] Nothing errMap [] sql)) ReadyForQuery txState -> do- writeIORef connection.txStatus txState+ writeIORef (txStatus connection) txState pure (fromMaybe (describeResult builder) finished) _ -> go builder finished @@ -407,12 +409,12 @@ commandResult :: Builder -> Maybe ByteString -> NativeResult commandResult builder tag = NativeResult- (if builder.accSawRowDescription then TuplesOk else CommandOk)- builder.accFields- (reverse builder.accRevRows)+ (if (accSawRowDescription builder) then TuplesOk else CommandOk)+ (accFields builder)+ (reverse (accRevRows builder)) tag Map.empty- builder.accParamOids+ (accParamOids builder) "" -- | A result with no command completion (a @Describe@\/@Parse@-only flow):@@ -421,11 +423,11 @@ describeResult builder = NativeResult CommandOk- builder.accFields- (reverse builder.accRevRows)+ (accFields builder)+ (reverse (accRevRows builder)) Nothing Map.empty- builder.accParamOids+ (accParamOids builder) "" lastMaybe :: [a] -> Maybe a
src/library/Pqi/Native/Types.hs view
@@ -145,35 +145,35 @@ mkResult :: NativeResult -> Result mkResult result = Result- { resultStatus = pure result.status,- resultErrorMessage = pure (Just (formatResultError result.queryText result.errorFields)),- resultErrorField = \field -> pure (Map.lookup (fieldCodeByte field) result.errorFields),+ { resultStatus = pure (status result),+ resultErrorMessage = pure (Just (formatResultError (queryText result) (errorFields result))),+ resultErrorField = \field -> pure (Map.lookup (fieldCodeByte field) (errorFields result)), unsafeFreeResult = pure (),- ntuples = pure (fromIntegral (length result.rows)),- nfields = pure (fromIntegral (length result.fields)),+ ntuples = pure (fromIntegral (length (rows result))),+ nfields = pure (fromIntegral (length (fields result))), fname = \column -> pure $ do- fd <- atMay result.fields column- if ByteString.null fd.name then Nothing else Just fd.name,- fnumber = \name ->- pure (fromIntegral <$> findIndex (\field -> field.name == foldIdentifier name) result.fields),- ftable = \column -> pure (maybe 0 (.tableOid) (atMay result.fields column)),+ fd <- atMay (fields result) column+ if ByteString.null (name fd) then Nothing else Just (name fd),+ fnumber = \colName ->+ pure (fromIntegral <$> findIndex (\field -> name field == foldIdentifier colName) (fields result)),+ ftable = \column -> pure (maybe 0 tableOid (atMay (fields result) column)), ftablecol = \column ->- pure (maybe 0 (\field -> fromIntegral (field.columnAttributeNumber :: Int16)) (atMay result.fields column)),+ pure (maybe 0 (\field -> fromIntegral (columnAttributeNumber field :: Int16)) (atMay (fields result) column)), fformat = \column ->- pure (maybe Text (\field -> formatOf field.formatCode) (atMay result.fields column)),- ftype = \column -> pure (maybe 0 (.typeOid) (atMay result.fields column)),+ pure (maybe Text (\field -> formatOf (formatCode field)) (atMay (fields result) column)),+ ftype = \column -> pure (maybe 0 typeOid (atMay (fields result) column)), fmod = \column ->- pure (maybe 0 (\field -> fromIntegral (field.typeModifier :: Int32)) (atMay result.fields column)),+ pure (maybe 0 (\field -> fromIntegral (typeModifier field :: Int32)) (atMay (fields result) column)), fsize = \column ->- pure (maybe 0 (\field -> fromIntegral (field.typeSize :: Int16)) (atMay result.fields column)),+ pure (maybe 0 (\field -> fromIntegral (typeSize field :: Int16)) (atMay (fields result) column)), getvalue = \row column -> pure (join (cellAt result row column)), getvalue' = \row column -> pure (join (cellAt result row column)), getisnull = \row column -> pure (maybe True isNothing (cellAt result row column)), getlength = \row column -> pure (maybe 0 (maybe 0 ByteString.length) (cellAt result row column)),- nparams = pure (fromIntegral (length result.paramOids)),- paramtype = \index -> pure (fromMaybe 0 (atMay result.paramOids index)),- cmdStatus = pure (Just (fromMaybe "" result.commandTag)),- cmdTuples = pure (Just (maybe "" affectedRows result.commandTag))+ nparams = pure (fromIntegral (length (paramOids result))),+ paramtype = \index -> pure (fromMaybe 0 (atMay (paramOids result) index)),+ cmdStatus = pure (Just (fromMaybe "" (commandTag result))),+ cmdTuples = pure (Just (maybe "" affectedRows (commandTag result))) } -- | Build a 'Cancel' whose field closes over the given 'NativeCancel'.@@ -181,12 +181,12 @@ mkCancel nc = Cancel { cancel = do- pending <- readIORef nc.asyncPendingRef+ pending <- readIORef (asyncPendingRef nc) if not pending then pure (Right ()) else do- transport <- Transport.connect nc.host nc.port- Transport.send transport (cancelRequest nc.pid nc.secret)+ transport <- Transport.connect (host nc) (port nc)+ Transport.send transport (cancelRequest (pid nc) (secret nc)) -- Read until EOF to ensure the server has processed the cancel request -- before we close the connection. This matches libpq's PQcancel behavior -- and prevents the cancel signal from racing with the next query.@@ -206,7 +206,7 @@ cellAt :: NativeResult -> Int32 -> Int32 -> Maybe (Maybe ByteString) cellAt result row column = do- rowValues <- atMay result.rows row+ rowValues <- atMay (rows result) row atMay rowValues column formatOf :: Int16 -> Format
src/transport/Pqi/Native/Transport.hs view
@@ -53,30 +53,30 @@ -- | Close the connection. close :: Transport -> IO ()-close transport = Socket.close transport.socket+close transport = Socket.close (socket transport) -- | The underlying socket file descriptor. socketFd :: Transport -> IO Int32-socketFd transport = fromIntegral <$> Socket.unsafeFdSocket transport.socket+socketFd transport = fromIntegral <$> Socket.unsafeFdSocket (socket transport) -- | Send a serialized message. send :: Transport -> Poker.Write -> IO ()-send transport write = Socket.ByteString.sendAll transport.socket (Poker.toByteString write)+send transport write = Socket.ByteString.sendAll (socket transport) (Poker.toByteString write) -- | Read exactly @n@ bytes, looping over @recv@ (which yields up to @n@) and -- buffering any overshoot. Throws on EOF before @n@ bytes arrive. receiveExactly :: Transport -> Int -> IO ByteString receiveExactly transport n = do- buffered <- readIORef transport.readBuffer+ buffered <- readIORef (readBuffer transport) go buffered where go accumulated | ByteString.length accumulated >= n = do let (result, rest) = ByteString.splitAt n accumulated- writeIORef transport.readBuffer rest+ writeIORef (readBuffer transport) rest pure result | otherwise = do- chunk <- Socket.ByteString.recv transport.socket (max 4096 (n - ByteString.length accumulated))+ chunk <- Socket.ByteString.recv (socket transport) (max 4096 (n - ByteString.length accumulated)) if ByteString.null chunk then ioError (mkIOError eofErrorType "pqi-native: connection closed by server" Nothing Nothing) else go (accumulated <> chunk)@@ -102,7 +102,7 @@ readUntilClosed transport = go where go = do- chunk <- Socket.ByteString.recv transport.socket 4096+ chunk <- Socket.ByteString.recv (socket transport) 4096 if ByteString.null chunk then pure () else go@@ -111,7 +111,7 @@ -- @\"127.0.0.1\"@). Throws if the socket has no peer (unconnected). peerIp :: Transport -> IO ByteString peerIp transport = do- addr <- Socket.getPeerName transport.socket+ addr <- Socket.getPeerName (socket transport) (Just ip, _) <- Socket.getNameInfo [Socket.NI_NUMERICHOST] True False addr pure (ByteString.Char8.pack ip)