Skip to content

Commit 769ca47

Browse files
diogobdiogob
committed
Reformat
1 parent 88502ee commit 769ca47

File tree

1 file changed

+125
-120
lines changed

1 file changed

+125
-120
lines changed

‎src/Hasql/Notifications.hs

+125-120
Original file line numberDiff line numberDiff line change
@@ -1,48 +1,44 @@
11
{-# LANGUAGE CPP #-}
2-
{-|
3-
This module has functions to send commands LISTEN and NOTIFY to the database server.
4-
It also has a function to wait for and handle notifications on a database connection.
52

6-
For more information check the [PostgreSQL documentation](https://www.postgresql.org/docs/current/libpq-notify.html).
7-
8-
-}
3+
-- |
4+
-- This module has functions to send commands LISTEN and NOTIFY to the database server.
5+
-- It also has a function to wait for and handle notifications on a database connection.
6+
--
7+
-- For more information check the [PostgreSQL documentation](https://www.postgresql.org/docs/current/libpq-notify.html).
98
module Hasql.Notifications
10-
( notifyPool
11-
, notify
12-
, listen
13-
, unlisten
14-
, waitForNotifications
15-
, PgIdentifier
16-
, toPgIdentifier
17-
, fromPgIdentifier
18-
) where
19-
20-
import Hasql.Pool (Pool, UsageError, use)
21-
import Hasql.Session (sql, run, statement)
22-
import qualified Hasql.Session as S
23-
import qualified Hasql.Statement as HST
24-
import Hasql.Connection (Connection, withLibPQConnection)
25-
import qualified Hasql.Decoders as HD
26-
import qualified Hasql.Encoders as HE
27-
import qualified Database.PostgreSQL.LibPQ as PQ
9+
( notifyPool,
10+
notify,
11+
listen,
12+
unlisten,
13+
waitForNotifications,
14+
PgIdentifier,
15+
toPgIdentifier,
16+
fromPgIdentifier,
17+
)
18+
where
19+
20+
import Control.Concurrent (threadDelay, threadWaitRead)
21+
import Control.Exception (Exception, throw)
22+
import Control.Monad (forever, unless, void)
23+
import Data.ByteString.Char8 (ByteString)
24+
import Data.Functor.Contravariant (contramap)
2825
import Data.Text (Text)
2926
import qualified Data.Text as T
3027
import qualified Data.Text.Encoding as T
31-
import Data.ByteString.Char8 (ByteString)
32-
import Data.Functor.Contravariant (contramap)
33-
import Control.Monad (void, forever)
34-
#if defined(mingw32_HOST_OS)
35-
import Control.Concurrent ( threadDelay )
36-
#else
37-
import Control.Concurrent (threadWaitRead, threadDelay)
38-
#endif
39-
import Control.Exception (Exception, throw)
28+
import qualified Database.PostgreSQL.LibPQ as PQ
29+
import Hasql.Connection (Connection, withLibPQConnection)
30+
import qualified Hasql.Decoders as HD
31+
import qualified Hasql.Encoders as HE
32+
import Hasql.Pool (Pool, UsageError, use)
33+
import Hasql.Session (run, sql, statement)
34+
import qualified Hasql.Session as S
35+
import qualified Hasql.Statement as HST
4036

4137
-- | A wrapped text that represents a properly escaped and quoted PostgreSQL identifier
4238
newtype PgIdentifier = PgIdentifier Text deriving (Show)
4339

4440
-- | Uncatchable exceptions thrown and never caught.
45-
newtype FatalError = FatalError { fatalErrorMessage :: String }
41+
newtype FatalError = FatalError {fatalErrorMessage :: String}
4642
deriving (Show)
4743

4844
instance Exception FatalError
@@ -60,101 +56,114 @@ toPgIdentifier x =
6056
strictlyReplaceQuotes = T.replace "\"" ("\"\"" :: Text)
6157

6258
-- | Given a Hasql Pool, a channel and a message sends a notify command to the database
63-
notifyPool :: Pool -- ^ Pool from which the connection will be used to issue a NOTIFY command.
64-
-> Text -- ^ Channel where to send the notification
65-
-> Text -- ^ Payload to be sent with the notification
66-
-> IO (Either UsageError ())
59+
notifyPool ::
60+
-- | Pool from which the connection will be used to issue a NOTIFY command.
61+
Pool ->
62+
-- | Channel where to send the notification
63+
Text ->
64+
-- | Payload to be sent with the notification
65+
Text ->
66+
IO (Either UsageError ())
6767
notifyPool pool channel mesg =
68-
use pool (statement (channel, mesg) callStatement)
69-
where
70-
callStatement = HST.Statement ("SELECT pg_notify" <> "($1, $2)") encoder HD.noResult False
71-
encoder = contramap fst (HE.param $ HE.nonNullable HE.text) <> contramap snd (HE.param $ HE.nonNullable HE.text)
68+
use pool (statement (channel, mesg) callStatement)
69+
where
70+
callStatement = HST.Statement ("SELECT pg_notify" <> "($1, $2)") encoder HD.noResult False
71+
encoder = contramap fst (HE.param $ HE.nonNullable HE.text) <> contramap snd (HE.param $ HE.nonNullable HE.text)
7272

7373
-- | Given a Hasql Connection, a channel and a message sends a notify command to the database
74-
notify :: Connection -- ^ Connection to be used to send the NOTIFY command
75-
-> PgIdentifier -- ^ Channel where to send the notification
76-
-> Text -- ^ Payload to be sent with the notification
77-
-> IO (Either S.QueryError ())
74+
notify ::
75+
-- | Connection to be used to send the NOTIFY command
76+
Connection ->
77+
-- | Channel where to send the notification
78+
PgIdentifier ->
79+
-- | Payload to be sent with the notification
80+
Text ->
81+
IO (Either S.QueryError ())
7882
notify con channel mesg =
79-
run (sql $ T.encodeUtf8 ("NOTIFY " <> fromPgIdentifier channel <> ", '" <> mesg <> "'")) con
80-
81-
{-|
82-
Given a Hasql Connection and a channel sends a listen command to the database.
83-
Once the connection sends the LISTEN command the server register its interest in the channel.
84-
Hence it's important to keep track of which connection was used to open the listen command.
85-
86-
Example of listening and waiting for a notification:
87-
88-
@
89-
import System.Exit (die)
90-
91-
import Hasql.Connection
92-
import Hasql.Notifications
93-
94-
main :: IO ()
95-
main = do
96-
dbOrError <- acquire "postgres://localhost/db_name"
97-
case dbOrError of
98-
Right db -> do
99-
let channelToListen = toPgIdentifier "sample-channel"
100-
listen db channelToListen
101-
waitForNotifications (\channel _ -> print $ "Just got notification on channel " <> channel) db
102-
_ -> die "Could not open database connection"
103-
@
104-
-}
105-
listen :: Connection -- ^ Connection to be used to send the LISTEN command
106-
-> PgIdentifier -- ^ Channel this connection will be registered to listen to
107-
-> IO ()
83+
run (sql $ T.encodeUtf8 ("NOTIFY " <> fromPgIdentifier channel <> ", '" <> mesg <> "'")) con
84+
85+
-- |
86+
-- Given a Hasql Connection and a channel sends a listen command to the database.
87+
-- Once the connection sends the LISTEN command the server register its interest in the channel.
88+
-- Hence it's important to keep track of which connection was used to open the listen command.
89+
--
90+
-- Example of listening and waiting for a notification:
91+
--
92+
-- @
93+
-- import System.Exit (die)
94+
--
95+
-- import Hasql.Connection
96+
-- import Hasql.Notifications
97+
--
98+
-- main :: IO ()
99+
-- main = do
100+
-- dbOrError <- acquire "postgres://localhost/db_name"
101+
-- case dbOrError of
102+
-- Right db -> do
103+
-- let channelToListen = toPgIdentifier "sample-channel"
104+
-- listen db channelToListen
105+
-- waitForNotifications (\channel _ -> print $ "Just got notification on channel " <> channel) db
106+
-- _ -> die "Could not open database connection"
107+
-- @
108+
listen ::
109+
-- | Connection to be used to send the LISTEN command
110+
Connection ->
111+
-- | Channel this connection will be registered to listen to
112+
PgIdentifier ->
113+
IO ()
108114
listen con channel =
109115
void $ withLibPQConnection con execListen
110116
where
111117
execListen pqCon = void $ PQ.exec pqCon $ T.encodeUtf8 $ "LISTEN " <> fromPgIdentifier channel
112118

113119
-- | Given a Hasql Connection and a channel sends a unlisten command to the database
114-
unlisten :: Connection -- ^ Connection currently registerd by a previous 'listen' call
115-
-> PgIdentifier -- ^ Channel this connection will be deregistered from
116-
-> IO ()
120+
unlisten ::
121+
-- | Connection currently registerd by a previous 'listen' call
122+
Connection ->
123+
-- | Channel this connection will be deregistered from
124+
PgIdentifier ->
125+
IO ()
117126
unlisten con channel =
118127
void $ withLibPQConnection con execListen
119128
where
120129
execListen pqCon = void $ PQ.exec pqCon $ T.encodeUtf8 $ "UNLISTEN " <> fromPgIdentifier channel
121130

122-
123-
{-|
124-
Given a function that handles notifications and a Hasql connection it will listen
125-
on the database connection and call the handler everytime a message arrives.
126-
127-
The message handler passed as first argument needs two parameters channel and payload.
128-
See an example of handling notification on a separate thread:
129-
130-
@
131-
import Control.Concurrent.Async (async)
132-
import Control.Monad (void)
133-
import System.Exit (die)
134-
135-
import Hasql.Connection
136-
import Hasql.Notifications
137-
138-
notificationHandler :: ByteString -> ByteString -> IO()
139-
notificationHandler channel payload =
140-
void $ async do
141-
print $ "Handle payload " <> payload <> " in its own thread"
142-
143-
main :: IO ()
144-
main = do
145-
dbOrError <- acquire "postgres://localhost/db_name"
146-
case dbOrError of
147-
Right db -> do
148-
let channelToListen = toPgIdentifier "sample-channel"
149-
listen db channelToListen
150-
waitForNotifications notificationHandler db
151-
_ -> die "Could not open database connection"
152-
@
153-
-}
154-
155-
waitForNotifications :: (ByteString -> ByteString -> IO()) -- ^ Callback function to handle incoming notifications
156-
-> Connection -- ^ Connection where we will listen to
157-
-> IO ()
131+
-- |
132+
-- Given a function that handles notifications and a Hasql connection it will listen
133+
-- on the database connection and call the handler everytime a message arrives.
134+
--
135+
-- The message handler passed as first argument needs two parameters channel and payload.
136+
-- See an example of handling notification on a separate thread:
137+
--
138+
-- @
139+
-- import Control.Concurrent.Async (async)
140+
-- import Control.Monad (void)
141+
-- import System.Exit (die)
142+
--
143+
-- import Hasql.Connection
144+
-- import Hasql.Notifications
145+
--
146+
-- notificationHandler :: ByteString -> ByteString -> IO()
147+
-- notificationHandler channel payload =
148+
-- void $ async do
149+
-- print $ "Handle payload " <> payload <> " in its own thread"
150+
--
151+
-- main :: IO ()
152+
-- main = do
153+
-- dbOrError <- acquire "postgres://localhost/db_name"
154+
-- case dbOrError of
155+
-- Right db -> do
156+
-- let channelToListen = toPgIdentifier "sample-channel"
157+
-- listen db channelToListen
158+
-- waitForNotifications notificationHandler db
159+
-- _ -> die "Could not open database connection"
160+
-- @
161+
waitForNotifications ::
162+
-- | Callback function to handle incoming notifications
163+
(ByteString -> ByteString -> IO ()) ->
164+
-- | Connection where we will listen to
165+
Connection ->
166+
IO ()
158167
waitForNotifications sendNotification con =
159168
withLibPQConnection con $ void . forever . pqFetch
160169
where
@@ -164,11 +173,7 @@ waitForNotifications sendNotification con =
164173
Nothing -> do
165174
mfd <- PQ.socket pqCon
166175
case mfd of
167-
Nothing -> void $ threadDelay 1000000
168-
#if defined(mingw32_HOST_OS)
169-
Just _ -> do
170-
void $ threadDelay 1000000
171-
#else
176+
Nothing -> void $ threadDelay 1000000
172177
Just fd -> do
173178
void $ threadWaitRead fd
174179

@@ -177,6 +182,6 @@ waitForNotifications sendNotification con =
177182
mError <- PQ.errorMessage pqCon
178183
panic $ maybe "Error checking for PostgreSQL notifications" show mError
179184
Just notification ->
180-
sendNotification (PQ.notifyRelname notification) (PQ.notifyExtra notification)
185+
sendNotification (PQ.notifyRelname notification) (PQ.notifyExtra notification)
181186
panic :: String -> a
182187
panic a = throw (FatalError a)

0 commit comments

Comments
 (0)