8000 Simple session works · postgres-haskell/postgres-wire@fb0de46 · GitHub
[go: up one dir, main page]
Skip to content

Commit fb0de46

Browse files
VyacheslavHashovVyacheslavHashov
committed
Simple session works
1 parent 7f994e0 commit fb0de46

File tree

3 files changed

+129
-50
lines changed

3 files changed

+129
-50
lines changed

‎postgres-wire.cabal

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,7 @@ library
2020
, Database.PostgreSQL.Settings
2121
, Database.PostgreSQL.StatementStorage
2222
, Database.PostgreSQL.Types
23+
, Database.PostgreSQL.Session
2324

2425
, Database.PostgreSQL.Protocol.Types
2526
, Database.PostgreSQL.Protocol.Encoders
@@ -35,6 +36,7 @@ library
3536
, hashtables
3637
, unagi-chan
3738
, unordered-containers
39+
, postgresql-binary
3840
default-language: Haskell2010
3941
default-extensions:
4042
OverloadedStrings

‎src/Database/PostgreSQL/Connection.hs

Lines changed: 41 additions & 36 deletions
Original file line numberDiff line numberDiff line change
@@ -217,52 +217,23 @@ data Query = Query
217217
, qResultFormat :: Format
218218
} deriving (Show)
219219

220-
query1 = Query "SELECT $1 + $2" [Oid 23, Oid 23] ["1", "3"] Text Text
221-
query2 = Query "SELECT $1 + $2" [Oid 23, Oid 23] ["a", "3"] Text Text
222-
query3 = Query "SELECT $1 + $2" [Oid 23, Oid 23] ["3", "3"] Text Text
223-
query4 = Query "SELECT $1 + $2" [Oid 23, Oid 23] ["4", "3"] Text Text
224-
225220
sendBatch :: Connection -> [Query] -> IO ()
226-
sendBatch conn qs = do
227-
traverse sendSingle $ take 5 qs
228-
sendMessage s Sync
221+
sendBatch conn = traverse_ sendSingle
229222
where
230223
s = connSocket conn
224+
sname = StatementName ""
225+
pname = PortalName ""
231226
sendSingle q = do
232-
let sname = StatementName ""
233-
pname = PortalName ""
234227
sendMessage s $ Parse sname (StatementSQL $ qStatement q) (qOids q)
235228
sendMessage s $
236229
Bind pname sname (qParamsFormat q) (qValues q) (qResultFormat q)
237230
sendMessage s $ Execute pname noLimitToReceive
238231

232+
sendSync :: Connection -> IO ()
233+
sendSync conn = sendMessage (connSocket conn) Sync
239234

240-
test :: IO ()
241-
test = do
242-
c <- connect defaultConnectionSettings
243-
sendBatch c queries
244-
readResults c $ length queries
245-
readReadyForQuery c >>= print
246-
close c
247-
where
248-
queries = [query1, query2, query3, query4 ]
249-
readResults c 0 = pure ()
250-
readResults c n = do
251-
r <- readNextData c
252-
print r
253-
case r of
254-
Left _ -> pure ()
255-
Right _ -> readResults c $ n - 1
256-
257-
-- sendBatchAndSync :: IsQuery a => [a] -> Connection -> IO ()
258-
-- sendBatchAndSync = undefined
259-
260-
-- sendBatchAndFlush :: IsQuery a => [a] -> Connection -> IO ()
261-
-- sendBatchAndFlush = undefined
262-
263-
-- internal helper
264-
-- sendBatch :: IsQuery a => [a] -> Connection -> IO ()
265-
-- sendBatch = undefined
235+
sendFlush :: Connection -> IO ()
236+
sendFlush conn = sendMessage (connSocket conn) Flush
266237

267238
readNextData :: Connection -> IO (Either Error DataMessage)
268239
readNextData conn = readChan $ connOutDataChan conn
@@ -308,6 +279,40 @@ describeStatement conn stmt = do
308279
xs -> maybe (error "Impossible happened") (Left . PostgresError )
309280
$ findFirstError xs
310281

282+
query1 = Query "SELECT $1 + $2" [Oid 23, Oid 23] ["1", "3"] Text Text
283+
query2 = Query "SELECT $1 + $2" [Oid 23, Oid 23] ["a", "3"] Text Text
284+
query3 = Query "SELECT $1 + $2" [Oid 23, Oid 23] ["3", "3"] Text Text
285+
query4 = Query "SELECT $1 + $2" [Oid 23, Oid 23] ["4", "3"] Text Text
286+
287+
288+
test :: IO ()
289+
test = do
290+
c <- connect defaultConnectionSettings
291+
sendBatch c queries
292+
readResults c $ length queries
293+
readReadyForQuery c >>= print
294+
close c
295+
where
296+
queries = [query1, query2, query3, query4 ]
297+
readResults c 0 = pure ()
298+
readResults c n = do
299+
r <- readNextData c
300+
print r
301+
case r of
302+
Left _ -> pure ()
303+
Right _ -> readResults c $ n - 1
304+
305+
-- sendBatchAndSync :: IsQuery a => [a] -> Connection -> IO ()
306+
-- sendBatchAndSync = undefined
307+
308+
-- sendBatchAndFlush :: IsQuery a => [a] -> Connection -> IO ()
309+
-- sendBatchAndFlush = undefined
310+
311+
-- internal helper
312+
-- sendBatch :: IsQuery a => [a] -> Connection -> IO ()
313+
-- sendBatch = undefined
314+
315+
311316
testDescribe1 :: IO ()
312317
testDescribe1 = do
313318
c <- connect defaultConnectionSettings

‎src/Database/PostgreSQL/Session.hs

Lines changed: 86 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -1,10 +1,33 F438 @@
1+
{-# language ApplicativeDo #-}
2+
{-# language OverloadedLists #-}
3+
{-# language OverloadedStrings #-}
4+
{-# language ExistentialQuantification #-}
5+
{-# language TypeSynonymInstances #-}
6+
{-# language FlexibleInstances #-}
7+
module Database.PostgreSQL.Session where
8+
9+
import Control.Monad
10+
import Control.Applicative
11+
import Data.Monoid
12+
import Data.Int
13+
import Data.ByteString (ByteString)
14+
import Data.Either
15+
import qualified Data.Vector as V
16+
17+
import PostgreSQL.Binary.Encoder (int8_int64, run)
18+
import qualified PostgreSQL.Binary.Decoder as D(int, run)
19+
20+
import Database.PostgreSQL.Protocol.Types
21+
import Database.PostgreSQL.Connection
22+
import Database.PostgreSQL.Settings
23+
124
data Count = One | Many
225
deriving (Eq, Show)
326

427
data Session a
528
= Done a
629
| forall r . Decode r => Receive (r -> Session a)
7-
| Send Count [Request] (Session a)
30+
| Send Count [Query] (Session a)
831

932
instance Functor Session where
1033
f `fmap` (Done a) = Done $ f a
@@ -43,21 +66,70 @@ instance Monad Session where
4366

4467
(>>) = (*>)
4568

46-
runSession :: Show a => Connection -> Session a -> IO a
47-
runSession conn@(Connection sock _ chan) = go
69+
class Encode a where
70+
encode :: a -> ByteString
71+
getOid :: a -> Oid
72+
73+
class Decode a where
74+
decode :: ByteString -> a
75+
76+
instance Encode Int64 where
77+
encode = run int8_int64
78+
getOid _ = Oid 20
79+
80+
instance Decode Int64 where
81+
decode = fromRight . D.run D.int
82+
where
83+
fromRight (Right v) = v
84+
fromRight _ = error "bad fromRight"
85+
86+
data SessionQuery a b = SessionQuery { sqStatement :: ByteString }
87+
deriving (Show)
88+
89+
query :: (Encode a, Decode b) => SessionQuery a b -> a -> Session b
90+
query sq val =
91+
let q = Query { qStatement = sqStatement sq
92+
, qOids = [getOid val]
93+
, qValues = [encode val]
94+
, qParamsFormat = Binary
95+
, qResultFormat = Binary }
96+
in Send One [q] $ Receive Done
97+
98+
runSession :: Show a => Connection -> Session a -> IO (Either Error a)
99+
runSession conn = go 0
48100
where
49-
go (Done x) = do
101+
go n (Done x) = do
50102
putStrLn $ "Return " ++ show x
51-
pure x
52-
go (Receive f) = do
103+
when (n > 0) $ void $ sendSync conn >> readReadyForQuery conn
104+
pure $ Right x
105+
go n (Receive f) = do
53106
putStrLn "Receiving"
54-
-- TODO receive here
55-
-- x <- receive
56-
x <- getLine
57-
go (f $ decode x)
58-
go (Send _ rs c) = do
107+
r <- readNextData conn
108+
case r of
109+
Left e -> pure $ Left e
110+
Right (DataMessage rows) -> go n (f $ decode $ V.head $ head rows)
111+
go n (Send _ qs c) = do
59112
putStrLn "Sending requests "
60-
-- TODO send requests here in batch
61-
sendBatch conn rs
62-
go c
113+
sendBatch conn qs
114+
sendFlush conn
115+
go (n + 1) c
116+
117+
q1 :: SessionQuery Int64 Int64
118+
q1 = SessionQuery "SELECT $1"
119+
120+
q2 :: SessionQuery Int64 Int64
121+
q2 = SessionQuery "SELECT count(*) from a where v < $1"
122+
123+
q3 :: SessionQuery Int64 Int64
124+
q3 = SessionQuery "SELECT 5 + $1"
125+
126+
testSession :: IO ()
127+
testSession = do
128+
c <- connect defaultConnectionSettings
129+
r <- runSession c $ do
130+
b <- query q1 10
131+
a <- query q2 b
132+
query q3 a
133+
print r
134+
close c
63135

0 commit comments

Comments
 (0)
0