hasql-listen-notify (empty) → 0.1.0

raw patch · 5 files changed

+304/−0 lines, 5 filesdep +basedep +bytestringdep +hasql

Dependencies added: base, bytestring, hasql, mtl, postgresql-libpq, text

Files

CHANGELOG.md +3/−0
LICENSE +11/−0
README.md +9/−0
hasql-listen-notify.cabal +59/−0
src/Hasql/ListenNotify.hs +222/−0

+ CHANGELOG.md view

@@ -0,0 +1,3 @@+## [0.1.0] - 2023-01-23++Initial release.

+ LICENSE view

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

+ README.md view

@@ -0,0 +1,9 @@+# `hasql-listen-notify`++`LISTEN`/`NOTIFY` using `hasql`.++[![Build Status](https://github.com/awkward-squad/hasql-listen-notify/workflows/Haskell-CI/badge.svg)](https://github.com/awkward-squad/hasql-listen-notify/actions?query=workflow%3AHaskell-CI)+[![Hackage](https://img.shields.io/hackage/v/hasql-listen-notify.svg?label=hasql-listen-notify&logo=haskell)](https://hackage.haskell.org/package/hasql-listen-notify)+[![Stackage LTS](https://stackage.org/package/hasql-listen-notify/badge/lts)](https://www.stackage.org/lts/package/hasql-listen-notify)+[![Stackage Nightly](https://stackage.org/package/hasql-listen-notify/badge/nightly)](https://www.stackage.org/nightly/package/hasql-listen-notify)+[![Dependencies](https://img.shields.io/hackage-deps/v/hasql-listen-notify)](https://packdeps.haskellers.com/reverse/hasql-listen-notify)

+ hasql-listen-notify.cabal view

@@ -0,0 +1,59 @@+cabal-version: 2.2++author: Mitchell Rosen, Travis Staton+bug-reports: https://github.com/awkward-squad/hasql-listen-notify/issues+category: Concurrency+copyright: Copyright (C) 2023 Mitchell Rosen, Travis Staton+description: LISTEN/NOTIFY with @hasql@.+homepage: https://github.com/awkward-squad/hasql-listen-notify+license-file: LICENSE+license: BSD-3-Clause+maintainer: Mitchell Rosen <mitchellwrosen@gmail.com>, Travis Staton <hello@travisstaton.com>+name: hasql-listen-notify+synopsis: LISTEN/NOTIFY with hasql+tested-with: GHC == 9.0.2, GHC == 9.2.5, GHC == 9.4.4+version: 0.1.0++extra-source-files:+  CHANGELOG.md+  README.md++source-repository head+  type: git+  location: https://github.com/awkward-squad/hasql-listen-notify.git++library+  build-depends:+    base ^>= 4.15 || ^>= 4.16 || ^>= 4.17,+    bytestring ^>= 0.10 || ^>= 0.11,+    hasql ^>= 1.6,+    mtl ^>= 2.2 || ^>= 2.3.1,+    postgresql-libpq ^>= 0.9.4.3,+    text ^>= 1.2.5.0 || ^>= 2.0,+  default-extensions:+    BlockArguments+    DeriveGeneric+    DerivingStrategies+    DuplicateRecordFields+    GeneralizedNewtypeDeriving+    LambdaCase+    NamedFieldPuns+    OverloadedStrings+  default-language: Haskell2010+  exposed-modules: Hasql.ListenNotify+  ghc-options:+    -Weverything+    -Wno-all-missed-specialisations+    -Wno-implicit-prelude+    -Wno-missed-specialisations+    -Wno-missing-import-lists+    -Wno-safe+    -Wno-unsafe+  if impl(ghc >= 8.10)+    ghc-options:+      -Wno-missing-safe-haskell-mode+      -Wno-prepositive-qualified-module+  if impl(ghc >= 9.2)+    ghc-options:+      -Wno-missing-kind-signatures+  hs-source-dirs: src

+ src/Hasql/ListenNotify.hs view

@@ -0,0 +1,222 @@+-- | @LISTEN@/@NOTIFY@ with @hasql@.+module Hasql.ListenNotify+  ( -- * Listen+    Identifier (..),+    listen,+    unlisten,+    unlistenAll,+    escapeIdentifier,+    Notification (..),+    await,+    poll,+    backendPid,++    -- * Notify+    Notify (..),+    notify,+  )+where++import Control.Exception (throwIO, try)+import Control.Monad.Except (throwError)+import Control.Monad.IO.Class+import Control.Monad.Reader (ask)+import Data.ByteString (ByteString)+import qualified Data.ByteString.Builder as ByteString (Builder)+import qualified Data.ByteString.Builder as ByteString.Builder+import qualified Data.ByteString.Lazy as ByteString.Lazy+import Data.Functor.Contravariant ((>$<))+import Data.Text (Text)+import qualified Data.Text.Encoding as Text+import qualified Database.PostgreSQL.LibPQ as LibPQ+import GHC.Conc.IO (threadWaitRead)+import GHC.Generics (Generic)+import qualified Hasql.Connection as Connection+import qualified Hasql.Decoders as Decoders+import qualified Hasql.Encoders as Encoders+import Hasql.Session (Session)+import qualified Hasql.Session as Session+import Hasql.Statement (Statement (..))+import System.Posix.Types (CPid)++-- | Listen to a channel.+--+-- https://www.postgresql.org/docs/current/sql-listen.html+listen :: Identifier -> Statement () ()+listen (Identifier chan) =+  Statement (builderToByteString sql) Encoders.noParams Decoders.noResult False+  where+    sql :: ByteString.Builder+    sql =+      "LISTEN " <> ByteString.Builder.byteString chan++-- | Stop listening to a channel.+--+-- https://www.postgresql.org/docs/current/sql-unlisten.html+unlisten :: Identifier -> Statement () ()+unlisten (Identifier chan) =+  Statement (builderToByteString sql) Encoders.noParams Decoders.noResult False+  where+    sql :: ByteString.Builder+    sql =+      "UNLISTEN " <> ByteString.Builder.byteString chan++-- | Stop listening to all channels.+--+-- https://www.postgresql.org/docs/current/sql-unlisten.html+unlistenAll :: Statement () ()+unlistenAll =+  Statement "UNLISTEN *" Encoders.noParams Decoders.noResult False++-- | A Postgres identifier.+newtype Identifier+  = Identifier ByteString+  deriving newtype (Eq, Ord, Show)++-- | Escape a string as a Postgres identifier.+--+--+-- https://www.postgresql.org/docs/15/libpq-exec.html+escapeIdentifier :: Text -> Session Identifier+escapeIdentifier text = do+  libpq (\conn -> try (escapeIdentifier_ conn text)) >>= \case+    Left err -> throwError err+    Right identifier -> pure (Identifier identifier)++escapeIdentifier_ :: LibPQ.Connection -> Text -> IO ByteString+escapeIdentifier_ conn text =+  LibPQ.escapeIdentifier conn (Text.encodeUtf8 text) >>= \case+    Nothing -> throwQueryError conn "PQescapeIdentifier()" [text]+    Just identifier -> pure identifier++-- | An incoming notification.+data Notification = Notification+  { channel :: !Text,+    payload :: !Text,+    pid :: !CPid+  }+  deriving stock (Eq, Generic, Show)++-- | Get the next notification received from the server.+--+-- https://www.postgresql.org/docs/current/libpq-notify.html+await :: Session Notification+await =+  libpq (\conn -> try (await_ conn)) >>= \case+    Left err -> throwError err+    Right notification -> pure (parseNotification notification)++await_ :: LibPQ.Connection -> IO LibPQ.Notify+await_ conn =+  pollForNotification+  where+    pollForNotification :: IO LibPQ.Notify+    pollForNotification =+      poll_ conn >>= \case+        -- Block until a notification arrives. Snag: the connection might be closed (what). If so, attempt to reset it+        -- and poll for a notification on the new connection.+        Nothing ->+          LibPQ.socket conn >>= \case+            -- "No connection is currently open"+            Nothing -> do+              pqReset conn+              pollForNotification+            Just socket -> do+              threadWaitRead socket+              -- Data has appeared on the socket, but libPQ won't buffer it for us unless we do something (PQexec, etc).+              -- PQconsumeInput is provided for when we don't have anything to do except populate the notification+              -- buffer.+              pqConsumeInput conn+              pollForNotification+        Just notification -> pure notification++-- | Variant of 'await' that doesn't block.+poll :: Session (Maybe Notification)+poll =+  libpq (\conn -> try (poll_ conn)) >>= \case+    Left err -> throwError err+    Right maybeNotification -> pure (parseNotification <$> maybeNotification)++-- First call `notifies` to pop a notification off of the buffer, if there is one. If there isn't, try `consumeInput` to+-- populate the buffer, followed by another followed by another `notifies`.+poll_ :: LibPQ.Connection -> IO (Maybe LibPQ.Notify)+poll_ conn =+  LibPQ.notifies conn >>= \case+    Nothing -> do+      pqConsumeInput conn+      LibPQ.notifies conn+    notification -> pure notification++-- | Get the PID of the backend process handling this session. This can be used to filter out notifications that+-- originate from this session.+--+-- https://www.postgresql.org/docs/current/libpq-status.html+backendPid :: Session CPid+backendPid =+  libpq LibPQ.backendPID++-- | An outgoing notification.+data Notify = Notify+  { channel :: !Text,+    payload :: !Text+  }+  deriving stock (Eq, Generic, Show)++-- | Notify a channel.+--+-- https://www.postgresql.org/docs/current/sql-notify.html+notify :: Statement Notify ()+notify =+  Statement sql encoder Decoders.noResult True+  where+    sql :: ByteString+    sql =+      "SELECT pg_notify($1, $2)"++    encoder :: Encoders.Params Notify+    encoder =+      ((\Notify {channel} -> channel) >$< Encoders.param (Encoders.nonNullable Encoders.text))+        <> ((\Notify {payload} -> payload) >$< Encoders.param (Encoders.nonNullable Encoders.text))++------------------------------------------------------------------------------------------------------------------------+-- Little wrappers that throw++pqConsumeInput :: LibPQ.Connection -> IO ()+pqConsumeInput conn =+  LibPQ.consumeInput conn >>= \case+    False -> throwQueryError conn "PQconsumeInput()" []+    True -> pure ()++pqReset :: LibPQ.Connection -> IO ()+pqReset conn = do+  LibPQ.reset conn+  LibPQ.status conn >>= \case+    LibPQ.ConnectionOk -> throwQueryError conn "PQreset()" []+    _ -> pure ()++-- Throws a QueryError+throwQueryError :: LibPQ.Connection -> ByteString -> [Text] -> IO void+throwQueryError conn context params = do+  message <- LibPQ.errorMessage conn+  throwIO (Session.QueryError context params (Session.ClientError message))++--++libpq :: (LibPQ.Connection -> IO a) -> Session a+libpq action = do+  conn <- ask+  liftIO (Connection.withLibPQConnection conn action)++builderToByteString :: ByteString.Builder -> ByteString+builderToByteString =+  ByteString.Lazy.toStrict . ByteString.Builder.toLazyByteString+{-# INLINE builderToByteString #-}++-- Parse a Notify from a LibPQ.Notify+parseNotification :: LibPQ.Notify -> Notification+parseNotification notification =+  Notification+    { channel = Text.decodeUtf8 (LibPQ.notifyRelname notification),+      payload = Text.decodeUtf8 (LibPQ.notifyExtra notification),+      pid = LibPQ.notifyBePid notification+    }