@@ -1,4 +1,4 @@
1−{-# LANGUAGE OverloadedStrings #-}
1+{-# LANGUAGE OverloadedStrings, DeriveAnyClass, DeriveGeneric #-}
22module ProjectM36.Client
33 (ConnectionInfo(..),
44 Connection(..),
@@ -34,6 +34,7 @@ module ProjectM36.Client
3434 createSessionAtCommit,
3535 createSessionAtHead,
3636 closeSession,
37+ addClientNode,
3738 TransactionGraphOperator(..),
3839 Atomable,
3940 NodeId(..),
@@ -51,18 +52,19 @@ import ProjectM36.TransactionGraph
5152import ProjectM36.TransactionGraph.Persist
5253import ProjectM36.Attribute
5354import ProjectM36.Persist (DiskSync(..))
55+import ProjectM36.Notifications
5456import ProjectM36.Server.RemoteCallTypes
5557import Network.Transport.TCP (createTransport, defaultTCPParameters, encodeEndPointAddress)
5658import Control.Distributed.Process.Node (newLocalNode, initRemoteTable, runProcess, LocalNode)
5759import Control.Distributed.Process.Extras.Internal.Types (whereisRemote)
5860import Control.Distributed.Process.ManagedProcess.Client (call, safeCall)
59−import Control.Distributed.Process (NodeId(..))
61+import Control.Distributed.Process (NodeId(..), getSelfPid)
6062
6163import Data.UUID (UUID)
6264import Data.UUID.V4 (nextRandom)
6365import Control.Concurrent.STM
6466import Data.Word
65−import Control.Distributed.Process (ProcessId, Process)
67+import Control.Distributed.Process (ProcessId, Process, receiveWait, send, match)
6668import Control.Exception (IOException)
6769import Control.Concurrent.MVar
6870import qualified Data.Map as M
@@ -73,6 +75,8 @@ import qualified STMContainers.Set as STMSet
7375import ProjectM36.Session
7476import ProjectM36.Sessions
7577import ListT
78+import Data.Binary (Binary)
79+import GHC.Generics (Generic)
7680
7781type Hostname = String
7882
@@ -80,9 +84,23 @@ type Port = Word16
8084
8185type DatabaseName = String
8286
87+-- | Construct a 'ConnectionInfo' to describe how to make the 'Connection'.
8388data ConnectionInfo = InProcessConnectionInfo PersistenceStrategy |
8489 RemoteProcessConnectionInfo DatabaseName NodeId
8590
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+
86104createNodeId :: Hostname -> Port -> NodeId
87105createNodeId host port = NodeId $ encodeEndPointAddress host (show port) 0
88106
@@ -121,6 +139,21 @@ commonLocalNode = do
121139 case eLocalTransport of
122140 Left err -> pure (Left $ IOExceptionError err)
123141 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
124157
125158-- | To create a 'Connection' to a remote or local database, create a connectionInfo and call 'connectProjectM36'.
126159connectProjectM36 :: ConnectionInfo -> IO (Either ConnectionError Connection)
@@ -135,10 +168,10 @@ connectProjectM36 (InProcessConnectionInfo strat) = do
135168 graphTvar <- newTVarIO freshGraph
136169 clientNodes <- STMSet.newIO
137170 sessions <- STMMap.newIO
171+ notProcId <- startNotificationListener
138172 return $ Right $ InProcessConnection strat clientNodes sessions graphTvar
139173 MinimalPersistence dbdir -> connectPersistentProjectM36 strat NoDiskSync dbdir freshGraph
140174 CrashSafePersistence dbdir -> connectPersistentProjectM36 strat FsyncDiskSync dbdir freshGraph
141−
142175
143176connectProjectM36 (RemoteProcessConnectionInfo databaseName serverNodeId) = do
144177 connStatus <- newEmptyMVar
@@ -150,10 +183,11 @@ connectProjectM36 (RemoteProcessConnectionInfo databaseName serverNodeId) = do
150183 Right localNode -> do
151184 runProcess localNode $ do
152185 mServerProcessId <- whereisRemote serverNodeId dbName
186+ selfPid <- getSelfPid
153187 case mServerProcessId of
154188 Nothing -> liftIO $ putMVar connStatus $ Left (NoSuchDatabaseByNameError databaseName)
155189 Just serverProcessId -> do
156− loginConfirmation <- call serverProcessId (Login loginNode)
190+ loginConfirmation <- call serverProcessId (Login selfPid)
157191 if not loginConfirmation then
158192 liftIO $ putMVar connStatus (Left LoginError)
159193 else do
@@ -177,18 +211,19 @@ connectPersistentProjectM36 strat sync dbdir freshGraph = do
177211 Right graph' -> do
178212 tvarGraph <- newTVarIO graph'
179213 sessions <- STMMap.newIO
180− return $ Right $ InProcessConnection strat sessions tvarGraph
214+ clientNodes <- STMSet.newIO
215+ return $ Right $ InProcessConnection strat clientNodes sessions tvarGraph
181216
182217-- | Create a new session at the transaction UUID and return the session's Id.
183218createSessionAtCommit :: UUID -> Connection -> IO (Either RelationalError SessionId)
184−createSessionAtCommit commitUUID conn@(InProcessConnection _ _ _) = do
219+createSessionAtCommit commitUUID conn@(InProcessConnection _ _ _ _) = do
185220 newSessionId <- nextRandom
186221 atomically $ do
187222 createSessionAtCommit_ commitUUID newSessionId conn
188223createSessionAtCommit uuid conn@(RemoteProcessConnection _ _) = remoteCall conn (CreateSessionAtCommit uuid)
189224
190225createSessionAtCommit_ :: UUID -> SessionId -> Connection -> STM (Either RelationalError SessionId)
191−createSessionAtCommit_ commitUUID newSessionId (InProcessConnection _ sessions graphTvar) = do
226+createSessionAtCommit_ commitUUID newSessionId (InProcessConnection _ _ sessions graphTvar) = do
192227 graph <- readTVar graphTvar
193228 case transactionForUUID commitUUID graph of
194229 Left err -> pure (Left err)
@@ -204,32 +239,31 @@ createSessionAtCommit_ _ _ (RemoteProcessConnection _ _) = error "createSessionA
204239
205240-- | 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.
206241createSessionAtHead :: HeadName -> Connection -> IO (Either RelationalError SessionId)
207−createSessionAtHead headn conn@(InProcessConnection _ _ tvar) = do
242+createSessionAtHead headn conn@(InProcessConnection _ _ _ graphTvar) = do
208243 newSessionId <- nextRandom
209244 atomically $ do
210− graph <- readTVar tvar
245+ graph <- readTVar graphTvar
211246 case transactionForHead headn graph of
212247 Nothing -> pure $ Left (NoSuchHeadNameError headn)
213248 Just trans -> createSessionAtCommit_ (transactionUUID trans) newSessionId conn
214249createSessionAtHead headn conn@(RemoteProcessConnection _ _) = remoteCall conn (CreateSessionAtHead headn)
215250
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.
217252addClientNode :: Connection -> ProcessId -> IO ()
218−addClientNode (RemoteProcessConnection _ _) = error "addClientNode called on remote connection"
253+addClientNode (RemoteProcessConnection _ _) _ = error "addClientNode called on remote connection"
219254addClientNode (InProcessConnection _ clientNodes _ _) newProcessId = atomically (STMSet.insert newProcessId clientNodes)
220255
221−
222256-- | Discards a session, eliminating any uncommitted changes present in the session.
223257closeSession :: SessionId -> Connection -> IO ()
224−closeSession sessionId (InProcessConnection _ sessions _) = do
225− atomically $ do
226− STMMap.delete sessionId sessions
258+closeSession sessionId (InProcessConnection _ _ sessions _) = do
259+ atomically $ STMMap.delete sessionId sessions
227260closeSession sessionId conn@(RemoteProcessConnection _ _) = remoteCall conn (CloseSession sessionId)
228261-- | 'close' cleans up the database access connection. Note that sessions persist even after the connection is closed.
229262close :: Connection -> IO ()
230−close (InProcessConnection _ sessions _) = atomically $ do
263+close (InProcessConnection _ _ sessions _) = atomically $ do
231264 traverse_ (\(k,_) -> STMMap.delete k sessions) (STMMap.stream sessions)
232265 pure ()
266+
233267close (RemoteProcessConnection localNode serverProcessId) = do
234268 runProcessResult localNode $ do
235269 call serverProcessId Logout
@@ -243,7 +277,7 @@ runProcessResult localNode proc = do
243277 takeMVar ret
244278
245279remoteCall :: (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"
247281remoteCall (RemoteProcessConnection localNode serverProcessId) arg = runProcessResult localNode $ do
248282 ret <- safeCall serverProcessId arg
249283 case ret of
@@ -257,7 +291,7 @@ sessionForSessionId sessionId sessions = do
257291
258292-- | Execute a relational expression in the context of the session and connection. Relational expressions are queries and therefore cannot alter the database.
259293executeRelationalExpr :: SessionId -> Connection -> RelationalExpr -> IO (Either RelationalError Relation)
260−executeRelationalExpr sessionId (InProcessConnection _ sessions _) expr = atomically $ do
294+executeRelationalExpr sessionId (InProcessConnection _ _ sessions _) expr = atomically $ do
261295 eSession <- sessionForSessionId sessionId sessions
262296 case eSession of
263297 Left err -> pure $ Left err
@@ -266,7 +300,7 @@ executeRelationalExpr sessionId conn@(RemoteProcessConnection _ _) relExpr = rem
266300
267301-- | 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.
268302executeDatabaseContextExpr :: SessionId -> Connection -> DatabaseExpr -> IO (Maybe RelationalError)
269−executeDatabaseContextExpr sessionId (InProcessConnection _ sessions _) expr = atomically $ do
303+executeDatabaseContextExpr sessionId (InProcessConnection _ _ sessions _) expr = atomically $ do
270304 eSession <- sessionForSessionId sessionId sessions
271305 case eSession of
272306 Left err -> pure $ Just err
@@ -279,9 +313,8 @@ executeDatabaseContextExpr sessionId (InProcessConnection _ sessions _) expr = a
279313 return Nothing
280314executeDatabaseContextExpr sessionId conn@(RemoteProcessConnection _ _) dbExpr = remoteCall conn (ExecuteDatabaseContextExpr sessionId dbExpr)
281315
282−executeGraphExprSTM_ :: SessionId -> Sessions -> TransactionGraphOperator -> TVar TransactionGraph -> STM (Either RelationalError TransactionGraph)
283−executeGraphExprSTM_ sessionId sessions graphExpr graphTvar = do
284− manip <- atomically $ do
316+executeGraphExprSTM_ :: UUID -> SessionId -> Sessions -> TransactionGraphOperator -> TVar TransactionGraph -> STM (Either RelationalError TransactionGraph)
317+executeGraphExprSTM_ freshUUID sessionId sessions graphExpr graphTvar = do
285318 graph <- readTVar graphTvar
286319 eSession <- sessionForSessionId sessionId sessions
287320 case eSession of
@@ -293,12 +326,13 @@ executeGraphExprSTM_ sessionId sessions graphExpr graphTvar = do
293326 let newSession = Session discon'
294327 STMMap.insert newSession sessionId sessions
295328 pure $ Right graph'
329+
296330
297331-- | Execute a transaction graph expression in the context of the session and connection. Transaction graph operators modify the transaction graph state.
298332executeGraphExpr :: SessionId -> Connection -> TransactionGraphOperator -> IO (Maybe RelationalError)
299−executeGraphExpr sessionId (InProcessConnection strat sessions graphTvar) graphExpr = do
333+executeGraphExpr sessionId (InProcessConnection strat _ sessions graphTvar) graphExpr = do
300334 freshUUID <- nextRandom
301− manip <- atomically $ executeGraphExprSTM_ sessionId sessions graphTvar
335+ manip <- atomically $ executeGraphExprSTM_ freshUUID sessionId sessions graphExpr graphTvar
302336 case manip of
303337 Left err -> return $ Just err
304338 Right newGraph -> do
@@ -308,37 +342,58 @@ executeGraphExpr sessionId (InProcessConnection strat sessions graphTvar) graphE
308342 return Nothing
309343executeGraphExpr sessionId conn@(RemoteProcessConnection _ _) graphExpr = remoteCall conn (ExecuteGraphExpr sessionId graphExpr)
310344
311−commitSTM_ :: SessionId -> Sessions -> STM (Maybe RelationalError)
312−commitSTM_ sessionId sessions = executeGraphExprSTM_ sessionId sessions graph Commit
345+commitSTM_ :: UUID -> SessionId -> Sessions -> TVar TransactionGraph -> STM (Maybe RelationalError)
346+commitSTM_ freshUUID sessionId sessions graph = do
347+ mErr <- executeGraphExprSTM_ freshUUID sessionId sessions Commit graph
348+ pure $ case mErr of
349+ Left err -> Just err
350+ Right _ -> Nothing
313351
314352-- | 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.
315353commit :: SessionId -> Connection -> IO (Maybe RelationalError)
316−commit sessionId conn@(InProcessConnection _ sessions graphTvar) = do
354+commit sessionId (InProcessConnection _ clientNodes sessions graphTvar) = do
355+ freshUUID <- nextRandom
317356 mNots <- atomically $ do
318357 graph <- readTVar graphTvar
319358 -- take the notifications of the current context and scan them for changes in the previous -> current contexts
320359 eSession <- sessionForSessionId sessionId sessions
321360 case eSession of
322361 Left err -> pure (Left err)
323− Right (Session (DisconnectedTransaction parentUUID currentContext) -> do
362+ Right (Session (DisconnectedTransaction parentUUID currentContext)) -> do
324363 --grab previous context
325364 case transactionForUUID parentUUID graph of
326365 Left err -> pure $ Left err
327366 Right (Transaction _ _ previousContext) -> do
328367 let nots = notifications currentContext
329368 fireNots = notificationChanges nots previousContext currentContext
330− commitSTM_ sessionId sessions graph
331− pure (Right fireNots)
369+ evaldNots = M.map mkEvaldNot fireNots
370+ mkEvaldNot notif = EvaluatedNotification { notification = notif, reportRelation = evalState (RE.evalRelationalExpr (reportExpr notif)) currentContext }
371+ mErr <- commitSTM_ freshUUID sessionId sessions graphTvar
372+ case mErr of
373+ Just err -> pure (Left err)
374+ Nothing -> do
375+ notifyNodes <- toList (STMSet.stream clientNodes)
376+ pure (Right (evaldNots, notifyNodes))
332377 case mNots of
333378 Left err -> pure (Just err)
334− Right notsToFire -> do
335− --trigger the notifications on all clients
336−
379+ Right (notsToFire, nodesToNotify) -> do
380+ --trigger the notifications on all clients
381+ sendNotifications nodesToNotify notsToFire
382+ pure Nothing
337383commit 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)
338393
339394-- | 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.
340395rollback :: SessionId -> Connection -> IO (Maybe RelationalError)
341−rollback sessionId conn@(InProcessConnection _ _ _) = executeGraphExpr sessionId conn Rollback
396+rollback sessionId conn@(InProcessConnection _ _ _ _) = executeGraphExpr sessionId conn Rollback
342397rollback sessionId conn@(RemoteProcessConnection _ _) = remoteCall conn (ExecuteGraphExpr sessionId Rollback)
343398
344399processPersistence :: PersistenceStrategy -> TransactionGraph -> IO ()
@@ -348,11 +403,11 @@ processPersistence (CrashSafePersistence dbdir) graph = transactionGraphPersist
348403
349404-- | 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.
350405typeForRelationalExpr :: 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
352407typeForRelationalExpr sessionId conn@(RemoteProcessConnection _ _) relExpr = remoteCall conn (ExecuteTypeForRelationalExpr sessionId relExpr)
353408
354409typeForRelationalExprSTM :: SessionId -> Connection -> RelationalExpr -> STM (Either RelationalError Relation)
355−typeForRelationalExprSTM sessionId (InProcessConnection _ sessions _) relExpr = do
410+typeForRelationalExprSTM sessionId (InProcessConnection _ _ sessions _) relExpr = do
356411 eSession <- sessionForSessionId sessionId sessions
357412 case eSession of
358413 Left err -> pure $ Left err
@@ -362,7 +417,7 @@ typeForRelationalExprSTM _ _ _ = error "typeForRelationalExprSTM called on non-l
362417
363418-- | Return a 'Map' of the database's constraints at the context of the session and connection.
364419inclusionDependencies :: SessionId -> Connection -> IO (Either RelationalError (M.Map IncDepName InclusionDependency))
365−inclusionDependencies sessionId (InProcessConnection _ sessions _) = do
420+inclusionDependencies sessionId (InProcessConnection _ _ sessions _) = do
366421 atomically $ do
367422 eSession <- sessionForSessionId sessionId sessions
368423 case eSession of
@@ -373,7 +428,7 @@ inclusionDependencies sessionId conn@(RemoteProcessConnection _ _) = remoteCall
373428
374429-- | 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.
375430planForDatabaseContextExpr :: SessionId -> Connection -> DatabaseExpr -> IO (Either RelationalError DatabaseExpr)
376−planForDatabaseContextExpr sessionId (InProcessConnection _ sessions _) dbExpr = do
431+planForDatabaseContextExpr sessionId (InProcessConnection _ _ sessions _) dbExpr = do
377432 atomically $ do
378433 eSession <- sessionForSessionId sessionId sessions
379434 case eSession of
@@ -387,7 +442,7 @@ planForDatabaseContextExpr sessionId conn@(RemoteProcessConnection _ _) dbExpr =
387442-- * id- UUID attribute of the transaction
388443-- * parents- a relation-valued attribute which contains a relation of UUIDs which are parent transaction to the transaction
389444transactionGraphAsRelation :: SessionId -> Connection -> IO (Either RelationalError Relation)
390−transactionGraphAsRelation sessionId (InProcessConnection _ sessions tvar) = do
445+transactionGraphAsRelation sessionId (InProcessConnection _ _ sessions tvar) = do
391446 atomically $ do
392447 eSession <- sessionForSessionId sessionId sessions
393448 case eSession of
@@ -400,7 +455,7 @@ transactionGraphAsRelation sessionId conn@(RemoteProcessConnection _ _) = remote
400455
401456-- | Returns the UUID for the connection's disconnected transaction committed parent transaction.
402457headTransactionUUID :: SessionId -> Connection -> IO (Maybe UUID)
403−headTransactionUUID sessionId (InProcessConnection _ sessions _) = do
458+headTransactionUUID sessionId (InProcessConnection _ _ sessions _) = do
404459 atomically $ do
405460 eSession <- sessionForSessionId sessionId sessions
406461 case eSession of
@@ -408,8 +463,8 @@ headTransactionUUID sessionId (InProcessConnection _ sessions _) = do
408463 Right session -> pure $ Just (sessionParentUUID session)
409464headTransactionUUID sessionId conn@(RemoteProcessConnection _ _) = remoteCall conn (RetrieveHeadTransactionUUID sessionId)
410465
411−headNameSTM_ :: SessionId -> Sessions -> TransactionGraph -> STM (Maybe HeadName)
412−headNameSTM_ sessionId sessions graphTvar= do
466+headNameSTM_ :: SessionId -> Sessions -> TVar TransactionGraph -> STM (Maybe HeadName)
467+headNameSTM_ sessionId sessions graphTvar = do
413468 graph <- readTVar graphTvar
414469 eSession <- sessionForSessionId sessionId sessions
415470 case eSession of
@@ -420,7 +475,7 @@ headNameSTM_ sessionId sessions graphTvar= do
420475
421476-- | Returns Just the name of the head of the current disconnected transaction or Nothing.
422477headName :: SessionId -> Connection -> IO (Maybe HeadName)
423−headName sessionId (InProcessConnection _ sessions graphTvar) = do
478+headName sessionId (InProcessConnection _ _ sessions graphTvar) = do
424479 atomically (headNameSTM_ sessionId sessions graphTvar)
425480headName sessionId conn@(RemoteProcessConnection _ _) = remoteCall conn (ExecuteHeadName sessionId)
426481