a1d63ca · project-m36 · git.iko.soy

git.iko.soy/project-m36

on master help sign in

WIP notification generation machinery

a1d63caAgentM11 years ago

ProjectM36/Base.hs+1−1

compare with master

unifiedsplit

@@ -254,7 +254,7 @@ data Notification = Notification {

254 changeExpr :: RelationalExpr,

255 reportExpr :: RelationalExpr

256 }

257− deriving (Show, Eq)

257+ deriving (Show, Eq, Binary, Generic)

258

259-- | The DatabaseContext is a snapshot of a database's evolving state and contains everything a database client can change over time.

260data DatabaseContext = DatabaseContext {

ProjectM36/Client.hs+99−44

@@ -1,4 +1,4 @@

1−{-# LANGUAGE OverloadedStrings #-}

1+{-# LANGUAGE OverloadedStrings, DeriveAnyClass, DeriveGeneric #-}

2module ProjectM36.Client

3 (ConnectionInfo(..),

4 Connection(..),

@@ -34,6 +34,7 @@ module ProjectM36.Client

34 createSessionAtCommit,

35 createSessionAtHead,

36 closeSession,

37+ addClientNode,

37 TransactionGraphOperator(..),

38 TransactionGraphOperator(..),

38 Atomable,

39 Atomable,

39 NodeId(..),

40 NodeId(..),

@@ -51,18 +52,19 @@ import ProjectM36.TransactionGraph

51import ProjectM36.TransactionGraph.Persist

52import ProjectM36.TransactionGraph.Persist

52import ProjectM36.Attribute

53import ProjectM36.Attribute

53import ProjectM36.Persist (DiskSync(..))

54import ProjectM36.Persist (DiskSync(..))

55+import ProjectM36.Notifications

54import ProjectM36.Server.RemoteCallTypes

56import ProjectM36.Server.RemoteCallTypes

55import Network.Transport.TCP (createTransport, defaultTCPParameters, encodeEndPointAddress)

57import Network.Transport.TCP (createTransport, defaultTCPParameters, encodeEndPointAddress)

56import Control.Distributed.Process.Node (newLocalNode, initRemoteTable, runProcess, LocalNode)

58import Control.Distributed.Process.Node (newLocalNode, initRemoteTable, runProcess, LocalNode)

57import Control.Distributed.Process.Extras.Internal.Types (whereisRemote)

59import Control.Distributed.Process.Extras.Internal.Types (whereisRemote)

58import Control.Distributed.Process.ManagedProcess.Client (call, safeCall)

60import Control.Distributed.Process.ManagedProcess.Client (call, safeCall)

59−import Control.Distributed.Process (NodeId(..))

61+import Control.Distributed.Process (NodeId(..), getSelfPid)

61import Data.UUID (UUID)

63import Data.UUID (UUID)

62import Data.UUID.V4 (nextRandom)

64import Data.UUID.V4 (nextRandom)

63import Control.Concurrent.STM

65import Control.Concurrent.STM

64import Data.Word

66import Data.Word

65−import Control.Distributed.Process (ProcessId, Process)

67+import Control.Distributed.Process (ProcessId, Process, receiveWait, send, match)

66import Control.Exception (IOException)

68import Control.Exception (IOException)

67import Control.Concurrent.MVar

69import Control.Concurrent.MVar

68import qualified Data.Map as M

70import qualified Data.Map as M

@@ -73,6 +75,8 @@ import qualified STMContainers.Set as STMSet

73import ProjectM36.Session

75import ProjectM36.Session

74import ProjectM36.Sessions

76import ProjectM36.Sessions

75import ListT

77import ListT

78+import Data.Binary (Binary)

79+import GHC.Generics (Generic)

77type Hostname = String

81type Hostname = String

@@ -80,9 +84,23 @@ type Port = Word16

81type DatabaseName = String

85type DatabaseName = String

87+-- | Construct a 'ConnectionInfo' to describe how to make the 'Connection'.

83data ConnectionInfo = InProcessConnectionInfo PersistenceStrategy |

88data ConnectionInfo = InProcessConnectionInfo PersistenceStrategy |

84 RemoteProcessConnectionInfo DatabaseName NodeId

89 RemoteProcessConnectionInfo DatabaseName NodeId

91+type EvaluatedNotifications = M.Map NotificationName EvaluatedNotification

92+

93+-- | Used for callbacks from the server when monitored changes have been made.

94+data NotificationMessage = NotificationMessage EvaluatedNotifications

95+ deriving (Binary, Eq, Show, Generic)

96+

97+-- | When a notification is fired, the 'reportExpr' is evaluated in the commit's context, so that is returned along with the original notification.

98+data EvaluatedNotification = EvaluatedNotification {

99+ notification :: Notification,

100+ reportRelation :: Either RelationalError Relation

101+ }

102+ deriving(Binary, Eq, Show, Generic)

103+

86createNodeId :: Hostname -> Port -> NodeId

104createNodeId :: Hostname -> Port -> NodeId

87createNodeId host port = NodeId $ encodeEndPointAddress host (show port) 0

105createNodeId host port = NodeId $ encodeEndPointAddress host (show port) 0

106

@@ -121,6 +139,21 @@ commonLocalNode = do

121 case eLocalTransport of

139 case eLocalTransport of

122 Left err -> pure (Left $ IOExceptionError err)

140 Left err -> pure (Left $ IOExceptionError err)

123 Right localTransport -> newLocalNode localTransport initRemoteTable >>= pure . Right

141 Right localTransport -> newLocalNode localTransport initRemoteTable >>= pure . Right

142+

143+notificationListener :: Process ()

144+notificationListener = do

145+ _ <- receiveWait [

146+ match (\(NotificationMessage eNots) -> undefined)

147+ ]

148+ liftIO $ putStrLn $ "LISTENER THREAD EXIT"

149+ pure ()

150+

151+startNotificationListener :: IO (ProcessId)

152+startNotificationListener = do

153+ eLocalNode <- commonLocalNode

154+ case eLocalNode of

155+ Left err -> error "Failed to start local notification listener"

156+ Right localNode -> forkProcess localNode notificationListener

124

157

125-- | To create a 'Connection' to a remote or local database, create a connectionInfo and call 'connectProjectM36'.

158-- | To create a 'Connection' to a remote or local database, create a connectionInfo and call 'connectProjectM36'.

126connectProjectM36 :: ConnectionInfo -> IO (Either ConnectionError Connection)

159connectProjectM36 :: ConnectionInfo -> IO (Either ConnectionError Connection)

@@ -135,10 +168,10 @@ connectProjectM36 (InProcessConnectionInfo strat) = do

135 graphTvar <- newTVarIO freshGraph

168 graphTvar <- newTVarIO freshGraph

136 clientNodes <- STMSet.newIO

169 clientNodes <- STMSet.newIO

137 sessions <- STMMap.newIO

170 sessions <- STMMap.newIO

171+ notProcId <- startNotificationListener

138 return $ Right $ InProcessConnection strat clientNodes sessions graphTvar

172 return $ Right $ InProcessConnection strat clientNodes sessions graphTvar

139 MinimalPersistence dbdir -> connectPersistentProjectM36 strat NoDiskSync dbdir freshGraph

173 MinimalPersistence dbdir -> connectPersistentProjectM36 strat NoDiskSync dbdir freshGraph

140 CrashSafePersistence dbdir -> connectPersistentProjectM36 strat FsyncDiskSync dbdir freshGraph

174 CrashSafePersistence dbdir -> connectPersistentProjectM36 strat FsyncDiskSync dbdir freshGraph

141−

142

175

143connectProjectM36 (RemoteProcessConnectionInfo databaseName serverNodeId) = do

176connectProjectM36 (RemoteProcessConnectionInfo databaseName serverNodeId) = do

144 connStatus <- newEmptyMVar

177 connStatus <- newEmptyMVar

@@ -150,10 +183,11 @@ connectProjectM36 (RemoteProcessConnectionInfo databaseName serverNodeId) = do

150 Right localNode -> do

183 Right localNode -> do

151 runProcess localNode $ do

184 runProcess localNode $ do

152 mServerProcessId <- whereisRemote serverNodeId dbName

185 mServerProcessId <- whereisRemote serverNodeId dbName

186+ selfPid <- getSelfPid

153 case mServerProcessId of

187 case mServerProcessId of

154 Nothing -> liftIO $ putMVar connStatus $ Left (NoSuchDatabaseByNameError databaseName)

188 Nothing -> liftIO $ putMVar connStatus $ Left (NoSuchDatabaseByNameError databaseName)

155 Just serverProcessId -> do

189 Just serverProcessId -> do

156− loginConfirmation <- call serverProcessId (Login loginNode)

190+ loginConfirmation <- call serverProcessId (Login selfPid)

157 if not loginConfirmation then

191 if not loginConfirmation then

158 liftIO $ putMVar connStatus (Left LoginError)

192 liftIO $ putMVar connStatus (Left LoginError)

159 else do

193 else do

@@ -177,18 +211,19 @@ connectPersistentProjectM36 strat sync dbdir freshGraph = do

177 Right graph' -> do

211 Right graph' -> do

178 tvarGraph <- newTVarIO graph'

212 tvarGraph <- newTVarIO graph'

179 sessions <- STMMap.newIO

213 sessions <- STMMap.newIO

214+ clientNodes <- STMSet.newIO

180− return $ Right $ InProcessConnection strat sessions tvarGraph

215+ return $ Right $ InProcessConnection strat clientNodes sessions tvarGraph

181

216

182-- | Create a new session at the transaction UUID and return the session's Id.

217-- | Create a new session at the transaction UUID and return the session's Id.

183createSessionAtCommit :: UUID -> Connection -> IO (Either RelationalError SessionId)

218createSessionAtCommit :: UUID -> Connection -> IO (Either RelationalError SessionId)

184−createSessionAtCommit commitUUID conn@(InProcessConnection _ _ _) = do

219+createSessionAtCommit commitUUID conn@(InProcessConnection _ _ _ _) = do

185 newSessionId <- nextRandom

220 newSessionId <- nextRandom

186 atomically $ do

221 atomically $ do

187 createSessionAtCommit_ commitUUID newSessionId conn

222 createSessionAtCommit_ commitUUID newSessionId conn

188createSessionAtCommit uuid conn@(RemoteProcessConnection _ _) = remoteCall conn (CreateSessionAtCommit uuid)

223createSessionAtCommit uuid conn@(RemoteProcessConnection _ _) = remoteCall conn (CreateSessionAtCommit uuid)

189

224

190createSessionAtCommit_ :: UUID -> SessionId -> Connection -> STM (Either RelationalError SessionId)

225createSessionAtCommit_ :: UUID -> SessionId -> Connection -> STM (Either RelationalError SessionId)

191−createSessionAtCommit_ commitUUID newSessionId (InProcessConnection _ sessions graphTvar) = do

226+createSessionAtCommit_ commitUUID newSessionId (InProcessConnection _ _ sessions graphTvar) = do

192 graph <- readTVar graphTvar

227 graph <- readTVar graphTvar

193 case transactionForUUID commitUUID graph of

228 case transactionForUUID commitUUID graph of

194 Left err -> pure (Left err)

229 Left err -> pure (Left err)

@@ -204,32 +239,31 @@ createSessionAtCommit_ _ _ (RemoteProcessConnection _ _) = error "createSessionA

204

239

205-- | Call 'createSessionAtHead' with a transaction graph's head's name to create a new session pinned to that head. This function returns a 'SessionId' which can be used in other function calls to reference the point in the transaction graph.

240-- | Call 'createSessionAtHead' with a transaction graph's head's name to create a new session pinned to that head. This function returns a 'SessionId' which can be used in other function calls to reference the point in the transaction graph.

206createSessionAtHead :: HeadName -> Connection -> IO (Either RelationalError SessionId)

241createSessionAtHead :: HeadName -> Connection -> IO (Either RelationalError SessionId)

207−createSessionAtHead headn conn@(InProcessConnection _ _ tvar) = do

242+createSessionAtHead headn conn@(InProcessConnection _ _ _ graphTvar) = do

208 newSessionId <- nextRandom

243 newSessionId <- nextRandom

209 atomically $ do

244 atomically $ do

210− graph <- readTVar tvar

245+ graph <- readTVar graphTvar

211 case transactionForHead headn graph of

246 case transactionForHead headn graph of

212 Nothing -> pure $ Left (NoSuchHeadNameError headn)

247 Nothing -> pure $ Left (NoSuchHeadNameError headn)

213 Just trans -> createSessionAtCommit_ (transactionUUID trans) newSessionId conn

248 Just trans -> createSessionAtCommit_ (transactionUUID trans) newSessionId conn

214createSessionAtHead headn conn@(RemoteProcessConnection _ _) = remoteCall conn (CreateSessionAtHead headn)

249createSessionAtHead headn conn@(RemoteProcessConnection _ _) = remoteCall conn (CreateSessionAtHead headn)

215

250

216−-- | Used internall for server connections to keep track of remote nodes for the purpose of sending notifications later.

251+-- | Used internally for server connections to keep track of remote nodes for the purpose of sending notifications later.

217addClientNode :: Connection -> ProcessId -> IO ()

252addClientNode :: Connection -> ProcessId -> IO ()

218−addClientNode (RemoteProcessConnection _ _) = error "addClientNode called on remote connection"

253+addClientNode (RemoteProcessConnection _ _) _ = error "addClientNode called on remote connection"

219addClientNode (InProcessConnection _ clientNodes _ _) newProcessId = atomically (STMSet.insert newProcessId clientNodes)

254addClientNode (InProcessConnection _ clientNodes _ _) newProcessId = atomically (STMSet.insert newProcessId clientNodes)

220

255

221−

222-- | Discards a session, eliminating any uncommitted changes present in the session.

256-- | Discards a session, eliminating any uncommitted changes present in the session.

223closeSession :: SessionId -> Connection -> IO ()

257closeSession :: SessionId -> Connection -> IO ()

224−closeSession sessionId (InProcessConnection _ sessions _) = do

258+closeSession sessionId (InProcessConnection _ _ sessions _) = do

225− atomically $ do

226− STMMap.delete sessionId sessions

259+ atomically $ STMMap.delete sessionId sessions

227closeSession sessionId conn@(RemoteProcessConnection _ _) = remoteCall conn (CloseSession sessionId)

260closeSession sessionId conn@(RemoteProcessConnection _ _) = remoteCall conn (CloseSession sessionId)

228-- | 'close' cleans up the database access connection. Note that sessions persist even after the connection is closed.

261-- | 'close' cleans up the database access connection. Note that sessions persist even after the connection is closed.

229close :: Connection -> IO ()

262close :: Connection -> IO ()

230−close (InProcessConnection _ sessions _) = atomically $ do

263+close (InProcessConnection _ _ sessions _) = atomically $ do

231 traverse_ (\(k,_) -> STMMap.delete k sessions) (STMMap.stream sessions)

264 traverse_ (\(k,_) -> STMMap.delete k sessions) (STMMap.stream sessions)

232 pure ()

265 pure ()

266+

233close (RemoteProcessConnection localNode serverProcessId) = do

267close (RemoteProcessConnection localNode serverProcessId) = do

234 runProcessResult localNode $ do

268 runProcessResult localNode $ do

235 call serverProcessId Logout

269 call serverProcessId Logout

@@ -243,7 +277,7 @@ runProcessResult localNode proc = do

243 takeMVar ret

277 takeMVar ret

244

278

245remoteCall :: (Serializable a, Serializable b) => Connection -> a -> IO b

279remoteCall :: (Serializable a, Serializable b) => Connection -> a -> IO b

246−remoteCall (InProcessConnection _ _ _) _ = error "remoteCall called on local connection"

280+remoteCall (InProcessConnection _ _ _ _) _ = error "remoteCall called on local connection"

247remoteCall (RemoteProcessConnection localNode serverProcessId) arg = runProcessResult localNode $ do

281remoteCall (RemoteProcessConnection localNode serverProcessId) arg = runProcessResult localNode $ do

248 ret <- safeCall serverProcessId arg

282 ret <- safeCall serverProcessId arg

249 case ret of

283 case ret of

@@ -257,7 +291,7 @@ sessionForSessionId sessionId sessions = do

257

291

258-- | Execute a relational expression in the context of the session and connection. Relational expressions are queries and therefore cannot alter the database.

292-- | Execute a relational expression in the context of the session and connection. Relational expressions are queries and therefore cannot alter the database.

259executeRelationalExpr :: SessionId -> Connection -> RelationalExpr -> IO (Either RelationalError Relation)

293executeRelationalExpr :: SessionId -> Connection -> RelationalExpr -> IO (Either RelationalError Relation)

260−executeRelationalExpr sessionId (InProcessConnection _ sessions _) expr = atomically $ do

294+executeRelationalExpr sessionId (InProcessConnection _ _ sessions _) expr = atomically $ do

261 eSession <- sessionForSessionId sessionId sessions

295 eSession <- sessionForSessionId sessionId sessions

262 case eSession of

296 case eSession of

263 Left err -> pure $ Left err

297 Left err -> pure $ Left err

@@ -266,7 +300,7 @@ executeRelationalExpr sessionId conn@(RemoteProcessConnection _ _) relExpr = rem

266

300

267-- | Execute a database context expression in the context of the session and connection. Database expressions modify the current session's disconnected transaction but cannot modify the transaction graph.

301-- | Execute a database context expression in the context of the session and connection. Database expressions modify the current session's disconnected transaction but cannot modify the transaction graph.

268executeDatabaseContextExpr :: SessionId -> Connection -> DatabaseExpr -> IO (Maybe RelationalError)

302executeDatabaseContextExpr :: SessionId -> Connection -> DatabaseExpr -> IO (Maybe RelationalError)

269−executeDatabaseContextExpr sessionId (InProcessConnection _ sessions _) expr = atomically $ do

303+executeDatabaseContextExpr sessionId (InProcessConnection _ _ sessions _) expr = atomically $ do

270 eSession <- sessionForSessionId sessionId sessions

304 eSession <- sessionForSessionId sessionId sessions

271 case eSession of

305 case eSession of

272 Left err -> pure $ Just err

306 Left err -> pure $ Just err

@@ -279,9 +313,8 @@ executeDatabaseContextExpr sessionId (InProcessConnection _ sessions _) expr = a

279 return Nothing

313 return Nothing

280executeDatabaseContextExpr sessionId conn@(RemoteProcessConnection _ _) dbExpr = remoteCall conn (ExecuteDatabaseContextExpr sessionId dbExpr)

314executeDatabaseContextExpr sessionId conn@(RemoteProcessConnection _ _) dbExpr = remoteCall conn (ExecuteDatabaseContextExpr sessionId dbExpr)

281

315

282−executeGraphExprSTM_ :: SessionId -> Sessions -> TransactionGraphOperator -> TVar TransactionGraph -> STM (Either RelationalError TransactionGraph)

316+executeGraphExprSTM_ :: UUID -> SessionId -> Sessions -> TransactionGraphOperator -> TVar TransactionGraph -> STM (Either RelationalError TransactionGraph)

283−executeGraphExprSTM_ sessionId sessions graphExpr graphTvar = do

317+executeGraphExprSTM_ freshUUID sessionId sessions graphExpr graphTvar = do

284− manip <- atomically $ do

285 graph <- readTVar graphTvar

318 graph <- readTVar graphTvar

286 eSession <- sessionForSessionId sessionId sessions

319 eSession <- sessionForSessionId sessionId sessions

287 case eSession of

320 case eSession of

@@ -293,12 +326,13 @@ executeGraphExprSTM_ sessionId sessions graphExpr graphTvar = do

293 let newSession = Session discon'

326 let newSession = Session discon'

294 STMMap.insert newSession sessionId sessions

327 STMMap.insert newSession sessionId sessions

295 pure $ Right graph'

328 pure $ Right graph'

329+

296

330

297-- | Execute a transaction graph expression in the context of the session and connection. Transaction graph operators modify the transaction graph state.

331-- | Execute a transaction graph expression in the context of the session and connection. Transaction graph operators modify the transaction graph state.

298executeGraphExpr :: SessionId -> Connection -> TransactionGraphOperator -> IO (Maybe RelationalError)

332executeGraphExpr :: SessionId -> Connection -> TransactionGraphOperator -> IO (Maybe RelationalError)

299−executeGraphExpr sessionId (InProcessConnection strat sessions graphTvar) graphExpr = do

333+executeGraphExpr sessionId (InProcessConnection strat _ sessions graphTvar) graphExpr = do

300 freshUUID <- nextRandom

334 freshUUID <- nextRandom

301− manip <- atomically $ executeGraphExprSTM_ sessionId sessions graphTvar

335+ manip <- atomically $ executeGraphExprSTM_ freshUUID sessionId sessions graphExpr graphTvar

302 case manip of

336 case manip of

303 Left err -> return $ Just err

337 Left err -> return $ Just err

304 Right newGraph -> do

338 Right newGraph -> do

@@ -308,37 +342,58 @@ executeGraphExpr sessionId (InProcessConnection strat sessions graphTvar) graphE

308 return Nothing

342 return Nothing

309executeGraphExpr sessionId conn@(RemoteProcessConnection _ _) graphExpr = remoteCall conn (ExecuteGraphExpr sessionId graphExpr)

343executeGraphExpr sessionId conn@(RemoteProcessConnection _ _) graphExpr = remoteCall conn (ExecuteGraphExpr sessionId graphExpr)

310

344

311−commitSTM_ :: SessionId -> Sessions -> STM (Maybe RelationalError)

345+commitSTM_ :: UUID -> SessionId -> Sessions -> TVar TransactionGraph -> STM (Maybe RelationalError)

346+commitSTM_ freshUUID sessionId sessions graph = do

312−commitSTM_ sessionId sessions = executeGraphExprSTM_ sessionId sessions graph Commit

347+ mErr <- executeGraphExprSTM_ freshUUID sessionId sessions Commit graph

348+ pure $ case mErr of

349+ Left err -> Just err

350+ Right _ -> Nothing

313

351

314-- | After modifying a session, 'commit' the transaction to the transaction graph at the head which the session is referencing. This will also trigger checks for any notifications which need to be propagated.

352-- | After modifying a session, 'commit' the transaction to the transaction graph at the head which the session is referencing. This will also trigger checks for any notifications which need to be propagated.

315commit :: SessionId -> Connection -> IO (Maybe RelationalError)

353commit :: SessionId -> Connection -> IO (Maybe RelationalError)

316−commit sessionId conn@(InProcessConnection _ sessions graphTvar) = do

354+commit sessionId (InProcessConnection _ clientNodes sessions graphTvar) = do

355+ freshUUID <- nextRandom

317 mNots <- atomically $ do

356 mNots <- atomically $ do

318 graph <- readTVar graphTvar

357 graph <- readTVar graphTvar

319 -- take the notifications of the current context and scan them for changes in the previous -> current contexts

358 -- take the notifications of the current context and scan them for changes in the previous -> current contexts

320 eSession <- sessionForSessionId sessionId sessions

359 eSession <- sessionForSessionId sessionId sessions

321 case eSession of

360 case eSession of

322 Left err -> pure (Left err)

361 Left err -> pure (Left err)

323− Right (Session (DisconnectedTransaction parentUUID currentContext) -> do

362+ Right (Session (DisconnectedTransaction parentUUID currentContext)) -> do

324 --grab previous context

363 --grab previous context

325 case transactionForUUID parentUUID graph of

364 case transactionForUUID parentUUID graph of

326 Left err -> pure $ Left err

365 Left err -> pure $ Left err

327 Right (Transaction _ _ previousContext) -> do

366 Right (Transaction _ _ previousContext) -> do

328 let nots = notifications currentContext

367 let nots = notifications currentContext

329 fireNots = notificationChanges nots previousContext currentContext

368 fireNots = notificationChanges nots previousContext currentContext

369+ evaldNots = M.map mkEvaldNot fireNots

370+ mkEvaldNot notif = EvaluatedNotification { notification = notif, reportRelation = evalState (RE.evalRelationalExpr (reportExpr notif)) currentContext }

330− commitSTM_ sessionId sessions graph

371+ mErr <- commitSTM_ freshUUID sessionId sessions graphTvar

331− pure (Right fireNots)

372+ case mErr of

373+ Just err -> pure (Left err)

374+ Nothing -> do

375+ notifyNodes <- toList (STMSet.stream clientNodes)

376+ pure (Right (evaldNots, notifyNodes))

332 case mNots of

377 case mNots of

333 Left err -> pure (Just err)

378 Left err -> pure (Just err)

334− Right notsToFire -> do

379+ Right (notsToFire, nodesToNotify) -> do

335− --trigger the notifications on all clients

380+ --trigger the notifications on all clients

336−

381+ sendNotifications nodesToNotify notsToFire

382+ pure Nothing

337commit sessionId conn@(RemoteProcessConnection _ _) = remoteCall conn (ExecuteGraphExpr sessionId Commit)

383commit sessionId conn@(RemoteProcessConnection _ _) = remoteCall conn (ExecuteGraphExpr sessionId Commit)

384+

385+sendNotifications :: [ProcessId] -> EvaluatedNotifications -> IO ()

386+sendNotifications pids nots = mapM_ sendNots pids

387+ where

388+ sendNots remoteClientPid = do

389+ eLocalNode <- commonLocalNode

390+ case eLocalNode of

391+ Left err -> error ("Failed to get local node: " ++ show err)

392+ Right localNode -> runProcess localNode $ send remoteClientPid (NotificationMessage nots)

338

393

339-- | Discard any changes made in the current session. This resets the disconnected transaction to reference the original database context of the parent transaction and is a very cheap operation.

394-- | Discard any changes made in the current session. This resets the disconnected transaction to reference the original database context of the parent transaction and is a very cheap operation.

340rollback :: SessionId -> Connection -> IO (Maybe RelationalError)

395rollback :: SessionId -> Connection -> IO (Maybe RelationalError)

341−rollback sessionId conn@(InProcessConnection _ _ _) = executeGraphExpr sessionId conn Rollback

396+rollback sessionId conn@(InProcessConnection _ _ _ _) = executeGraphExpr sessionId conn Rollback

342rollback sessionId conn@(RemoteProcessConnection _ _) = remoteCall conn (ExecuteGraphExpr sessionId Rollback)

397rollback sessionId conn@(RemoteProcessConnection _ _) = remoteCall conn (ExecuteGraphExpr sessionId Rollback)

343

398

344processPersistence :: PersistenceStrategy -> TransactionGraph -> IO ()

399processPersistence :: PersistenceStrategy -> TransactionGraph -> IO ()

@@ -348,11 +403,11 @@ processPersistence (CrashSafePersistence dbdir) graph = transactionGraphPersist

348

403

349-- | Return a relation whose type would match that of the relational expression if it were executed. This is useful for checking types and validating a relational expression's types.

404-- | Return a relation whose type would match that of the relational expression if it were executed. This is useful for checking types and validating a relational expression's types.

350typeForRelationalExpr :: SessionId -> Connection -> RelationalExpr -> IO (Either RelationalError Relation)

405typeForRelationalExpr :: SessionId -> Connection -> RelationalExpr -> IO (Either RelationalError Relation)

351−typeForRelationalExpr sessionId conn@(InProcessConnection _ _ _) relExpr = atomically $ typeForRelationalExprSTM sessionId conn relExpr

406+typeForRelationalExpr sessionId conn@(InProcessConnection _ _ _ _) relExpr = atomically $ typeForRelationalExprSTM sessionId conn relExpr

352typeForRelationalExpr sessionId conn@(RemoteProcessConnection _ _) relExpr = remoteCall conn (ExecuteTypeForRelationalExpr sessionId relExpr)

407typeForRelationalExpr sessionId conn@(RemoteProcessConnection _ _) relExpr = remoteCall conn (ExecuteTypeForRelationalExpr sessionId relExpr)

353

408

354typeForRelationalExprSTM :: SessionId -> Connection -> RelationalExpr -> STM (Either RelationalError Relation)

409typeForRelationalExprSTM :: SessionId -> Connection -> RelationalExpr -> STM (Either RelationalError Relation)

355−typeForRelationalExprSTM sessionId (InProcessConnection _ sessions _) relExpr = do

410+typeForRelationalExprSTM sessionId (InProcessConnection _ _ sessions _) relExpr = do

356 eSession <- sessionForSessionId sessionId sessions

411 eSession <- sessionForSessionId sessionId sessions

357 case eSession of

412 case eSession of

358 Left err -> pure $ Left err

413 Left err -> pure $ Left err

@@ -362,7 +417,7 @@ typeForRelationalExprSTM _ _ _ = error "typeForRelationalExprSTM called on non-l

362

417

363-- | Return a 'Map' of the database's constraints at the context of the session and connection.

418-- | Return a 'Map' of the database's constraints at the context of the session and connection.

364inclusionDependencies :: SessionId -> Connection -> IO (Either RelationalError (M.Map IncDepName InclusionDependency))

419inclusionDependencies :: SessionId -> Connection -> IO (Either RelationalError (M.Map IncDepName InclusionDependency))

365−inclusionDependencies sessionId (InProcessConnection _ sessions _) = do

420+inclusionDependencies sessionId (InProcessConnection _ _ sessions _) = do

366 atomically $ do

421 atomically $ do

367 eSession <- sessionForSessionId sessionId sessions

422 eSession <- sessionForSessionId sessionId sessions

368 case eSession of

423 case eSession of

@@ -373,7 +428,7 @@ inclusionDependencies sessionId conn@(RemoteProcessConnection _ _) = remoteCall

373

428

374-- | Return an optimized database expression which is logically equivalent to the input database expression. This function can be used to determine which expression will actually be evaluated.

429-- | Return an optimized database expression which is logically equivalent to the input database expression. This function can be used to determine which expression will actually be evaluated.

375planForDatabaseContextExpr :: SessionId -> Connection -> DatabaseExpr -> IO (Either RelationalError DatabaseExpr)

430planForDatabaseContextExpr :: SessionId -> Connection -> DatabaseExpr -> IO (Either RelationalError DatabaseExpr)

376−planForDatabaseContextExpr sessionId (InProcessConnection _ sessions _) dbExpr = do

431+planForDatabaseContextExpr sessionId (InProcessConnection _ _ sessions _) dbExpr = do

377 atomically $ do

432 atomically $ do

378 eSession <- sessionForSessionId sessionId sessions

433 eSession <- sessionForSessionId sessionId sessions

379 case eSession of

434 case eSession of

@@ -387,7 +442,7 @@ planForDatabaseContextExpr sessionId conn@(RemoteProcessConnection _ _) dbExpr =

387-- * id- UUID attribute of the transaction

442-- * id- UUID attribute of the transaction

388-- * parents- a relation-valued attribute which contains a relation of UUIDs which are parent transaction to the transaction

443-- * parents- a relation-valued attribute which contains a relation of UUIDs which are parent transaction to the transaction

389transactionGraphAsRelation :: SessionId -> Connection -> IO (Either RelationalError Relation)

444transactionGraphAsRelation :: SessionId -> Connection -> IO (Either RelationalError Relation)

390−transactionGraphAsRelation sessionId (InProcessConnection _ sessions tvar) = do

445+transactionGraphAsRelation sessionId (InProcessConnection _ _ sessions tvar) = do

391 atomically $ do

446 atomically $ do

392 eSession <- sessionForSessionId sessionId sessions

447 eSession <- sessionForSessionId sessionId sessions

393 case eSession of

448 case eSession of

@@ -400,7 +455,7 @@ transactionGraphAsRelation sessionId conn@(RemoteProcessConnection _ _) = remote

400

455

401-- | Returns the UUID for the connection's disconnected transaction committed parent transaction.

456-- | Returns the UUID for the connection's disconnected transaction committed parent transaction.

402headTransactionUUID :: SessionId -> Connection -> IO (Maybe UUID)

457headTransactionUUID :: SessionId -> Connection -> IO (Maybe UUID)

403−headTransactionUUID sessionId (InProcessConnection _ sessions _) = do

458+headTransactionUUID sessionId (InProcessConnection _ _ sessions _) = do

404 atomically $ do

459 atomically $ do

405 eSession <- sessionForSessionId sessionId sessions

460 eSession <- sessionForSessionId sessionId sessions

406 case eSession of

461 case eSession of

@@ -408,8 +463,8 @@ headTransactionUUID sessionId (InProcessConnection _ sessions _) = do

408 Right session -> pure $ Just (sessionParentUUID session)

463 Right session -> pure $ Just (sessionParentUUID session)

409headTransactionUUID sessionId conn@(RemoteProcessConnection _ _) = remoteCall conn (RetrieveHeadTransactionUUID sessionId)

464headTransactionUUID sessionId conn@(RemoteProcessConnection _ _) = remoteCall conn (RetrieveHeadTransactionUUID sessionId)

410

465

411−headNameSTM_ :: SessionId -> Sessions -> TransactionGraph -> STM (Maybe HeadName)

466+headNameSTM_ :: SessionId -> Sessions -> TVar TransactionGraph -> STM (Maybe HeadName)

412−headNameSTM_ sessionId sessions graphTvar= do

467+headNameSTM_ sessionId sessions graphTvar = do

413 graph <- readTVar graphTvar

468 graph <- readTVar graphTvar

414 eSession <- sessionForSessionId sessionId sessions

469 eSession <- sessionForSessionId sessionId sessions

415 case eSession of

470 case eSession of

@@ -420,7 +475,7 @@ headNameSTM_ sessionId sessions graphTvar= do

420

475

421-- | Returns Just the name of the head of the current disconnected transaction or Nothing.

476-- | Returns Just the name of the head of the current disconnected transaction or Nothing.

422headName :: SessionId -> Connection -> IO (Maybe HeadName)

477headName :: SessionId -> Connection -> IO (Maybe HeadName)

423−headName sessionId (InProcessConnection _ sessions graphTvar) = do

478+headName sessionId (InProcessConnection _ _ sessions graphTvar) = do

424 atomically (headNameSTM_ sessionId sessions graphTvar)

479 atomically (headNameSTM_ sessionId sessions graphTvar)

425headName sessionId conn@(RemoteProcessConnection _ _) = remoteCall conn (ExecuteHeadName sessionId)

480headName sessionId conn@(RemoteProcessConnection _ _) = remoteCall conn (ExecuteHeadName sessionId)

426

481

ProjectM36/DatabaseContext.hs+4−9

@@ -6,20 +6,15 @@ import qualified Data.HashSet as HS

6empty :: DatabaseContext

7empty = DatabaseContext { inclusionDependencies = M.empty,

8 relationVariables = M.empty,

9+ notifications = M.empty,

9 atomFunctions = HS.empty }

10 atomFunctions = HS.empty }

11-- | convert an existing database context into its constituent expression.

12-- | convert an existing database context into its constituent expression.

12databaseContextAsDatabaseContextExpr :: DatabaseContext -> DatabaseExpr

13databaseContextAsDatabaseContextExpr :: DatabaseContext -> DatabaseExpr

13−databaseContextAsDatabaseContextExpr (DatabaseContext incDeps relVars funcs) = MultipleExpr $ incDepsExprs ++ relVarsExprs ++ funcsExprs

14+databaseContextAsDatabaseContextExpr context = MultipleExpr $ incDepsExprs ++ relVarsExprs ++ funcsExprs

14 where

15 where

15− relVarsExprs = map (\(name, rel) -> Assign name (ExistingRelation rel)) (M.toList relVars)

16+ relVarsExprs = map (\(name, rel) -> Assign name (ExistingRelation rel)) (M.toList (relationVariables context))

16 incDepsExprs :: [DatabaseExpr]

17 incDepsExprs :: [DatabaseExpr]

17− incDepsExprs = map (\(name, dep) -> AddInclusionDependency name dep) (M.toList incDeps)

18+ incDepsExprs = map (\(name, dep) -> AddInclusionDependency name dep) (M.toList (inclusionDependencies context))

18 funcsExprs = [] -- map (\func -> ) (HS.toList funcs) -- there are no databaseExprs to add atom functions yet

19 funcsExprs = [] -- map (\func -> ) (HS.toList funcs) -- there are no databaseExprs to add atom functions yet

20−-- | Returns the notifications which should be triggered based on the transition from the first 'DatabaseContext' to the second 'DatabaseContext'.

21−notificationChanges :: Notifications -> DatabaseContext -> DatabaseContext -> Notifications

22−notificationChanges notifications context1 context2 = M.filter notificationFilter notifications

23− where

24− notificationFilter (Notification changeExpr _) = evalChangeExpr changeExpr context1 /= evalChangeExpr changeExpr context2

25− evalChangeExpr changeExpr = evalState (evalRelationalExpr changeExpr)

ProjectM36/Notifications.hsnew file+12

@@ -0,0 +1,12 @@

1+module ProjectM36.Notifications where

2+import ProjectM36.Base

3+import ProjectM36.RelationalExpression

4+import Control.Monad.State

5+import qualified Data.Map as M

7+-- | Returns the notifications which should be triggered based on the transition from the first 'DatabaseContext' to the second 'DatabaseContext'.

8+notificationChanges :: Notifications -> DatabaseContext -> DatabaseContext -> Notifications

9+notificationChanges nots context1 context2 = M.filter notificationFilter nots

10+ where

11+ notificationFilter (Notification chExpr _) = evalChangeExpr chExpr context1 /= evalChangeExpr chExpr context2

12+ evalChangeExpr chExpr = evalState (evalRelationalExpr chExpr)

ProjectM36/RelationalExpression.hs+5−2

@@ -116,19 +116,22 @@ evalRelationalExpr (Extend tupleExpression relExpr) = do

116emptyDatabaseContext :: DatabaseContext

117emptyDatabaseContext = DatabaseContext { inclusionDependencies = M.empty,

118 relationVariables = M.empty,

119− atomFunctions = HS.empty

119+ atomFunctions = HS.empty,

120+ notifications = M.empty

120 }

121 }

121

122

122basicDatabaseContext :: DatabaseContext

123basicDatabaseContext :: DatabaseContext

123basicDatabaseContext = DatabaseContext { inclusionDependencies = M.empty,

124basicDatabaseContext = DatabaseContext { inclusionDependencies = M.empty,

124 relationVariables = M.fromList [("true", relationTrue),

125 relationVariables = M.fromList [("true", relationTrue),

125 ("false", relationFalse)],

126 ("false", relationFalse)],

126− atomFunctions = basicAtomFunctions

127+ atomFunctions = basicAtomFunctions,

128+ notifications = M.empty

127 }

129 }

128

130

129dateExamples :: DatabaseContext

131dateExamples :: DatabaseContext

130dateExamples = DatabaseContext { inclusionDependencies = dateIncDeps,

132dateExamples = DatabaseContext { inclusionDependencies = dateIncDeps,

131 relationVariables = M.union (relationVariables basicDatabaseContext) dateRelVars,

133 relationVariables = M.union (relationVariables basicDatabaseContext) dateRelVars,

134+ notifications = M.empty,

132 atomFunctions = basicAtomFunctions}

135 atomFunctions = basicAtomFunctions}

133 where

136 where

134 dateRelVars = M.fromList [("S", suppliers),

137 dateRelVars = M.fromList [("S", suppliers),

ProjectM36/Server/EntryPoints.hs+2−2

@@ -25,8 +25,8 @@ handleExecuteHeadName sessionId conn = do

25 reply ret conn

27handleLogin :: Connection -> ProcessId -> Process (ProcessReply Bool Connection)

28−handleLogin conn@(Connection _ clientNodesTvar _ _) = do

28+handleLogin conn newClientProcessId = do

29−

29+ liftIO (addClientNode newClientProcessId)

30 reply True conn

32handleExecuteGraphExpr :: SessionId -> Connection -> TransactionGraphOperator -> Process (ProcessReply (Maybe RelationalError) Connection)

ProjectM36/Server/RemoteCallTypes.hs+4−1

@@ -6,9 +6,12 @@ import ProjectM36.Session

6import GHC.Generics

7import Data.Binary

8import Data.UUID

9+import Control.Distributed.Process (ProcessId)

11+-- | The initial login message. The argument should be the process id of the initiating client. This ProcessId will receive notification callbacks.

10−data Login = Login

12+data Login = Login ProcessId

11 deriving (Binary, Generic)

13 deriving (Binary, Generic)

14+

12data Logout = Logout

15data Logout = Logout

13 deriving (Binary, Generic)

16 deriving (Binary, Generic)

14data ExecuteRelationalExpr = ExecuteRelationalExpr SessionId RelationalExpr

17data ExecuteRelationalExpr = ExecuteRelationalExpr SessionId RelationalExpr