8000 Dispatch notification by default dispatcher · postgres-haskell/postgres-wire@d758bad · GitHub
[go: up one dir, main page]
Skip to content
8000

Commit d758bad

Browse files
VyacheslavHashovVyacheslavHashov
committed
Dispatch notification by default dispatcher
1 parent 5bc8274 commit d758bad

File tree

2 files changed

+22
-18
lines changed

2 files changed

+22
-18
lines changed

‎src/Database/PostgreSQL/Driver/Connection.hs

Lines changed: 20 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -67,16 +67,15 @@ data ConnectionParameters = ConnectionParameters
6767

6868
-- | Public
6969
data Connection = Connection
70-
{ connRawConnection :: RawConnection
71-
, connReceiverThread :: ThreadId
70+
{ connRawConnection :: RawConnection
71+
, connReceiverThread :: ThreadId
7272
-- channel only for Data messages
73-
, connOutDataChan :: OutChan (Either Error DataMessage)
73+
, connOutDataChan :: OutChan (Either Error DataMessage)
7474
-- channel for all the others messages
75-
, connOutAllChan :: OutChan ServerMessage
76-
, connStatementStorage :: StatementStorage
77-
, connParameters :: ConnectionParameters
78-
, connMode :: IORef ConnectionMode
79-
, connNotificationHandler :: NotificationHandler
75+
, connOutAllChan :: OutChan ServerMessage
76+
, connStatementStorage :: StatementStorage
77+
, connParameters :: ConnectionParameters
78+
, connMode :: IORef ConnectionMode
8079
}
8180

8281
-- | Public
@@ -107,6 +106,7 @@ buildConnection rawConn connParams msgFilter = do
107106

108107
tid <- forkIO $
109108
receiverThread msgFilter rawConn inDataChan inAllChan modeRef
109+
defaultNotificationHandler
110110
pure Connection
111111
{ connRawConnection = rawConn
112112
, connReceiverThread = tid
@@ -115,7 +115,6 @@ buildConnection rawConn connParams msgFilter = do
115115
, connStatementStorage = storage
116116
, connParameters = connParams
117117
, connMode = modeRef
118-
, connNotificationHandler = defaultNotificationHandler
119118
}
120119

121120
-- | Authorizes on the server and reads connection parameters.
@@ -203,8 +202,10 @@ receiverThread
203202
-> InChan (Either Error DataMessage)
204203
-> InChan ServerMessage
205204
-> IORef ConnectionMode
205+
-> NotificationHandler
206206
-> IO ()
207-
receiverThread msgFilter rawConn dataChan allChan modeRef = receiveLoop []
207+
receiverThread msgFilter rawConn dataChan allChan modeRef ntfHandler =
208+
receiveLoop []
208209
where
209210
receiveLoop :: [V.Vector (Maybe B.ByteString)] -> IO ()
210211
receiveLoop acc = do
@@ -215,23 +216,25 @@ receiverThread msgFilter rawConn dataChan allChan modeRef = receiveLoop []
215216
go :: B.ByteString -> [V.Vector (Maybe B.ByteString)] -> IO [V.Vector (Maybe B.ByteString)]
216217
go str acc = case runDecode decodeServerMessage str of
217218
Right (rest, v) -> do
219+
dispatchIfNotification v
218220
when (msgFilter v) $ writeChan allChan v
219221
mode <- readIORef modeRef
220222
newAcc <- dispatch mode dataChan v acc
221223
if B.null rest
222224
then pure newAcc
223225
else go rest newAcc
224226
Left reason -> error reason
227+
dispatchIfNotification (NotificationResponse n) = ntfHandler n
228+
dispatchIfNotification _ = pure ()
229+
225230

226231
dispatch :: ConnectionMode -> Dispatcher
227232
dispatch SimpleQueryMode = dispatchSimple
228233
dispatch ExtendedQueryMode = dispatchExtended
229234

230235
-- | Dispatcher for the SimpleQuery mode.
231236
dispatchSimple :: Dispatcher
232-
dispatchSimple dataChan message acc = case message of
233-
NotificationResponse n -> pure acc
234-
_ -> pure acc
237+
dispatchSimple dataChan message = pure
235238

236239
-- | Dispatcher for the ExtendedQuery mode.
237240
dispatchExtended :: Dispatcher
@@ -247,15 +250,14 @@ dispatchExtended dataChan message acc = case message of
247250
EmptyQueryResponse -> do
248251
writeChan dataChan . Right . DataMessage $ reverse acc
249252
pure []
250-
-- On ErrorResponse we should discard all the collected datarows
253+
-- On ErrorResponse we should discard all the collected datarows.
251254
ErrorResponse desc -> do
252255
writeChan dataChan $ Left $ PostgresError desc
253256
pure []
254-
-- TODO handle notifications
255-
NotificationResponse n -> pure acc
256-
-- We does not handled this case because we always send `execute`
257+
-- We does not handled `PortalSuspended` because we always send `execute`
257258
-- with no limit.
258-
PortalSuspended -> pure acc
259+
-- PortalSuspended -> pure acc
260+
259261
-- do nothing on other messages
260262
_ -> pure acc
261263

‎src/Database/PostgreSQL/Protocol/Decoders.hs

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -135,6 +135,7 @@ decodeCommandResult s =
135135
(pure . RowsCount . fromInteger . fst)
136136
. readInteger . B.dropWhile (== space)
137137

138+
-- Helper to parse, not used by decoder itself
138139
parseServerVersion :: B.ByteString -> Maybe ServerVersion
139140
parseServerVersion bs =
140141
let (numbersStr, desc) = B.span isDigitDot bs
@@ -148,6 +149,7 @@ parseServerVersion bs =
148149
| c >= 48 && c < 58 = True -- digits
149150
| otherwise = False
150151

152+
-- Helper to parse, not used by decoder itself
151153
parseIntegerDatetimes :: B.ByteString -> Bool
152154
parseIntegerDatetimes bs | bs == "on" || bs == "yes" || bs == "1" = True
153155
| otherwise = False

0 commit comments

Comments
 (0)
0