@@ -55,7 +55,7 @@ import ProjectM36.Persist (DiskSync(..))
5555import ProjectM36.Notifications
5656import ProjectM36.Server.RemoteCallTypes
5757import Network.Transport.TCP (createTransport, defaultTCPParameters, encodeEndPointAddress)
58−import Control.Distributed.Process.Node (newLocalNode, initRemoteTable, runProcess, LocalNode)
58+import Control.Distributed.Process.Node (newLocalNode, initRemoteTable, runProcess, LocalNode, forkProcess)
5959import Control.Distributed.Process.Extras.Internal.Types (whereisRemote)
6060import Control.Distributed.Process.ManagedProcess.Client (call, safeCall)
6161import Control.Distributed.Process (NodeId(..), getSelfPid)
@@ -64,7 +64,7 @@ import Data.UUID (UUID)
6464import Data.UUID.V4 (nextRandom)
6565import Control.Concurrent.STM
6666import Data.Word
67−import Control.Distributed.Process (ProcessId, Process, receiveWait, send, match)
67+import Control.Distributed.Process (ProcessId, Process, receiveWait, send, match, say)
6868import Control.Exception (IOException)
6969import Control.Concurrent.MVar
7070import qualified Data.Map as M
@@ -77,6 +77,7 @@ import ProjectM36.Sessions
7777import ListT
7878import Data.Binary (Binary)
7979import GHC.Generics (Generic)
80+--import Debug.Trace
8081
8182type Hostname = String
8283
@@ -142,17 +143,20 @@ commonLocalNode = do
142143
143144notificationListener :: Process ()
144145notificationListener = do
145− _ <- receiveWait [
146− match (\(NotificationMessage eNots) -> undefined)
147− ]
148− liftIO $ putStrLn $ "LISTENER THREAD EXIT"
146+ pid <- getSelfPid
147+ liftIO $ putStrLn $ "LISTENER THREAD START " ++ show pid
148+ _ <- forever $ do
149+ receiveWait [
150+ match (\(NotificationMessage eNots) -> say $ "NOTIFICATION: " ++ show eNots)
151+ ]
152+ say "LISTENER THREAD EXIT"
149153 pure ()
150154
151155startNotificationListener :: IO (ProcessId)
152156startNotificationListener = do
153157 eLocalNode <- commonLocalNode
154158 case eLocalNode of
155− Left err -> error "Failed to start local notification listener"
159+ Left err -> error ("Failed to start local notification listener: " ++ show err)
156160 Right localNode -> forkProcess localNode notificationListener
157161
158162-- | To create a 'Connection' to a remote or local database, create a connectionInfo and call 'connectProjectM36'.
@@ -168,7 +172,7 @@ connectProjectM36 (InProcessConnectionInfo strat) = do
168172 graphTvar <- newTVarIO freshGraph
169173 clientNodes <- STMSet.newIO
170174 sessions <- STMMap.newIO
171− notProcId <- startNotificationListener
175+ --notificationPid <- startNotificationListener
172176 return $ Right $ InProcessConnection strat clientNodes sessions graphTvar
173177 MinimalPersistence dbdir -> connectPersistentProjectM36 strat NoDiskSync dbdir freshGraph
174178 CrashSafePersistence dbdir -> connectPersistentProjectM36 strat FsyncDiskSync dbdir freshGraph
@@ -176,6 +180,7 @@ connectProjectM36 (InProcessConnectionInfo strat) = do
176180connectProjectM36 (RemoteProcessConnectionInfo databaseName serverNodeId) = do
177181 connStatus <- newEmptyMVar
178182 eLocalNode <- commonLocalNode
183+ notificationListenerPid <- startNotificationListener
179184 let dbName = remoteDBLookupName databaseName
180185 putStrLn $ show serverNodeId ++ " " ++ dbName
181186 case eLocalNode of
@@ -183,11 +188,10 @@ connectProjectM36 (RemoteProcessConnectionInfo databaseName serverNodeId) = do
183188 Right localNode -> do
184189 runProcess localNode $ do
185190 mServerProcessId <- whereisRemote serverNodeId dbName
186− selfPid <- getSelfPid
187191 case mServerProcessId of
188192 Nothing -> liftIO $ putMVar connStatus $ Left (NoSuchDatabaseByNameError databaseName)
189193 Just serverProcessId -> do
190− loginConfirmation <- call serverProcessId (Login selfPid)
194+ loginConfirmation <- call serverProcessId (Login notificationListenerPid)
191195 if not loginConfirmation then
192196 liftIO $ putMVar connStatus (Left LoginError)
193197 else do
@@ -315,9 +319,9 @@ executeDatabaseContextExpr sessionId conn@(RemoteProcessConnection _ _) dbExpr =
315319
316320executeGraphExprSTM_ :: UUID -> SessionId -> Sessions -> TransactionGraphOperator -> TVar TransactionGraph -> STM (Either RelationalError TransactionGraph)
317321executeGraphExprSTM_ freshUUID sessionId sessions graphExpr graphTvar = do
318− graph <- readTVar graphTvar
319− eSession <- sessionForSessionId sessionId sessions
320− case eSession of
322+ graph <- readTVar graphTvar
323+ eSession <- sessionForSessionId sessionId sessions
324+ case eSession of
321325 Left err -> pure $ Left err
322326 Right (Session discon) -> case evalGraphOp freshUUID discon graph graphExpr of
323327 Left err -> pure $ Left err
@@ -327,61 +331,63 @@ executeGraphExprSTM_ freshUUID sessionId sessions graphExpr graphTvar = do
327331 STMMap.insert newSession sessionId sessions
328332 pure $ Right graph'
329333
330−
334+-- process notifications for commits
335+executeCommitExprSTM_ :: DatabaseContext -> DatabaseContext -> ClientNodes -> STM (EvaluatedNotifications, ClientNodes)
336+executeCommitExprSTM_ oldContext newContext nodes = do
337+ let nots = notifications oldContext
338+ fireNots = notificationChanges nots oldContext newContext
339+ evaldNots = M.map mkEvaldNot fireNots
340+ mkEvaldNot notif = EvaluatedNotification { notification = notif, reportRelation = evalState (RE.evalRelationalExpr (reportExpr notif)) oldContext }
341+ pure (evaldNots, nodes)
342+
343+
331344-- | Execute a transaction graph expression in the context of the session and connection. Transaction graph operators modify the transaction graph state.
332345executeGraphExpr :: SessionId -> Connection -> TransactionGraphOperator -> IO (Maybe RelationalError)
333−executeGraphExpr sessionId (InProcessConnection strat _ sessions graphTvar) graphExpr = do
346+executeGraphExpr sessionId (InProcessConnection strat clientNodes sessions graphTvar) graphExpr = do
334347 freshUUID <- nextRandom
335− manip <- atomically $ executeGraphExprSTM_ freshUUID sessionId sessions graphExpr graphTvar
348+ manip <- atomically $ do
349+ eSession <- sessionForSessionId sessionId sessions
350+ oldGraph <- readTVar graphTvar
351+ case eSession of
352+ Left err -> pure (Left err)
353+ Right (Session (DisconnectedTransaction parentUUID currentContext)) -> do
354+ eGraph <- executeGraphExprSTM_ freshUUID sessionId sessions graphExpr graphTvar
355+ case eGraph of
356+ Left err -> pure (Left err)
357+ Right newGraph -> do
358+ if graphExpr == Commit then
359+ case transactionForUUID parentUUID oldGraph of
360+ Left err -> pure $ Left err
361+ Right (Transaction _ _ previousContext) -> do
362+ (evaldNots, nodes) <- executeCommitExprSTM_ previousContext currentContext clientNodes
363+ nodesToNotify <- toList (STMSet.stream nodes)
364+ pure $ Right (evaldNots, nodesToNotify, newGraph)
365+ else
366+ pure $ Right (M.empty, [], newGraph)
336367 case manip of
337368 Left err -> return $ Just err
338− Right newGraph -> do
369+ Right (notsToFire, nodesToNotify, newGraph) -> do
339370 --update filesystem database, if necessary
340371 --this should really grab a lock at the beginning of the method to be threadsafe
341372 processPersistence strat newGraph
373+ sendNotifications nodesToNotify notsToFire
342374 return Nothing
343375executeGraphExpr sessionId conn@(RemoteProcessConnection _ _) graphExpr = remoteCall conn (ExecuteGraphExpr sessionId graphExpr)
344376
377+{-
345378commitSTM_ :: UUID -> SessionId -> Sessions -> TVar TransactionGraph -> STM (Maybe RelationalError)
346379commitSTM_ freshUUID sessionId sessions graph = do
347380 mErr <- executeGraphExprSTM_ freshUUID sessionId sessions Commit graph
348381 pure $ case mErr of
349382 Left err -> Just err
350383 Right _ -> Nothing
384+-}
351385
352386-- | 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.
353387commit :: SessionId -> Connection -> IO (Maybe RelationalError)
354−commit sessionId (InProcessConnection _ clientNodes sessions graphTvar) = do
355− freshUUID <- nextRandom
356− mNots <- atomically $ do
357− graph <- readTVar graphTvar
358− -- take the notifications of the current context and scan them for changes in the previous -> current contexts
359− eSession <- sessionForSessionId sessionId sessions
360− case eSession of
361− Left err -> pure (Left err)
362− Right (Session (DisconnectedTransaction parentUUID currentContext)) -> do
363− --grab previous context
364− case transactionForUUID parentUUID graph of
365− Left err -> pure $ Left err
366− Right (Transaction _ _ previousContext) -> do
367− let nots = notifications 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 }
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))
377− case mNots of
378− Left err -> pure (Just err)
379− Right (notsToFire, nodesToNotify) -> do
380− --trigger the notifications on all clients
381− sendNotifications nodesToNotify notsToFire
382− pure Nothing
388+commit sessionId conn@(InProcessConnection _ _ _ _) = executeGraphExpr sessionId conn Commit
383389commit sessionId conn@(RemoteProcessConnection _ _) = remoteCall conn (ExecuteGraphExpr sessionId Commit)
384−
390+
385391sendNotifications :: [ProcessId] -> EvaluatedNotifications -> IO ()
386392sendNotifications pids nots = mapM_ sendNots pids
387393 where
@@ -389,7 +395,7 @@ sendNotifications pids nots = mapM_ sendNots pids
389395 eLocalNode <- commonLocalNode
390396 case eLocalNode of
391397 Left err -> error ("Failed to get local node: " ++ show err)
392− Right localNode -> runProcess localNode $ send remoteClientPid (NotificationMessage nots)
398+ Right localNode -> when (not (M.null nots)) $ runProcess localNode $ send remoteClientPid (NotificationMessage nots)
393399
394400-- | 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.
395401rollback :: SessionId -> Connection -> IO (Maybe RelationalError)