@@ -8,6 +8,7 @@ module ProjectM36.Client
8 ConnectionError,
8 ConnectionError,
9 connectProjectM36,
9 connectProjectM36,
11 executeRelationalExpr,
12 executeRelationalExpr,
12 executeDatabaseContextExpr,
13 executeDatabaseContextExpr,
13 executeDatabaseContextIOExpr,
14 executeDatabaseContextIOExpr,
@@ -47,6 +48,8 @@ module ProjectM36.Client
47 callTestTimeout_,
48 callTestTimeout_,
48 RelationCardinality(..),
49 RelationCardinality(..),
49 TransactionGraphOperator(..),
50 TransactionGraphOperator(..),
52+ disconnectedTransaction_,
50 TransGraphRelationalExpr,
53 TransGraphRelationalExpr,
51 TransactionIdLookup(..),
54 TransactionIdLookup(..),
52 TransactionIdHeadBacktrack(..),
55 TransactionIdHeadBacktrack(..),
@@ -83,11 +86,12 @@ import qualified ProjectM36.DisconnectedTransaction as Discon
83import ProjectM36.Relation (typesAsRelation)
86import ProjectM36.Relation (typesAsRelation)
84import ProjectM36.AtomFunctionBody (initScriptSession, ScriptSession)
87import ProjectM36.AtomFunctionBody (initScriptSession, ScriptSession)
85import qualified ProjectM36.Relation as R
88import qualified ProjectM36.Relation as R
89+import Network.Transport (Transport(closeTransport))
86import Network.Transport.TCP (createTransport, defaultTCPParameters, encodeEndPointAddress)
90import Network.Transport.TCP (createTransport, defaultTCPParameters, encodeEndPointAddress)
87−import Control.Distributed.Process.Node (newLocalNode, initRemoteTable, runProcess, LocalNode, forkProcess)
91+import Control.Distributed.Process.Node (newLocalNode, initRemoteTable, runProcess, LocalNode, forkProcess, closeLocalNode)
88import Control.Distributed.Process.Extras.Internal.Types (whereisRemote)
92import Control.Distributed.Process.Extras.Internal.Types (whereisRemote)
89import Control.Distributed.Process.ManagedProcess.Client (call, safeCall)
93import Control.Distributed.Process.ManagedProcess.Client (call, safeCall)
90−import Control.Distributed.Process (NodeId(..))
94+import Control.Distributed.Process (NodeId(..), reconnect)
92import Data.UUID.V4 (nextRandom)
96import Data.UUID.V4 (nextRandom)
93import Control.Concurrent.STM
97import Control.Concurrent.STM
@@ -108,8 +112,6 @@ import GHC.Generics (Generic)
108import Control.DeepSeq (force)
112import Control.DeepSeq (force)
109import System.IO
113import System.IO
113type Hostname = String
115type Hostname = String
115type Port = Word16
117type Port = Word16
@@ -165,8 +167,25 @@ defaultRemoteConnectionInfo = RemoteProcessConnectionInfo defaultDatabaseName (c
165-- | The 'Connection' represents either local or remote access to a database. All operations flow through the connection.
167-- | The 'Connection' represents either local or remote access to a database. All operations flow through the connection.
166type ClientNodes = STMSet.Set ProcessId
168type ClientNodes = STMSet.Set ProcessId
168−data Connection = InProcessConnection PersistenceStrategy ClientNodes Sessions (TVar TransactionGraph) (Maybe ScriptSession) |
170+-- internal structure specific to in-process connections
171+data InProcessConnectionConf = InProcessConnectionConf {
172+ ipPersistenceStrategy :: PersistenceStrategy,
173+ ipClientNodes :: ClientNodes,
174+ ipSessions :: Sessions,
175+ ipTransactionGraph :: TVar TransactionGraph,
176+ ipScriptSession :: Maybe ScriptSession,
177+ ipLocalNode :: LocalNode,
178+ ipTransport :: Transport -- we hold onto this so that we can close it gracefully
181+data RemoteProcessConnectionConf = RemoteProcessConnectionConf {
182+ rLocalNode :: LocalNode,
183+ rProcessId :: ProcessId, --remote processId
184+ rTransport :: Transport-- the TCP socket transport
187+data Connection = InProcessConnection InProcessConnectionConf |
169− RemoteProcessConnection LocalNode ProcessId
188+ RemoteProcessConnection RemoteProcessConnectionConf
171-- | There are several reasons why a connection can fail.
190-- | There are several reasons why a connection can fail.
172data ConnectionError = SetupDatabaseDirectoryError PersistenceError |
191data ConnectionError = SetupDatabaseDirectoryError PersistenceError |
@@ -178,12 +197,14 @@ data ConnectionError = SetupDatabaseDirectoryError PersistenceError |
178remoteDBLookupName :: DatabaseName -> String
197remoteDBLookupName :: DatabaseName -> String
179remoteDBLookupName = (++) "db-"
198remoteDBLookupName = (++) "db-"
181−commonLocalNode :: IO (Either ConnectionError LocalNode)
200+createLocalNode :: IO (LocalNode, Transport)
182−commonLocalNode = do
201+createLocalNode = do
183 eLocalTransport <- createTransport "127.0.0.1" "0" defaultTCPParameters
202 eLocalTransport <- createTransport "127.0.0.1" "0" defaultTCPParameters
184 case eLocalTransport of
203 case eLocalTransport of
185− Left err -> pure (Left $ IOExceptionError err)
204+ Left err -> error ("failed to create transport: " ++ show err)
205+ Right localTransport -> do
186− Right localTransport -> newLocalNode localTransport initRemoteTable >>= pure . Right
206+ localNode <- newLocalNode localTransport initRemoteTable
207+ pure (localNode, localTransport)
188notificationListener :: NotificationCallback -> Process ()
209notificationListener :: NotificationCallback -> Process ()
189notificationListener callback = do
210notificationListener callback = do
@@ -200,12 +221,8 @@ notificationListener callback = do
200 --say "LISTENER THREAD EXIT"
221 --say "LISTENER THREAD EXIT"
203−startNotificationListener :: NotificationCallback -> IO (ProcessId)
224+startNotificationListener :: LocalNode -> NotificationCallback -> IO (ProcessId)
204−startNotificationListener callback = do
205− eLocalNode <- commonLocalNode
207− Left err -> error ("Failed to start local notification listener: " ++ show err)
208− Right localNode -> forkProcess localNode (notificationListener callback)
225+startNotificationListener localNode callback = forkProcess localNode (notificationListener callback)
210createScriptSession :: [String] -> IO (Maybe ScriptSession)
227createScriptSession :: [String] -> IO (Maybe ScriptSession)
211createScriptSession ghcPkgPaths = do
228createScriptSession ghcPkgPaths = do
@@ -227,9 +244,17 @@ connectProjectM36 (InProcessConnectionInfo strat notificationCallback ghcPkgPath
227 graphTvar <- newTVarIO freshGraph
244 graphTvar <- newTVarIO freshGraph
228 clientNodes <- STMSet.newIO
245 clientNodes <- STMSet.newIO
229 sessions <- STMMap.newIO
246 sessions <- STMMap.newIO
247+ (localNode, transport) <- createLocalNode
230− notificationPid <- startNotificationListener notificationCallback
248+ notificationPid <- startNotificationListener localNode notificationCallback
231 mScriptSession <- createScriptSession ghcPkgPaths
249 mScriptSession <- createScriptSession ghcPkgPaths
232− let conn = InProcessConnection strat clientNodes sessions graphTvar mScriptSession
250+ let conn = InProcessConnection (InProcessConnectionConf {
251+ ipPersistenceStrategy =strat,
252+ ipClientNodes = clientNodes,
253+ ipSessions = sessions,
254+ ipTransactionGraph = graphTvar,
255+ ipScriptSession = mScriptSession,
256+ ipLocalNode = localNode,
257+ ipTransport = transport })
233 addClientNode conn notificationPid
258 addClientNode conn notificationPid
234 pure (Right conn)
259 pure (Right conn)
235 MinimalPersistence dbdir -> connectPersistentProjectM36 strat NoDiskSync dbdir freshGraph notificationCallback ghcPkgPaths
260 MinimalPersistence dbdir -> connectPersistentProjectM36 strat NoDiskSync dbdir freshGraph notificationCallback ghcPkgPaths
@@ -237,25 +262,23 @@ connectProjectM36 (InProcessConnectionInfo strat notificationCallback ghcPkgPath
238connectProjectM36 (RemoteProcessConnectionInfo databaseName serverNodeId notificationCallback) = do
263connectProjectM36 (RemoteProcessConnectionInfo databaseName serverNodeId notificationCallback) = do
239 connStatus <- newEmptyMVar
264 connStatus <- newEmptyMVar
240− eLocalNode <- commonLocalNode
265+ (localNode, transport) <- createLocalNode
241− notificationListenerPid <- startNotificationListener notificationCallback
242 let dbName = remoteDBLookupName databaseName
266 let dbName = remoteDBLookupName databaseName
243− putStrLn $ "Connecting to " ++ show serverNodeId ++ " " ++ dbName
267+ --putStrLn $ "Connecting to " ++ show serverNodeId ++ " " ++ dbName
244− case eLocalNode of
268+ notificationListenerPid <- startNotificationListener localNode notificationCallback
245− Left err -> pure (Left err)
246− Right localNode -> do
247− runProcess localNode $ do
269+ runProcess localNode $ do
270+ --even a failed connection retains an open socket!
248− mServerProcessId <- whereisRemote serverNodeId dbName
271+ mServerProcessId <- whereisRemote serverNodeId dbName
249− case mServerProcessId of
272+ case mServerProcessId of
250− Nothing -> liftIO $ putMVar connStatus $ Left (NoSuchDatabaseByNameError databaseName)
273+ Nothing -> liftIO $ putMVar connStatus $ Left (NoSuchDatabaseByNameError databaseName)
251− Just serverProcessId -> do
274+ Just serverProcessId -> do
252− loginConfirmation <- safeLogin (Login notificationListenerPid) serverProcessId
275+ loginConfirmation <- safeLogin (Login notificationListenerPid) serverProcessId
253− if not loginConfirmation then
276+ if not loginConfirmation then
254− liftIO $ putMVar connStatus (Left LoginError)
277+ liftIO $ putMVar connStatus (Left LoginError)
256− liftIO $ putMVar connStatus (Right $ RemoteProcessConnection localNode serverProcessId)
279+ liftIO $ putMVar connStatus (Right $ RemoteProcessConnection (RemoteProcessConnectionConf {rLocalNode = localNode, rProcessId = serverProcessId, rTransport = transport}))
257− status <- takeMVar connStatus
280+ status <- takeMVar connStatus
258− pure status
281+ pure status
260connectPersistentProjectM36 :: PersistenceStrategy ->
283connectPersistentProjectM36 :: PersistenceStrategy ->
261 DiskSync ->
284 DiskSync ->
@@ -277,21 +300,33 @@ connectPersistentProjectM36 strat sync dbdir freshGraph notificationCallback ghc
277 sessions <- STMMap.newIO
300 sessions <- STMMap.newIO
278 clientNodes <- STMSet.newIO
301 clientNodes <- STMSet.newIO
279 mScriptSession <- createScriptSession ghcPkgPaths
302 mScriptSession <- createScriptSession ghcPkgPaths
303+ (localNode, transport) <- createLocalNode
280− let conn = InProcessConnection strat clientNodes sessions tvarGraph mScriptSession
304+ let conn = InProcessConnection (InProcessConnectionConf {
305+ ipPersistenceStrategy = strat,
306+ ipClientNodes = clientNodes,
307+ ipSessions = sessions,
308+ ipTransactionGraph = tvarGraph,
309+ ipScriptSession = mScriptSession,
310+ ipLocalNode = localNode,
311+ ipTransport = transport
281− notificationPid <- startNotificationListener notificationCallback
314+ notificationPid <- startNotificationListener localNode notificationCallback
282 addClientNode conn notificationPid
315 addClientNode conn notificationPid
283 pure (Right conn)
316 pure (Right conn)
285-- | Create a new session at the transaction id and return the session's Id.
318-- | Create a new session at the transaction id and return the session's Id.
286createSessionAtCommit :: TransactionId -> Connection -> IO (Either RelationalError SessionId)
319createSessionAtCommit :: TransactionId -> Connection -> IO (Either RelationalError SessionId)
287−createSessionAtCommit commitId conn@(InProcessConnection _ _ _ _ _) = do
320+createSessionAtCommit commitId conn@(InProcessConnection _) = do
288 newSessionId <- nextRandom
321 newSessionId <- nextRandom
289 atomically $ do
322 atomically $ do
290 createSessionAtCommit_ commitId newSessionId conn
323 createSessionAtCommit_ commitId newSessionId conn
291−createSessionAtCommit uuid conn@(RemoteProcessConnection _ _) = remoteCall conn (CreateSessionAtCommit uuid)
324+createSessionAtCommit uuid conn@(RemoteProcessConnection _) = remoteCall conn (CreateSessionAtCommit uuid)
293createSessionAtCommit_ :: TransactionId -> SessionId -> Connection -> STM (Either RelationalError SessionId)
326createSessionAtCommit_ :: TransactionId -> SessionId -> Connection -> STM (Either RelationalError SessionId)
294−createSessionAtCommit_ commitId newSessionId (InProcessConnection _ _ sessions graphTvar _) = do
327+createSessionAtCommit_ commitId newSessionId (InProcessConnection conf) = do
328+ let sessions = ipSessions conf
329+ graphTvar = ipTransactionGraph conf
295 graph <- readTVar graphTvar
330 graph <- readTVar graphTvar
296 case transactionForId commitId graph of
331 case transactionForId commitId graph of
297 Left err -> pure (Left err)
332 Left err -> pure (Left err)
@@ -303,39 +338,53 @@ createSessionAtCommit_ commitId newSessionId (InProcessConnection _ _ sessions g
303 Nothing -> do
338 Nothing -> do
304 STMMap.insert (Session freshDiscon defaultSchemaName) newSessionId sessions
339 STMMap.insert (Session freshDiscon defaultSchemaName) newSessionId sessions
305 pure $ Right newSessionId
340 pure $ Right newSessionId
306−createSessionAtCommit_ _ _ (RemoteProcessConnection _ _) = error "createSessionAtCommit_ called on remote connection"
341+createSessionAtCommit_ _ _ (RemoteProcessConnection _) = error "createSessionAtCommit_ called on remote connection"
308-- | 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.
343-- | 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.
309createSessionAtHead :: HeadName -> Connection -> IO (Either RelationalError SessionId)
344createSessionAtHead :: HeadName -> Connection -> IO (Either RelationalError SessionId)
310−createSessionAtHead headn conn@(InProcessConnection _ _ _ graphTvar _) = do
345+createSessionAtHead headn conn@(InProcessConnection conf) = do
346+ let graphTvar = ipTransactionGraph conf
311 newSessionId <- nextRandom
347 newSessionId <- nextRandom
312 atomically $ do
348 atomically $ do
313 graph <- readTVar graphTvar
349 graph <- readTVar graphTvar
314 case transactionForHead headn graph of
350 case transactionForHead headn graph of
315 Nothing -> pure $ Left (NoSuchHeadNameError headn)
351 Nothing -> pure $ Left (NoSuchHeadNameError headn)
316 Just trans -> createSessionAtCommit_ (transactionId trans) newSessionId conn
352 Just trans -> createSessionAtCommit_ (transactionId trans) newSessionId conn
317−createSessionAtHead headn conn@(RemoteProcessConnection _ _) = remoteCall conn (CreateSessionAtHead headn)
353+createSessionAtHead headn conn@(RemoteProcessConnection _) = remoteCall conn (CreateSessionAtHead headn)
319-- | Used internally for server connections to keep track of remote nodes for the purpose of sending notifications later.
355-- | Used internally for server connections to keep track of remote nodes for the purpose of sending notifications later.
320addClientNode :: Connection -> ProcessId -> IO ()
356addClientNode :: Connection -> ProcessId -> IO ()
321−addClientNode (RemoteProcessConnection _ _) _ = error "addClientNode called on remote connection"
357+addClientNode (RemoteProcessConnection _) _ = error "addClientNode called on remote connection"
322−addClientNode (InProcessConnection _ clientNodes _ _ _) newProcessId = atomically (STMSet.insert newProcessId clientNodes)
358+addClientNode (InProcessConnection conf) newProcessId = atomically (STMSet.insert newProcessId (ipClientNodes conf))
324-- | Discards a session, eliminating any uncommitted changes present in the session.
360-- | Discards a session, eliminating any uncommitted changes present in the session.
325closeSession :: SessionId -> Connection -> IO ()
361closeSession :: SessionId -> Connection -> IO ()
326−closeSession sessionId (InProcessConnection _ _ sessions _ _) = do
362+closeSession sessionId (InProcessConnection conf) = do
327− atomically $ STMMap.delete sessionId sessions
363+ atomically $ STMMap.delete sessionId (ipSessions conf)
328−closeSession sessionId conn@(RemoteProcessConnection _ _) = remoteCall conn (CloseSession sessionId)
364+closeSession sessionId conn@(RemoteProcessConnection _) = remoteCall conn (CloseSession sessionId)
329-- | 'close' cleans up the database access connection. Note that sessions persist even after the connection is closed.
365-- | 'close' cleans up the database access connection. Note that sessions persist even after the connection is closed.
330close :: Connection -> IO ()
366close :: Connection -> IO ()
331−close (InProcessConnection _ _ sessions _ _) = atomically $ do
367+close (InProcessConnection conf) = do
369+ let sessions = ipSessions conf
332 traverse_ (\(k,_) -> STMMap.delete k sessions) (STMMap.stream sessions)
370 traverse_ (\(k,_) -> STMMap.delete k sessions) (STMMap.stream sessions)
372+ closeLocalNode (ipLocalNode conf)
373+ closeTransport (ipTransport conf)
335−close (RemoteProcessConnection localNode serverProcessId) = do
375+close conn@(RemoteProcessConnection conf) = do
336− runProcessResult localNode $ do
376+ _ <- (remoteCall conn Logout) :: IO Bool
337− call serverProcessId Logout
377+ --putStrLn ("closing " ++ show (localNodeId (rLocalNode conf)))
379+ closeLocalNode (rLocalNode conf)
380+ closeTransport (rTransport conf)
382+--used only by the server EntryPoints
383+closeRemote_ :: Connection -> IO ()
384+closeRemote_ (InProcessConnection _) = error "invalid call of closeRemote_ on InProcessConnection"
385+closeRemote_ (RemoteProcessConnection conf) = runProcess (rLocalNode conf) (reconnect (rProcessId conf))
387+ --we need to actually close the localNode's connection to the remote
339--within the database server, we must catch and handle all exception lest they take down the database process- this handling might be different for other use-cases
388--within the database server, we must catch and handle all exception lest they take down the database process- this handling might be different for other use-cases
340--exceptions should generally *NOT* be thrown from any Project:M36 code paths, but third-party code such as AtomFunction scripts could conceivably throw undefined, etc.
389--exceptions should generally *NOT* be thrown from any Project:M36 code paths, but third-party code such as AtomFunction scripts could conceivably throw undefined, etc.
@@ -367,14 +416,17 @@ safeLogin login procId = do
367 Right val -> pure val
416 Right val -> pure val
369remoteCall :: (Serializable a, Serializable b) => Connection -> a -> IO b
418remoteCall :: (Serializable a, Serializable b) => Connection -> a -> IO b
370−remoteCall (InProcessConnection _ _ _ _ _) _ = error "remoteCall called on local connection"
419+remoteCall (InProcessConnection _ ) _ = error "remoteCall called on local connection"
371−remoteCall (RemoteProcessConnection localNode serverProcessId) arg = runProcessResult localNode $ do
420+remoteCall (RemoteProcessConnection conf) arg = runProcessResult localNode $ do
372 ret <- safeCall serverProcessId arg
421 ret <- safeCall serverProcessId arg
373 case ret of
422 case ret of
374 Left err -> error ("server died: " ++ show err)
423 Left err -> error ("server died: " ++ show err)
375 Right ret' -> case ret' of
424 Right ret' -> case ret' of
376 Left RequestTimeoutError -> liftIO (throwIO RequestTimeoutException)
425 Left RequestTimeoutError -> liftIO (throwIO RequestTimeoutException)
377 Right val -> pure val
426 Right val -> pure val
428+ localNode = rLocalNode conf
429+ serverProcessId = rProcessId conf
379sessionForSessionId :: SessionId -> Sessions -> STM (Either RelationalError Session)
431sessionForSessionId :: SessionId -> Sessions -> STM (Either RelationalError Session)
380sessionForSessionId sessionId sessions = do
432sessionForSessionId sessionId sessions = do
@@ -403,26 +455,29 @@ sessionAndSchema sessionId sessions = do
403 Right schema -> pure (Right (session, schema))
455 Right schema -> pure (Right (session, schema))
405currentSchemaName :: SessionId -> Connection -> IO (Maybe SchemaName)
457currentSchemaName :: SessionId -> Connection -> IO (Maybe SchemaName)
406−currentSchemaName sessionId (InProcessConnection _ _ sessions _ _) = atomically $ do
458+currentSchemaName sessionId (InProcessConnection conf) = atomically $ do
459+ let sessions = ipSessions conf
407 eSession <- sessionForSessionId sessionId sessions
460 eSession <- sessionForSessionId sessionId sessions
408 case eSession of
461 case eSession of
409 Left _ -> pure Nothing
462 Left _ -> pure Nothing
410 Right session -> pure (Just (Sess.schemaName session))
463 Right session -> pure (Just (Sess.schemaName session))
411−currentSchemaName sessionId conn@(RemoteProcessConnection _ _) = remoteCall conn (RetrieveCurrentSchemaName sessionId)
464+currentSchemaName sessionId conn@(RemoteProcessConnection _) = remoteCall conn (RetrieveCurrentSchemaName sessionId)
413setCurrentSchemaName :: SessionId -> Connection -> SchemaName -> IO (Maybe RelationalError)
466setCurrentSchemaName :: SessionId -> Connection -> SchemaName -> IO (Maybe RelationalError)
414−setCurrentSchemaName sessionId (InProcessConnection _ _ sessions _ _) sname = atomically $ do
467+setCurrentSchemaName sessionId (InProcessConnection conf) sname = atomically $ do
468+ let sessions = ipSessions conf
415 eSession <- sessionForSessionId sessionId sessions
469 eSession <- sessionForSessionId sessionId sessions
416 case eSession of
470 case eSession of
417 Left _ -> pure Nothing
471 Left _ -> pure Nothing
418 Right session -> case Sess.setSchemaName sname session of
472 Right session -> case Sess.setSchemaName sname session of
419 Left err -> pure (Just err)
473 Left err -> pure (Just err)
420 Right newSession -> STMMap.insert newSession sessionId sessions >> pure Nothing
474 Right newSession -> STMMap.insert newSession sessionId sessions >> pure Nothing
421−setCurrentSchemaName sessionId conn@(RemoteProcessConnection _ _) sname = remoteCall conn (ExecuteSetCurrentSchema sessionId sname)
475+setCurrentSchemaName sessionId conn@(RemoteProcessConnection _) sname = remoteCall conn (ExecuteSetCurrentSchema sessionId sname)
423-- | Execute a relational expression in the context of the session and connection. Relational expressions are queries and therefore cannot alter the database.
477-- | Execute a relational expression in the context of the session and connection. Relational expressions are queries and therefore cannot alter the database.
424executeRelationalExpr :: SessionId -> Connection -> RelationalExpr -> IO (Either RelationalError Relation)
478executeRelationalExpr :: SessionId -> Connection -> RelationalExpr -> IO (Either RelationalError Relation)
425−executeRelationalExpr sessionId (InProcessConnection _ _ sessions _ _) expr = excEither $ atomically $ do
479+executeRelationalExpr sessionId (InProcessConnection conf) expr = excEither $ atomically $ do
480+ let sessions = ipSessions conf
426 eSession <- sessionAndSchema sessionId sessions
481 eSession <- sessionAndSchema sessionId sessions
427 case eSession of
482 case eSession of
428 Left err -> pure $ Left err
483 Left err -> pure $ Left err
@@ -436,11 +491,12 @@ executeRelationalExpr sessionId (InProcessConnection _ _ sessions _ _) expr = ex
436 Right expr'' -> case evalState (RE.evalRelationalExpr expr'') (RE.mkRelationalExprState (Sess.concreteDatabaseContext session)) of
491 Right expr'' -> case evalState (RE.evalRelationalExpr expr'') (RE.mkRelationalExprState (Sess.concreteDatabaseContext session)) of
437 Left err -> pure (Left err)
492 Left err -> pure (Left err)
438 Right rel -> pure (force (Right rel)) -- this is necessary so that any undefined/error exceptions are spit out here
493 Right rel -> pure (force (Right rel)) -- this is necessary so that any undefined/error exceptions are spit out here
439−executeRelationalExpr sessionId conn@(RemoteProcessConnection _ _) relExpr = remoteCall conn (ExecuteRelationalExpr sessionId relExpr)
494+executeRelationalExpr sessionId conn@(RemoteProcessConnection _) relExpr = remoteCall conn (ExecuteRelationalExpr sessionId relExpr)
441-- | 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.
496-- | 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.
442executeDatabaseContextExpr :: SessionId -> Connection -> DatabaseContextExpr -> IO (Maybe RelationalError)
497executeDatabaseContextExpr :: SessionId -> Connection -> DatabaseContextExpr -> IO (Maybe RelationalError)
443−executeDatabaseContextExpr sessionId (InProcessConnection _ _ sessions _ _) expr = excMaybe $ atomically $ do
498+executeDatabaseContextExpr sessionId (InProcessConnection conf) expr = excMaybe $ atomically $ do
499+ let sessions = ipSessions conf
444 eSession <- sessionAndSchema sessionId sessions
500 eSession <- sessionAndSchema sessionId sessions
445 case eSession of
501 case eSession of
446 Left err -> pure $ Just err
502 Left err -> pure $ Just err
@@ -461,12 +517,14 @@ executeDatabaseContextExpr sessionId (InProcessConnection _ _ sessions _ _) expr
461 STMMap.insert newSession sessionId sessions
517 STMMap.insert newSession sessionId sessions
462 pure Nothing
518 pure Nothing
464−executeDatabaseContextExpr sessionId conn@(RemoteProcessConnection _ _) dbExpr = remoteCall conn (ExecuteDatabaseContextExpr sessionId dbExpr)
520+executeDatabaseContextExpr sessionId conn@(RemoteProcessConnection _) dbExpr = remoteCall conn (ExecuteDatabaseContextExpr sessionId dbExpr)
466-- | Execute a database context IO-monad-based expression for the given session and connection. `DatabaseContextIOExpr` modify the DatabaseContext but cannot be purely implemented.
522-- | Execute a database context IO-monad-based expression for the given session and connection. `DatabaseContextIOExpr` modify the DatabaseContext but cannot be purely implemented.
467--this is almost completely identical to executeDatabaseContextExpr above
523--this is almost completely identical to executeDatabaseContextExpr above
468executeDatabaseContextIOExpr :: SessionId -> Connection -> DatabaseContextIOExpr -> IO (Maybe RelationalError)
524executeDatabaseContextIOExpr :: SessionId -> Connection -> DatabaseContextIOExpr -> IO (Maybe RelationalError)
469−executeDatabaseContextIOExpr sessionId (InProcessConnection _ _ sessions _ scriptSession) expr = excMaybe $ do
525+executeDatabaseContextIOExpr sessionId (InProcessConnection conf) expr = excMaybe $ do
526+ let sessions = ipSessions conf
527+ scriptSession = ipScriptSession conf
470 eSession <- atomically $ sessionForSessionId sessionId sessions --potentially race condition due to interleaved IO?
528 eSession <- atomically $ sessionForSessionId sessionId sessions --potentially race condition due to interleaved IO?
471 case eSession of
529 case eSession of
472 Left err -> pure $ Just err
530 Left err -> pure $ Just err
@@ -480,7 +538,7 @@ executeDatabaseContextIOExpr sessionId (InProcessConnection _ _ sessions _ scrip
480 newSession = Session newDiscon (Sess.schemaName session)
538 newSession = Session newDiscon (Sess.schemaName session)
481 atomically $ STMMap.insert newSession sessionId sessions
539 atomically $ STMMap.insert newSession sessionId sessions
482 pure Nothing
540 pure Nothing
483−executeDatabaseContextIOExpr sessionId conn@(RemoteProcessConnection _ _) dbExpr = remoteCall conn (ExecuteDatabaseContextIOExpr sessionId dbExpr)
541+executeDatabaseContextIOExpr sessionId conn@(RemoteProcessConnection _) dbExpr = remoteCall conn (ExecuteDatabaseContextIOExpr sessionId dbExpr)
485executeGraphExprSTM_ :: TransactionId -> SessionId -> Sessions -> TransactionGraphOperator -> TVar TransactionGraph -> STM (Either RelationalError TransactionGraph)
543executeGraphExprSTM_ :: TransactionId -> SessionId -> Sessions -> TransactionGraphOperator -> TVar TransactionGraph -> STM (Either RelationalError TransactionGraph)
486executeGraphExprSTM_ freshId sessionId sessions graphExpr graphTvar = do
544executeGraphExprSTM_ freshId sessionId sessions graphExpr graphTvar = do
@@ -508,7 +566,11 @@ executeCommitExprSTM_ oldContext newContext nodes = do
509-- | Execute a transaction graph expression in the context of the session and connection. Transaction graph operators modify the transaction graph state.
567-- | Execute a transaction graph expression in the context of the session and connection. Transaction graph operators modify the transaction graph state.
510executeGraphExpr :: SessionId -> Connection -> TransactionGraphOperator -> IO (Maybe RelationalError)
568executeGraphExpr :: SessionId -> Connection -> TransactionGraphOperator -> IO (Maybe RelationalError)
511−executeGraphExpr sessionId (InProcessConnection strat clientNodes sessions graphTvar _) graphExpr = excMaybe $ do
569+executeGraphExpr sessionId (InProcessConnection conf) graphExpr = excMaybe $ do
570+ let strat = ipPersistenceStrategy conf
571+ clientNodes = ipClientNodes conf
572+ sessions = ipSessions conf
573+ graphTvar = ipTransactionGraph conf
512 freshId <- nextRandom
574 freshId <- nextRandom
513 manip <- atomically $ do
575 manip <- atomically $ do
514 eSession <- sessionForSessionId sessionId sessions
576 eSession <- sessionForSessionId sessionId sessions
@@ -535,9 +597,9 @@ executeGraphExpr sessionId (InProcessConnection strat clientNodes sessions graph
535 --update filesystem database, if necessary
597 --update filesystem database, if necessary
536 --this should really grab a lock at the beginning of the method to be threadsafe
598 --this should really grab a lock at the beginning of the method to be threadsafe
537 processPersistence strat newGraph
599 processPersistence strat newGraph
538− sendNotifications nodesToNotify notsToFire
600+ sendNotifications nodesToNotify (ipLocalNode conf) notsToFire
539 return Nothing
601 return Nothing
540−executeGraphExpr sessionId conn@(RemoteProcessConnection _ _) graphExpr = remoteCall conn (ExecuteGraphExpr sessionId graphExpr)
602+executeGraphExpr sessionId conn@(RemoteProcessConnection _) graphExpr = remoteCall conn (ExecuteGraphExpr sessionId graphExpr)
543commitSTM_ :: UUID -> SessionId -> Sessions -> TVar TransactionGraph -> STM (Maybe RelationalError)
605commitSTM_ :: UUID -> SessionId -> Sessions -> TVar TransactionGraph -> STM (Maybe RelationalError)
@@ -550,17 +612,19 @@ commitSTM_ freshUUID sessionId sessions graph = do
551-- | A trans-graph expression is a relational query executed against the entirety of a transaction graph.
613-- | A trans-graph expression is a relational query executed against the entirety of a transaction graph.
552executeTransGraphRelationalExpr :: SessionId -> Connection -> TransGraphRelationalExpr -> IO (Either RelationalError Relation)
614executeTransGraphRelationalExpr :: SessionId -> Connection -> TransGraphRelationalExpr -> IO (Either RelationalError Relation)
553−executeTransGraphRelationalExpr _ (InProcessConnection _ _ _ graphTvar _) tgraphExpr = excEither . atomically $ do
615+executeTransGraphRelationalExpr _ (InProcessConnection conf) tgraphExpr = excEither . atomically $ do
616+ let graphTvar = ipTransactionGraph conf
554 graph <- readTVar graphTvar
617 graph <- readTVar graphTvar
555 case evalTransGraphRelationalExpr tgraphExpr graph of
618 case evalTransGraphRelationalExpr tgraphExpr graph of
556 Left err -> pure (Left err)
619 Left err -> pure (Left err)
557 Right relExpr -> case evalState (RE.evalRelationalExpr relExpr) (RE.mkRelationalExprState RE.emptyDatabaseContext) of
620 Right relExpr -> case evalState (RE.evalRelationalExpr relExpr) (RE.mkRelationalExprState RE.emptyDatabaseContext) of
558 Left err -> pure (Left err)
621 Left err -> pure (Left err)
559 Right rel -> pure (force (Right rel))
622 Right rel -> pure (force (Right rel))
560−executeTransGraphRelationalExpr sessionId conn@(RemoteProcessConnection _ _) tgraphExpr = remoteCall conn (ExecuteTransGraphRelationalExpr sessionId tgraphExpr)
623+executeTransGraphRelationalExpr sessionId conn@(RemoteProcessConnection _) tgraphExpr = remoteCall conn (ExecuteTransGraphRelationalExpr sessionId tgraphExpr)
562executeSchemaExpr :: SessionId -> Connection -> Schema.SchemaExpr -> IO (Maybe RelationalError)
625executeSchemaExpr :: SessionId -> Connection -> Schema.SchemaExpr -> IO (Maybe RelationalError)
563−executeSchemaExpr sessionId (InProcessConnection _ _ sessions _ _) schemaExpr = atomically $ do
626+executeSchemaExpr sessionId (InProcessConnection conf) schemaExpr = atomically $ do
627+ let sessions = ipSessions conf
564 eSession <- sessionAndSchema sessionId sessions
628 eSession <- sessionAndSchema sessionId sessions
565 case eSession of
629 case eSession of
566 Left err -> pure (Just err)
630 Left err -> pure (Just err)
@@ -575,26 +639,23 @@ executeSchemaExpr sessionId (InProcessConnection _ _ sessions _ _) schemaExpr =
575 newSession = Session (DisconnectedTransaction (Discon.parentId discon) newSchemas) (Sess.schemaName session)
639 newSession = Session (DisconnectedTransaction (Discon.parentId discon) newSchemas) (Sess.schemaName session)
576 STMMap.insert newSession sessionId sessions
640 STMMap.insert newSession sessionId sessions
577 pure Nothing
641 pure Nothing
578−executeSchemaExpr sessionId conn@(RemoteProcessConnection _ _) schemaExpr = remoteCall conn (ExecuteSchemaExpr sessionId schemaExpr)
642+executeSchemaExpr sessionId conn@(RemoteProcessConnection _) schemaExpr = remoteCall conn (ExecuteSchemaExpr sessionId schemaExpr)
580-- | 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.
644-- | 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.
581commit :: SessionId -> Connection -> IO (Maybe RelationalError)
645commit :: SessionId -> Connection -> IO (Maybe RelationalError)
582−commit sessionId conn@(InProcessConnection _ _ _ _ _) = executeGraphExpr sessionId conn Commit
646+commit sessionId conn@(InProcessConnection _) = executeGraphExpr sessionId conn Commit
583−commit sessionId conn@(RemoteProcessConnection _ _) = remoteCall conn (ExecuteGraphExpr sessionId Commit)
647+commit sessionId conn@(RemoteProcessConnection _) = remoteCall conn (ExecuteGraphExpr sessionId Commit)
585−sendNotifications :: [ProcessId] -> EvaluatedNotifications -> IO ()
649+sendNotifications :: [ProcessId] -> LocalNode -> EvaluatedNotifications -> IO ()
586−sendNotifications pids nots = mapM_ sendNots pids
650+sendNotifications pids localNode nots = mapM_ sendNots pids
588 sendNots remoteClientPid = do
652 sendNots remoteClientPid = do
589− eLocalNode <- commonLocalNode
591− Left err -> error ("Failed to get local node: " ++ show err)
592− Right localNode -> when (not (M.null nots)) $ runProcess localNode $ send remoteClientPid (NotificationMessage nots)
653+ when (not (M.null nots)) $ runProcess localNode $ send remoteClientPid (NotificationMessage nots)
594-- | 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.
655-- | 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.
595rollback :: SessionId -> Connection -> IO (Maybe RelationalError)
656rollback :: SessionId -> Connection -> IO (Maybe RelationalError)
596−rollback sessionId conn@(InProcessConnection _ _ _ _ _) = executeGraphExpr sessionId conn Rollback
657+rollback sessionId conn@(InProcessConnection _) = executeGraphExpr sessionId conn Rollback
597−rollback sessionId conn@(RemoteProcessConnection _ _) = remoteCall conn (ExecuteGraphExpr sessionId Rollback)
658+rollback sessionId conn@(RemoteProcessConnection _) = remoteCall conn (ExecuteGraphExpr sessionId Rollback)
599processPersistence :: PersistenceStrategy -> TransactionGraph -> IO ()
660processPersistence :: PersistenceStrategy -> TransactionGraph -> IO ()
600processPersistence NoPersistence _ = return ()
661processPersistence NoPersistence _ = return ()
@@ -603,11 +664,12 @@ processPersistence (CrashSafePersistence dbdir) graph = transactionGraphPersist
604-- | 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.
665-- | 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.
605typeForRelationalExpr :: SessionId -> Connection -> RelationalExpr -> IO (Either RelationalError Relation)
666typeForRelationalExpr :: SessionId -> Connection -> RelationalExpr -> IO (Either RelationalError Relation)
606−typeForRelationalExpr sessionId conn@(InProcessConnection _ _ _ _ _) relExpr = atomically $ typeForRelationalExprSTM sessionId conn relExpr
667+typeForRelationalExpr sessionId conn@(InProcessConnection _) relExpr = atomically $ typeForRelationalExprSTM sessionId conn relExpr
607−typeForRelationalExpr sessionId conn@(RemoteProcessConnection _ _) relExpr = remoteCall conn (ExecuteTypeForRelationalExpr sessionId relExpr)
668+typeForRelationalExpr sessionId conn@(RemoteProcessConnection _) relExpr = remoteCall conn (ExecuteTypeForRelationalExpr sessionId relExpr)
609typeForRelationalExprSTM :: SessionId -> Connection -> RelationalExpr -> STM (Either RelationalError Relation)
670typeForRelationalExprSTM :: SessionId -> Connection -> RelationalExpr -> STM (Either RelationalError Relation)
610−typeForRelationalExprSTM sessionId (InProcessConnection _ _ sessions _ _) relExpr = do
671+typeForRelationalExprSTM sessionId (InProcessConnection conf) relExpr = do
672+ let sessions = ipSessions conf
611 eSession <- sessionAndSchema sessionId sessions
673 eSession <- sessionAndSchema sessionId sessions
612 case eSession of
674 case eSession of
613 Left err -> pure $ Left err
675 Left err -> pure $ Left err
@@ -624,7 +686,8 @@ typeForRelationalExprSTM _ _ _ = error "typeForRelationalExprSTM called on non-l
625-- | Return a 'Map' of the database's constraints at the context of the session and connection.
687-- | Return a 'Map' of the database's constraints at the context of the session and connection.
626inclusionDependencies :: SessionId -> Connection -> IO (Either RelationalError InclusionDependencies)
688inclusionDependencies :: SessionId -> Connection -> IO (Either RelationalError InclusionDependencies)
627−inclusionDependencies sessionId (InProcessConnection _ _ sessions _ _) = do
689+inclusionDependencies sessionId (InProcessConnection conf) = do
690+ let sessions = ipSessions conf
628 atomically $ do
691 atomically $ do
629 eSession <- sessionAndSchema sessionId sessions
692 eSession <- sessionAndSchema sessionId sessions
630 case eSession of
693 case eSession of
@@ -636,11 +699,12 @@ inclusionDependencies sessionId (InProcessConnection _ _ sessions _ _) = do
637 pure (Schema.inclusionDependenciesInSchema schema (B.inclusionDependencies context))
700 pure (Schema.inclusionDependenciesInSchema schema (B.inclusionDependencies context))
639−inclusionDependencies sessionId conn@(RemoteProcessConnection _ _) = remoteCall conn (RetrieveInclusionDependencies sessionId)
702+inclusionDependencies sessionId conn@(RemoteProcessConnection _) = remoteCall conn (RetrieveInclusionDependencies sessionId)
641-- | 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.
704-- | 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.
642planForDatabaseContextExpr :: SessionId -> Connection -> DatabaseContextExpr -> IO (Either RelationalError DatabaseContextExpr)
705planForDatabaseContextExpr :: SessionId -> Connection -> DatabaseContextExpr -> IO (Either RelationalError DatabaseContextExpr)
643−planForDatabaseContextExpr sessionId (InProcessConnection _ _ sessions _ _) dbExpr = do
706+planForDatabaseContextExpr sessionId (InProcessConnection conf) dbExpr = do
707+ let sessions = ipSessions conf
644 atomically $ do
708 atomically $ do
645 eSession <- sessionAndSchema sessionId sessions
709 eSession <- sessionAndSchema sessionId sessions
646 case eSession of
710 case eSession of
@@ -650,7 +714,7 @@ planForDatabaseContextExpr sessionId (InProcessConnection _ _ sessions _ _) dbEx
650 else -- don't show any optimization because the current optimization infrastructure relies on access to the base context- this probably underscores the need for each schema to have its own DatabaseContext, even if it is generated on-the-fly
714 else -- don't show any optimization because the current optimization infrastructure relies on access to the base context- this probably underscores the need for each schema to have its own DatabaseContext, even if it is generated on-the-fly
651 pure (Right dbExpr)
715 pure (Right dbExpr)
653−planForDatabaseContextExpr sessionId conn@(RemoteProcessConnection _ _) dbExpr = remoteCall conn (RetrievePlanForDatabaseContextExpr sessionId dbExpr)
717+planForDatabaseContextExpr sessionId conn@(RemoteProcessConnection _) dbExpr = remoteCall conn (RetrievePlanForDatabaseContextExpr sessionId dbExpr)
655-- | Return a relation which represents the current state of the global transaction graph. The attributes are
719-- | Return a relation which represents the current state of the global transaction graph. The attributes are
656-- * current- boolean attribute representing whether or not the current session references this transaction
720-- * current- boolean attribute representing whether or not the current session references this transaction
@@ -658,7 +722,9 @@ planForDatabaseContextExpr sessionId conn@(RemoteProcessConnection _ _) dbExpr =
658-- * id- id attribute of the transaction
722-- * id- id attribute of the transaction
659-- * parents- a relation-valued attribute which contains a relation of transaction ids which are parent transaction to the transaction
723-- * parents- a relation-valued attribute which contains a relation of transaction ids which are parent transaction to the transaction
660transactionGraphAsRelation :: SessionId -> Connection -> IO (Either RelationalError Relation)
724transactionGraphAsRelation :: SessionId -> Connection -> IO (Either RelationalError Relation)
661−transactionGraphAsRelation sessionId (InProcessConnection _ _ sessions tvar _) = do
725+transactionGraphAsRelation sessionId (InProcessConnection conf) = do
726+ let sessions = ipSessions conf
727+ tvar = ipTransactionGraph conf
662 atomically $ do
728 atomically $ do
663 eSession <- sessionForSessionId sessionId sessions
729 eSession <- sessionForSessionId sessionId sessions
664 case eSession of
730 case eSession of
@@ -667,10 +733,11 @@ transactionGraphAsRelation sessionId (InProcessConnection _ _ sessions tvar _) =
667 graph <- readTVar tvar
733 graph <- readTVar tvar
668 pure $ graphAsRelation (Sess.disconnectedTransaction session) graph
734 pure $ graphAsRelation (Sess.disconnectedTransaction session) graph
670−transactionGraphAsRelation sessionId conn@(RemoteProcessConnection _ _) = remoteCall conn (RetrieveTransactionGraph sessionId)
736+transactionGraphAsRelation sessionId conn@(RemoteProcessConnection _) = remoteCall conn (RetrieveTransactionGraph sessionId)
672relationVariablesAsRelation :: SessionId -> Connection -> IO (Either RelationalError Relation)
738relationVariablesAsRelation :: SessionId -> Connection -> IO (Either RelationalError Relation)
673−relationVariablesAsRelation sessionId (InProcessConnection _ _ sessions _ _) = do
739+relationVariablesAsRelation sessionId (InProcessConnection conf) = do
740+ let sessions = ipSessions conf
674 atomically $ do
741 atomically $ do
675 eSession <- sessionAndSchema sessionId sessions
742 eSession <- sessionAndSchema sessionId sessions
676 case eSession of
743 case eSession of
@@ -684,17 +751,18 @@ relationVariablesAsRelation sessionId (InProcessConnection _ _ sessions _ _) = d
684 Left err -> pure (Left err)
751 Left err -> pure (Left err)
685 Right relvars -> pure $ R.relationVariablesAsRelation relvars
752 Right relvars -> pure $ R.relationVariablesAsRelation relvars
687−relationVariablesAsRelation sessionId conn@(RemoteProcessConnection _ _) = remoteCall conn (RetrieveRelationVariableSummary sessionId)
754+relationVariablesAsRelation sessionId conn@(RemoteProcessConnection _) = remoteCall conn (RetrieveRelationVariableSummary sessionId)
689-- | Returns the transaction id for the connection's disconnected transaction committed parent transaction.
756-- | Returns the transaction id for the connection's disconnected transaction committed parent transaction.
690headTransactionId :: SessionId -> Connection -> IO (Maybe TransactionId)
757headTransactionId :: SessionId -> Connection -> IO (Maybe TransactionId)
691−headTransactionId sessionId (InProcessConnection _ _ sessions _ _) = do
758+headTransactionId sessionId (InProcessConnection conf) = do
759+ let sessions = ipSessions conf
692 atomically $ do
760 atomically $ do
693 eSession <- sessionForSessionId sessionId sessions
761 eSession <- sessionForSessionId sessionId sessions
694 case eSession of
762 case eSession of
695 Left _ -> pure Nothing
763 Left _ -> pure Nothing
696 Right session -> pure $ Just (Sess.parentId session)
764 Right session -> pure $ Just (Sess.parentId session)
697−headTransactionId sessionId conn@(RemoteProcessConnection _ _) = remoteCall conn (RetrieveHeadTransactionId sessionId)
765+headTransactionId sessionId conn@(RemoteProcessConnection _) = remoteCall conn (RetrieveHeadTransactionId sessionId)
699headNameSTM_ :: SessionId -> Sessions -> TVar TransactionGraph -> STM (Maybe HeadName)
767headNameSTM_ :: SessionId -> Sessions -> TVar TransactionGraph -> STM (Maybe HeadName)
700headNameSTM_ sessionId sessions graphTvar = do
768headNameSTM_ sessionId sessions graphTvar = do
@@ -708,12 +776,15 @@ headNameSTM_ sessionId sessions graphTvar = do
709-- | Returns Just the name of the head of the current disconnected transaction or Nothing.
777-- | Returns Just the name of the head of the current disconnected transaction or Nothing.
710headName :: SessionId -> Connection -> IO (Maybe HeadName)
778headName :: SessionId -> Connection -> IO (Maybe HeadName)
711−headName sessionId (InProcessConnection _ _ sessions graphTvar _) = do
779+headName sessionId (InProcessConnection conf) = do
780+ let sessions = ipSessions conf
781+ graphTvar = ipTransactionGraph conf
712 atomically (headNameSTM_ sessionId sessions graphTvar)
782 atomically (headNameSTM_ sessionId sessions graphTvar)
713−headName sessionId conn@(RemoteProcessConnection _ _) = remoteCall conn (ExecuteHeadName sessionId)
783+headName sessionId conn@(RemoteProcessConnection _) = remoteCall conn (ExecuteHeadName sessionId)
715atomTypesAsRelation :: SessionId -> Connection -> IO (Either RelationalError Relation)
785atomTypesAsRelation :: SessionId -> Connection -> IO (Either RelationalError Relation)
716−atomTypesAsRelation sessionId (InProcessConnection _ _ sessions _ _) = do
786+atomTypesAsRelation sessionId (InProcessConnection conf) = do
787+ let sessions = ipSessions conf
717 atomically $ do
788 atomically $ do
718 eSession <- sessionForSessionId sessionId sessions
789 eSession <- sessionForSessionId sessionId sessions
719 case eSession of
790 case eSession of
@@ -722,9 +793,25 @@ atomTypesAsRelation sessionId (InProcessConnection _ _ sessions _ _) = do
722 case typesAsRelation (typeConstructorMapping (Sess.concreteDatabaseContext session)) of
793 case typesAsRelation (typeConstructorMapping (Sess.concreteDatabaseContext session)) of
723 Left err -> pure (Left err)
794 Left err -> pure (Left err)
724 Right rel -> pure (Right rel)
795 Right rel -> pure (Right rel)
725−atomTypesAsRelation sessionId conn@(RemoteProcessConnection _ _) = remoteCall conn (RetrieveAtomTypesAsRelation sessionId)
796+atomTypesAsRelation sessionId conn@(RemoteProcessConnection _) = remoteCall conn (RetrieveAtomTypesAsRelation sessionId)
727--used only for testing- we expect this to throw an exception
798--used only for testing- we expect this to throw an exception
728callTestTimeout_ :: SessionId -> Connection -> IO Bool
799callTestTimeout_ :: SessionId -> Connection -> IO Bool
729−callTestTimeout_ _ (InProcessConnection _ _ _ _ _) = error "bad testing call"
800+callTestTimeout_ _ (InProcessConnection _) = error "bad testing call"
730−callTestTimeout_ sessionId conn@(RemoteProcessConnection _ _) = remoteCall conn (TestTimeout sessionId)
801+callTestTimeout_ sessionId conn@(RemoteProcessConnection _) = remoteCall conn (TestTimeout sessionId)
804+transactionGraph_ :: Connection -> IO TransactionGraph
805+transactionGraph_ (InProcessConnection conf) = atomically $ readTVar (ipTransactionGraph conf)
806+transactionGraph_ _ = error "remote connection used"
809+disconnectedTransaction_ :: SessionId -> Connection -> IO DisconnectedTransaction
810+disconnectedTransaction_ sessionId (InProcessConnection conf) = do
811+ let sessions = ipSessions conf
812+ mSession <- atomically $ do
813+ STMMap.lookup sessionId sessions
815+ Nothing -> error "No such session"
816+ Just (Sess.Session discon _) -> pure discon
817+disconnectedTransaction_ _ _= error "remote connection used"