@@ -1,4 +1,4 @@
1−{-# LANGUAGE DeriveAnyClass, DeriveGeneric, ScopedTypeVariables, MonoLocalBinds #-}
1+{-# LANGUAGE DeriveAnyClass, DeriveGeneric, ScopedTypeVariables, MonoLocalBinds, DerivingVia, GeneralizedNewtypeDeriving #-}
22{-|
33Module: ProjectM36.Client
44
@@ -53,11 +53,9 @@ module ProjectM36.Client
5353 Attribute(..),
5454 MergeStrategy(..),
5555 attributesFromList,
56− createNodeId,
5756 createSessionAtCommit,
5857 createSessionAtHead,
5958 closeSession,
60− addClientNode,
6159 callTestTimeout_,
6260 RelationCardinality(..),
6361 TransactionGraphOperator(..),
@@ -67,7 +65,6 @@ module ProjectM36.Client
6765 TransGraphRelationalExpr,
6866 TransactionIdLookup(..),
6967 TransactionIdHeadBacktrack(..),
70− NodeId(..),
7168 Atom(..),
7269 Session,
7370 SessionId,
@@ -141,6 +138,7 @@ import Data.UUID.V4 (nextRandom)
141138import Data.Word
142139import Control.Exception (IOException, handle, AsyncException, throwIO, fromException, Exception)
143140import Control.Concurrent.MVar
141+import Codec.Winery hiding (Schema)
144142import qualified Data.Map as M
145143#if MIN_VERSION_stm_containers(1,0,0)
146144import qualified StmContainers.Map as StmMap
@@ -152,12 +150,15 @@ import qualified STMContainers.Set as StmSet
152150import qualified ProjectM36.Session as Sess
153151import ProjectM36.Session
154152import ProjectM36.Sessions
155−import Data.Binary (Binary)
156153import GHC.Generics (Generic)
157154import Control.DeepSeq (force)
158155import System.IO
159156import Data.Time.Clock
160157import Data.Typeable
158+import qualified Network.RPC.Curryer.Client as RPC
159+import qualified Network.RPC.Curryer.Server as RPC
160+import Network.Socket (PortNumber, AddrInfo(..), getAddrInfo, defaultHints, AddrInfoFlag(..), SocketType(..), ServiceName, hostAddressToTuple, SockAddr(..))
161+import Data.Word
161162
162163type Hostname = String
163164
@@ -185,14 +186,15 @@ data RequestTimeoutException = RequestTimeoutException
185186instance Exception RequestTimeoutException
186187
187188-- | Construct a 'ConnectionInfo' to describe how to make the 'Connection'. The database can be run within the current process or running remotely via distributed-process.
188−data ConnectionInfo = InProcessConnectionInfo PersistenceStrategy NotificationCallback [GhcPkgPath]|
189− RemoteProcessConnectionInfo DatabaseName NodeId NotificationCallback
189+data ConnectionInfo = InProcessConnectionInfo PersistenceStrategy NotificationCallback [GhcPkgPath] |
190+ RemoteConnectionInfo DatabaseName Hostname ServiceName NotificationCallback
190191
191192type EvaluatedNotifications = M.Map NotificationName EvaluatedNotification
192193
193194-- | Used for callbacks from the server when monitored changes have been made.
194195newtype NotificationMessage = NotificationMessage EvaluatedNotifications
195− deriving (Binary, Eq, Show, Generic)
196+ deriving (Eq, Show, Generic)
197+ deriving Serialise via WineryVariant NotificationMessage
196198
197199-- | When a notification is fired, the 'reportOldExpr' is evaluated in the commit's pre-change context while the 'reportNewExpr' is evaluated in the post-change context and they are returned along with the original notification.
198200data EvaluatedNotification = EvaluatedNotification {
@@ -200,16 +202,13 @@ data EvaluatedNotification = EvaluatedNotification {
200202 reportOldRelation :: Either RelationalError Relation,
201203 reportNewRelation :: Either RelationalError Relation
202204 }
203− deriving(Binary, Eq, Show, Generic)
205+ deriving (Eq, Show, Generic)
206+ deriving Serialise via WineryRecord EvaluatedNotification
204207
205208
206−-- | Create a 'NodeId' for use in connecting to a remote server using distributed-process.
207−createNodeId :: Hostname -> Port -> NodeId
208−createNodeId host port = NodeId $ encodeEndPointAddress host (show port) 1
209−
210209-- | Use this for connecting to remote servers on the default port.
211−defaultServerPort :: Port
212−defaultServerPort = 6543
210+defaultServerPort :: ServiceName
211+defaultServerPort = "6543"
213212
214213-- | Use this for connecting to remote servers with the default database name.
215214defaultDatabaseName :: DatabaseName
@@ -221,10 +220,14 @@ defaultHeadName = "master"
221220
222221-- | Create a connection configuration which connects to the localhost on the default server port and default server database name. The configured notification callback is set to ignore all events.
223222defaultRemoteConnectionInfo :: ConnectionInfo
224−defaultRemoteConnectionInfo = RemoteProcessConnectionInfo defaultDatabaseName (createNodeId "127.0.0.1" defaultServerPort) emptyNotificationCallback
223+defaultRemoteConnectionInfo =
224+ RemoteConnectionInfo defaultDatabaseName defaultServerHostname defaultServerPort emptyNotificationCallback
225+
226+defaultServerHostname :: Hostname
227+defaultServerHostname = "localhost"
225228
226229-- | The 'Connection' represents either local or remote access to a database. All operations flow through the connection.
227−type ClientNodes = StmSet.Set ProcessId
230+type ClientNodes = StmSet.Set RPCConnection -- not correct- we need an object to represent all client connection on the server side - this is missing in Curryer
228231
229232-- internal structure specific to in-process connections
230233data InProcessConnectionConf = InProcessConnectionConf {
@@ -233,19 +236,15 @@ data InProcessConnectionConf = InProcessConnectionConf {
233236 ipSessions :: Sessions,
234237 ipTransactionGraph :: TVar TransactionGraph,
235238 ipScriptSession :: Maybe ScriptSession,
236− ipLocalNode :: LocalNode,
237− ipTransport :: Transport, -- we hold onto this so that we can close it gracefully
238239 ipLocks :: Maybe (LockFile, MVar LockFileHash) -- nothing when NoPersistence
239240 }
240241
241−data RemoteProcessConnectionConf = RemoteProcessConnectionConf {
242− rLocalNode :: LocalNode,
243− rProcessId :: ProcessId, --remote processId
244− rTransport :: Transport --the TCP socket transport
242+data RemoteConnectionConf = RemoteConnectionConf {
243+ rConn :: RPCConnection
245244 }
246245
247246data Connection = InProcessConnection InProcessConnectionConf |
248− RemoteProcessConnection RemoteProcessConnectionConf
247+ RemoteConnection RemoteConnectionConf
249248
250249-- | There are several reasons why a connection can fail.
251250data ConnectionError = SetupDatabaseDirectoryError PersistenceError |
@@ -258,39 +257,6 @@ data ConnectionError = SetupDatabaseDirectoryError PersistenceError |
258257remoteDBLookupName :: DatabaseName -> String
259258remoteDBLookupName = (++) "db-"
260259
261−createLocalNode :: IO (LocalNode, Transport)
262−createLocalNode = do
263−#if MIN_VERSION_network_transport_tcp(0,7,0)
264− eLocalTransport <- createTransport (defaultTCPAddr "127.0.0.1" "0") defaultTCPParameters
265−#elif MIN_VERSION_network_transport_tcp(0,6,0)
266− eLocalTransport <- createTransport "127.0.0.1" "0" (\sn -> ("127.0.0.1", sn)) defaultTCPParameters
267−#else
268− eLocalTransport <- createTransport "127.0.0.1" "0" defaultTCPParameters
269−#endif
270− case eLocalTransport of
271− Left err -> error ("failed to create transport: " ++ show err)
272− Right localTransport -> do
273− localNode <- newLocalNode localTransport initRemoteTable
274− pure (localNode, localTransport)
275−
276−notificationListener :: NotificationCallback -> Process ()
277−notificationListener callback = do
278− --pid <- getSelfPid
279− --liftIO $ putStrLn $ "LISTENER THREAD START " ++ show pid
280− _ <- forever $
281− receiveWait [
282− match (\(NotificationMessage eNots) ->
283− --say $ "NOTIFICATION: " ++ show eNots
284− -- when notifications are thrown, they are not adjusted for the current schema which could be problematic, but we don't have access to the current session here
285− liftIO $ mapM_ (uncurry callback) (M.toList eNots)
286− )
287− ]
288− --say "LISTENER THREAD EXIT"
289− pure ()
290−
291−startNotificationListener :: LocalNode -> NotificationCallback -> IO ProcessId
292−startNotificationListener localNode callback = forkProcess localNode (notificationListener callback)
293−
294260createScriptSession :: [String] -> IO (Maybe ScriptSession)
295261createScriptSession ghcPkgPaths = do
296262 eScriptSession <- initScriptSession ghcPkgPaths
@@ -312,8 +278,6 @@ connectProjectM36 (InProcessConnectionInfo strat notificationCallback ghcPkgPath
312278 graphTvar <- newTVarIO freshGraph
313279 clientNodes <- StmSet.newIO
314280 sessions <- StmMap.newIO
315− (localNode, transport) <- createLocalNode
316− notificationPid <- startNotificationListener localNode notificationCallback
317281 mScriptSession <- createScriptSession ghcPkgPaths
318282
319283 let conn = InProcessConnection InProcessConnectionConf {
@@ -322,22 +286,27 @@ connectProjectM36 (InProcessConnectionInfo strat notificationCallback ghcPkgPath
322286 ipSessions = sessions,
323287 ipTransactionGraph = graphTvar,
324288 ipScriptSession = mScriptSession,
325− ipLocalNode = localNode,
326− ipTransport = transport,
327289 ipLocks = Nothing}
328− addClientNode conn notificationPid
329290 pure (Right conn)
330291 MinimalPersistence dbdir -> connectPersistentProjectM36 strat NoDiskSync dbdir freshGraph notificationCallback ghcPkgPaths
331292 CrashSafePersistence dbdir -> connectPersistentProjectM36 strat FsyncDiskSync dbdir freshGraph notificationCallback ghcPkgPaths
332293
333−connectProjectM36 (RemoteProcessConnectionInfo databaseName serverNodeId notificationCallback) = do
334− connStatus <- newEmptyMVar
335− (localNode, transport) <- createLocalNode
336− let dbName = remoteDBLookupName databaseName
294+connectProjectM36 (RemoteConnectionInfo databaseName hostName servicePort notificationCallback) = do
337295 --putStrLn $ "Connecting to " ++ show serverNodeId ++ " " ++ dbName
338− notificationListenerPid <- startNotificationListener localNode notificationCallback
339− runProcess localNode $ do
340− --even a failed connection retains an open socket!
296+ --TODO- add notification callback thread
297+ let resolutionHints = defaultHints { addrFlags = [AI_NUMERICHOST, AI_NUMERICSERV],
298+ addrSocketType = Stream
299+ }
300+ resolved <- getAddrInfo (Just resolutionHints) (Just hostName) (Just servicePort)
301+ case resolved of
302+ [] -> error ("DNS resolution failed for" <> hostName <> ":" <> servicePort)
303+ addrInfo:_ -> do
304+ --supports IPv4 only for now
305+ let (SockAddrInet port addr) = addrAddress addrInfo
306+ conn <- RPC.connect (const (pure ())) (hostAddressToTuple addr) port
307+ --TODO handle connection errors!
308+ pure (Right (RemoteConnection (RemoteConnectionConf conn)))
309+ {-
341310 mServerProcessId <- whereisRemote serverNodeId dbName
342311 case mServerProcessId of
343312 Nothing -> liftIO $ putMVar connStatus $ Left (NoSuchDatabaseByNameError databaseName)
@@ -346,8 +315,9 @@ connectProjectM36 (RemoteProcessConnectionInfo databaseName serverNodeId notific
346315 if not loginConfirmation then
347316 liftIO $ putMVar connStatus (Left LoginError)
348317 else
349− liftIO $ putMVar connStatus (Right $ RemoteProcessConnection RemoteProcessConnectionConf {rLocalNode = localNode, rProcessId = serverProcessId, rTransport = transport})
318+ liftIO $ putMVar connStatus (Right $ RemoteProcessConnection RemoteProcessConnectionConf {rConn = )
350319 takeMVar connStatus
320+-}
351321
352322connectPersistentProjectM36 :: PersistenceStrategy ->
353323 DiskSync ->
@@ -372,7 +342,6 @@ connectPersistentProjectM36 strat sync dbdir freshGraph notificationCallback ghc
372342 tvarGraph <- newTVarIO graph'
373343 sessions <- StmMap.newIO
374344 clientNodes <- StmSet.newIO
375− (localNode, transport) <- createLocalNode
376345 lockMVar <- newMVar digest
377346 let conn = InProcessConnection InProcessConnectionConf {
378347 ipPersistenceStrategy = strat,
@@ -380,13 +349,9 @@ connectPersistentProjectM36 strat sync dbdir freshGraph notificationCallback ghc
380349 ipSessions = sessions,
381350 ipTransactionGraph = tvarGraph,
382351 ipScriptSession = mScriptSession,
383− ipLocalNode = localNode,
384− ipTransport = transport,
385352 ipLocks = Just (lockFileH, lockMVar)
386353 }
387354
388− notificationPid <- startNotificationListener localNode notificationCallback
389− addClientNode conn notificationPid
390355 pure (Right conn)
391356
392357-- | Create a new session at the transaction id and return the session's Id.
@@ -394,7 +359,7 @@ createSessionAtCommit :: Connection -> TransactionId -> IO (Either RelationalErr
394359createSessionAtCommit conn@(InProcessConnection _) commitId = do
395360 newSessionId <- nextRandom
396361 atomically $ createSessionAtCommit_ commitId newSessionId conn
397−createSessionAtCommit conn@(RemoteProcessConnection _) uuid = remoteCall conn (CreateSessionAtCommit uuid)
362+createSessionAtCommit conn@(RemoteConnection _) uuid = remoteCall conn (CreateSessionAtCommit uuid)
398363
399364createSessionAtCommit_ :: TransactionId -> SessionId -> Connection -> STM (Either RelationalError SessionId)
400365createSessionAtCommit_ commitId newSessionId (InProcessConnection conf) = do
@@ -411,7 +376,7 @@ createSessionAtCommit_ commitId newSessionId (InProcessConnection conf) = do
411376 Nothing -> do
412377 StmMap.insert (Session freshDiscon defaultSchemaName) newSessionId sessions
413378 pure $ Right newSessionId
414−createSessionAtCommit_ _ _ (RemoteProcessConnection _) = error "createSessionAtCommit_ called on remote connection"
379+createSessionAtCommit_ _ _ (RemoteConnection _) = error "createSessionAtCommit_ called on remote connection"
415380
416381-- | 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.
417382createSessionAtHead :: Connection -> HeadName -> IO (Either RelationalError SessionId)
@@ -423,18 +388,13 @@ createSessionAtHead conn@(InProcessConnection conf) headn = do
423388 case transactionForHead headn graph of
424389 Nothing -> pure $ Left (NoSuchHeadNameError headn)
425390 Just trans -> createSessionAtCommit_ (transactionId trans) newSessionId conn
426−createSessionAtHead conn@(RemoteProcessConnection _) headn = remoteCall conn (CreateSessionAtHead headn)
427−
428−-- | Used internally for server connections to keep track of remote nodes for the purpose of sending notifications later.
429−addClientNode :: Connection -> ProcessId -> IO ()
430−addClientNode (RemoteProcessConnection _) _ = error "addClientNode called on remote connection"
431−addClientNode (InProcessConnection conf) newProcessId = atomically (StmSet.insert newProcessId (ipClientNodes conf))
391+createSessionAtHead conn@(RemoteConnection _) headn = remoteCall conn (CreateSessionAtHead headn)
432392
433393-- | Discards a session, eliminating any uncommitted changes present in the session.
434394closeSession :: SessionId -> Connection -> IO ()
435395closeSession sessionId (InProcessConnection conf) =
436396 atomically $ StmMap.delete sessionId (ipSessions conf)
437−closeSession sessionId conn@(RemoteProcessConnection _) = remoteCall conn (CloseSession sessionId)
397+closeSession sessionId conn@(RemoteConnection _) = remoteCall conn (CloseSession sessionId)
438398
439399-- | 'close' cleans up the database access connection and closes any relevant sockets.
440400close :: Connection -> IO ()
@@ -447,22 +407,18 @@ close (InProcessConnection conf) = do
447407 StmMap.deleteAll sessions
448408#endif
449409 pure ()
450− closeLocalNode (ipLocalNode conf)
451− closeTransport (ipTransport conf)
452410 let mLocks = ipLocks conf
453411 case mLocks of
454412 Nothing -> pure ()
455413 Just (lockFileH, _) -> closeLockFile lockFileH
456414
457−close conn@(RemoteProcessConnection conf) = do
458− _ <- remoteCall conn Logout :: IO Bool
459− closeLocalNode (rLocalNode conf)
460− closeTransport (rTransport conf)
415+close (RemoteConnection (RemoteConnectionConf conn)) = do
416+ RPC.close conn
461417
462418--used only by the server EntryPoints
463419closeRemote_ :: Connection -> IO ()
464420closeRemote_ (InProcessConnection _) = error "invalid call of closeRemote_ on InProcessConnection"
465−closeRemote_ (RemoteProcessConnection conf) = runProcess (rLocalNode conf) (reconnect (rProcessId conf))
421+closeRemote_ (RemoteConnection (RemoteConnectionConf conn)) = RPC.close conn
466422
467423 --we need to actually close the localNode's connection to the remote
468424--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
@@ -473,35 +429,20 @@ excEither = handle handler
473429 where
474430 handler exc | Just (_ :: AsyncException) <- fromException exc = throwIO exc
475431 | otherwise = pure (Left (UnhandledExceptionError (show exc)))
476−
477−runProcessResult :: LocalNode -> Process a -> IO a
478−runProcessResult localNode proc = do
479− ret <- newEmptyMVar
480− runProcess localNode $ do
481− val <- proc
482− liftIO $ putMVar ret val
483− takeMVar ret
484432
433+{-
485434safeLogin :: Login -> ProcessId -> Process Bool
486435safeLogin login procId = do
487436 ret <- call procId login
488437 case ret of
489438 Left (_ :: ServerError) -> pure False
490439 Right val -> pure val
440+-}
491441
492−remoteCall :: (Binary a, Binary b, Typeable a, Typeable b) => Connection -> a -> IO b
442+remoteCall :: (Serialise a, Serialise b) => Connection -> a -> IO b
493443remoteCall (InProcessConnection _ ) _ = error "remoteCall called on local connection"
494−remoteCall (RemoteProcessConnection conf) arg = runProcessResult localNode $ do
495− ret <- safeCall serverProcessId arg
496− case ret of
497− Left err -> error ("server died: " ++ show err)
498− Right ret' -> case ret' of
499− Left RequestTimeoutError -> liftIO (throwIO RequestTimeoutException)
500− Left (ProcessDiedError _) -> liftIO (throwIO RemoteProcessDiedException)
501− Right val -> pure val
502− where
503− localNode = rLocalNode conf
504− serverProcessId = rProcessId conf
444+remoteCall (RemoteConnection conf) arg = error "not implemented"
445+
505446
506447sessionForSessionId :: SessionId -> Sessions -> STM (Either RelationalError Session)
507448sessionForSessionId sessionId sessions =
@@ -536,7 +477,7 @@ currentSchemaName sessionId (InProcessConnection conf) = atomically $ do
536477 case eSession of
537478 Left err -> pure (Left err)
538479 Right session -> pure (Right (Sess.schemaName session))
539−currentSchemaName sessionId conn@(RemoteProcessConnection _) = remoteCall conn (RetrieveCurrentSchemaName sessionId)
480+currentSchemaName sessionId conn@(RemoteConnection _) = remoteCall conn (RetrieveCurrentSchemaName sessionId)
540481
541482-- | Switch to the named isomorphic schema.
542483setCurrentSchemaName :: SessionId -> Connection -> SchemaName -> IO (Either RelationalError ())
@@ -548,7 +489,7 @@ setCurrentSchemaName sessionId (InProcessConnection conf) sname = atomically $ d
548489 Right session -> case Sess.setSchemaName sname session of
549490 Left err -> pure (Left err)
550491 Right newSession -> StmMap.insert newSession sessionId sessions >> pure (Right ())
551−setCurrentSchemaName sessionId conn@(RemoteProcessConnection _) sname = remoteCall conn (ExecuteSetCurrentSchema sessionId sname)
492+setCurrentSchemaName sessionId conn@(RemoteConnection _) sname = remoteCall conn (ExecuteSetCurrentSchema sessionId sname)
552493
553494-- | Execute a relational expression in the context of the session and connection. Relational expressions are queries and therefore cannot alter the database.
554495executeRelationalExpr :: SessionId -> Connection -> RelationalExpr -> IO (Either RelationalError Relation)
@@ -572,7 +513,7 @@ executeRelationalExpr sessionId (InProcessConnection conf) expr = excEither $ at
572513 Right rel -> pure (force (Right rel)) -- this is necessary so that any undefined/error exceptions are spit out here
573514 Left err -> pure (Left err)
574515
575−executeRelationalExpr sessionId conn@(RemoteProcessConnection _) relExpr = remoteCall conn (ExecuteRelationalExpr sessionId relExpr)
516+executeRelationalExpr sessionId conn@(RemoteConnection _) relExpr = remoteCall conn (ExecuteRelationalExpr sessionId relExpr)
576517
577518-- | 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.
578519executeDatabaseContextExpr :: SessionId -> Connection -> DatabaseContextExpr -> IO (Either RelationalError ())
@@ -606,7 +547,7 @@ executeDatabaseContextExpr sessionId (InProcessConnection conf) expr = excEither
606547 newSession = Session newDiscon (Sess.schemaName session)
607548 StmMap.insert newSession sessionId sessions
608549 pure (Right ())
609−executeDatabaseContextExpr sessionId conn@(RemoteProcessConnection _) dbExpr = remoteCall conn (ExecuteDatabaseContextExpr sessionId dbExpr)
550+executeDatabaseContextExpr sessionId conn@(RemoteConnection _) dbExpr = remoteCall conn (ExecuteDatabaseContextExpr sessionId dbExpr)
610551
611552-- | Similar to a git rebase, 'autoMergeToHead' atomically creates a temporary branch and merges it to the latest commit of the branch referred to by the 'HeadName' and commits the merge. This is useful to reduce incidents of 'TransactionIsNotAHeadError's but at the risk of merge errors (thus making it similar to rebasing). Alternatively, as an optimization, if a simple commit is possible (meaning that the head has not changed), then a fast-forward commit takes place instead.
612553autoMergeToHead :: SessionId -> Connection -> MergeStrategy -> HeadName -> IO (Either RelationalError ())
@@ -635,7 +576,7 @@ autoMergeToHead sessionId (InProcessConnection conf) strat headName' = do
635576 Left err -> pure (Left err)
636577 Right ((discon', graph'), transactionIdsAdded) ->
637578 pure (Right (discon', graph', transactionIdsAdded))
638−autoMergeToHead sessionId conn@(RemoteProcessConnection _) strat headName' = remoteCall conn (ExecuteAutoMergeToHead sessionId strat headName')
579+autoMergeToHead sessionId conn@(RemoteConnection _) strat headName' = remoteCall conn (ExecuteAutoMergeToHead sessionId strat headName')
639580
640581-- | Execute a database context IO-monad-based expression for the given session and connection. `DatabaseContextIOExpr`s modify the DatabaseContext but cannot be purely implemented.
641582--this is almost completely identical to executeDatabaseContextExpr above
@@ -661,7 +602,7 @@ executeDatabaseContextIOExpr sessionId (InProcessConnection conf) expr = excEith
661602 context' = RE.dbc_context newState
662603 atomically $ StmMap.insert newSession sessionId sessions
663604 pure (Right ())
664−executeDatabaseContextIOExpr sessionId conn@(RemoteProcessConnection _) dbExpr = remoteCall conn (ExecuteDatabaseContextIOExpr sessionId dbExpr)
605+executeDatabaseContextIOExpr sessionId conn@(RemoteConnection _) dbExpr = remoteCall conn (ExecuteDatabaseContextIOExpr sessionId dbExpr)
665606
666607{-
667608executeGraphExprSTM_ :: TransactionId -> SessionId -> Session -> Sessions -> TransactionGraphOperator -> TransactionGraph -> TVar TransactionGraph -> STM (Either RelationalError (TransactionGraph, DisconnectedTransaction)
@@ -788,7 +729,7 @@ executeGraphExpr sessionId (InProcessConnection conf) graphExpr = excEither $ do
788729 sendNotifications nodesToNotify (ipLocalNode conf) notsToFire
789730 pure Nothing
790731-}
791−executeGraphExpr sessionId conn@(RemoteProcessConnection _) graphExpr = remoteCall conn (ExecuteGraphExpr sessionId graphExpr)
732+executeGraphExpr sessionId conn@(RemoteConnection _) graphExpr = remoteCall conn (ExecuteGraphExpr sessionId graphExpr)
792733
793734-- | A trans-graph expression is a relational query executed against the entirety of a transaction graph.
794735executeTransGraphRelationalExpr :: SessionId -> Connection -> TransGraphRelationalExpr -> IO (Either RelationalError Relation)
@@ -799,7 +740,7 @@ executeTransGraphRelationalExpr _ (InProcessConnection conf) tgraphExpr = excEit
799740{- case runReader (RE.evalRelationalExpr relExpr) (RE.mkRelationalExprState DBC.empty) of
800741 Left err -> pure (Left err)
801742 Right rel -> pure (force (Right rel))-}
802−executeTransGraphRelationalExpr sessionId conn@(RemoteProcessConnection _) tgraphExpr = remoteCall conn (ExecuteTransGraphRelationalExpr sessionId tgraphExpr)
743+executeTransGraphRelationalExpr sessionId conn@(RemoteConnection _) tgraphExpr = remoteCall conn (ExecuteTransGraphRelationalExpr sessionId tgraphExpr)
803744
804745-- | Schema expressions manipulate the isomorphic schemas for the current 'DatabaseContext'.
805746executeSchemaExpr :: SessionId -> Connection -> Schema.SchemaExpr -> IO (Either RelationalError ())
@@ -822,23 +763,24 @@ executeSchemaExpr sessionId (InProcessConnection conf) schemaExpr = atomically $
822763 newSession = Session (DisconnectedTransaction (Discon.parentId discon) newSchemas False) (Sess.schemaName session)
823764 StmMap.insert newSession sessionId sessions
824765 pure (Right ())
825−executeSchemaExpr sessionId conn@(RemoteProcessConnection _) schemaExpr = remoteCall conn (ExecuteSchemaExpr sessionId schemaExpr)
766+executeSchemaExpr sessionId conn@(RemoteConnection _) schemaExpr = remoteCall conn (ExecuteSchemaExpr sessionId schemaExpr)
826767
827768-- | After modifying a 'DatabaseContext', '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.
828769commit :: SessionId -> Connection -> IO (Either RelationalError ())
829770commit sessionId conn@(InProcessConnection _) = executeGraphExpr sessionId conn Commit
830−commit sessionId conn@(RemoteProcessConnection _) = remoteCall conn (ExecuteGraphExpr sessionId Commit)
831−
832−sendNotifications :: [ProcessId] -> LocalNode -> EvaluatedNotifications -> IO ()
833−sendNotifications pids localNode nots = mapM_ sendNots pids
771+commit sessionId conn@(RemoteConnection _) = remoteCall conn (ExecuteGraphExpr sessionId Commit)
772+
773+{-
774+sendNotifications :: EvaluatedNotifications -> IO ()
775+sendNotifications nots = mapM_ sendNots pids
834776 where
835777 sendNots remoteClientPid =
836778 unless (M.null nots) $ runProcess localNode $ send remoteClientPid (NotificationMessage nots)
837−
779+ -}
838780-- | Discard any changes made in the current 'Session' and 'DatabaseContext'. This resets the disconnected transaction to reference the original database context of the parent transaction and is a very cheap operation.
839781rollback :: SessionId -> Connection -> IO (Either RelationalError ())
840782rollback sessionId conn@(InProcessConnection _) = executeGraphExpr sessionId conn Rollback
841−rollback sessionId conn@(RemoteProcessConnection _) = remoteCall conn (ExecuteGraphExpr sessionId Rollback)
783+rollback sessionId conn@(RemoteConnection _) = remoteCall conn (ExecuteGraphExpr sessionId Rollback)
842784
843785-- | Write the transaction graph to disk. This function can be used to incrementally write new transactions to disk.
844786processTransactionGraphPersistence :: PersistenceStrategy -> [TransactionId] -> TransactionGraph -> IO ()
@@ -854,7 +796,7 @@ readGraphTransactionIdDigest (CrashSafePersistence dbdir) = readGraphTransaction
854796-- | 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.
855797typeForRelationalExpr :: SessionId -> Connection -> RelationalExpr -> IO (Either RelationalError Relation)
856798typeForRelationalExpr sessionId conn@(InProcessConnection _) relExpr = atomically $ typeForRelationalExprSTM sessionId conn relExpr
857−typeForRelationalExpr sessionId conn@(RemoteProcessConnection _) relExpr = remoteCall conn (ExecuteTypeForRelationalExpr sessionId relExpr)
799+typeForRelationalExpr sessionId conn@(RemoteConnection _) relExpr = remoteCall conn (ExecuteTypeForRelationalExpr sessionId relExpr)
858800
859801typeForRelationalExprSTM :: SessionId -> Connection -> RelationalExpr -> STM (Either RelationalError Relation)
860802typeForRelationalExprSTM sessionId (InProcessConnection conf) relExpr = do
@@ -891,7 +833,7 @@ inclusionDependencies sessionId (InProcessConnection conf) = do
891833 else
892834 pure (Schema.inclusionDependenciesInSchema schema (B.inclusionDependencies context))
893835
894−inclusionDependencies sessionId conn@(RemoteProcessConnection _) = remoteCall conn (RetrieveInclusionDependencies sessionId)
836+inclusionDependencies sessionId conn@(RemoteConnection _) = remoteCall conn (RetrieveInclusionDependencies sessionId)
895837
896838typeConstructorMapping :: SessionId -> Connection -> IO (Either RelationalError TypeConstructorMapping)
897839typeConstructorMapping sessionId (InProcessConnection conf) = do
@@ -902,7 +844,7 @@ typeConstructorMapping sessionId (InProcessConnection conf) = do
902844 Left err -> pure $ Left err
903845 Right (session, _) -> --warning, no schema support for typeconstructors
904846 pure (Right (B.typeConstructorMapping (Sess.concreteDatabaseContext session)))
905−typeConstructorMapping sessionId conn@(RemoteProcessConnection _) = remoteCall conn (RetrieveTypeConstructorMapping sessionId)
847+typeConstructorMapping sessionId conn@(RemoteConnection _) = remoteCall conn (RetrieveTypeConstructorMapping sessionId)
906848
907849-- | 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.
908850planForDatabaseContextExpr :: SessionId -> Connection -> DatabaseContextExpr -> IO (Either RelationalError GraphRefDatabaseContextExpr)
@@ -922,7 +864,7 @@ planForDatabaseContextExpr sessionId (InProcessConnection conf) dbExpr = do
922864 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-}
923865 pure (Left NonConcreteSchemaPlanError)
924866
925−planForDatabaseContextExpr sessionId conn@(RemoteProcessConnection _) dbExpr = remoteCall conn (RetrievePlanForDatabaseContextExpr sessionId dbExpr)
867+planForDatabaseContextExpr sessionId conn@(RemoteConnection _) dbExpr = remoteCall conn (RetrievePlanForDatabaseContextExpr sessionId dbExpr)
926868
927869-- | Return a relation which represents the current state of the global transaction graph. The attributes are
928870-- * current- boolean attribute representing whether or not the current session references this transaction
@@ -940,7 +882,7 @@ transactionGraphAsRelation sessionId (InProcessConnection conf) = do
940882 Right session ->
941883 graphAsRelation (Sess.disconnectedTransaction session) <$> readTVar tvar
942884
943−transactionGraphAsRelation sessionId conn@(RemoteProcessConnection _) = remoteCall conn (RetrieveTransactionGraph sessionId)
885+transactionGraphAsRelation sessionId conn@(RemoteConnection _) = remoteCall conn (RetrieveTransactionGraph sessionId)
944886
945887-- | Returns the names and types of the relation variables in the current 'Session'.
946888relationVariablesAsRelation :: SessionId -> Connection -> IO (Either RelationalError Relation)
@@ -962,7 +904,7 @@ relationVariablesAsRelation sessionId (InProcessConnection conf) = do
962904 let schemaContext = context {relationVariables = relvars }
963905 pure $ RE.relationVariablesAsRelation schemaContext graph
964906
965−relationVariablesAsRelation sessionId conn@(RemoteProcessConnection _) = remoteCall conn (RetrieveRelationVariableSummary sessionId)
907+relationVariablesAsRelation sessionId conn@(RemoteConnection _) = remoteCall conn (RetrieveRelationVariableSummary sessionId)
966908
967909-- | Returns the names and types of the atom functions in the current 'Session'.
968910atomFunctionsAsRelation :: SessionId -> Connection -> IO (Either RelationalError Relation)
@@ -975,7 +917,7 @@ atomFunctionsAsRelation sessionId (InProcessConnection conf) = do
975917 Right (session, _) ->
976918 pure (AF.atomFunctionsAsRelation (atomFunctions (concreteDatabaseContext session)))
977919
978−atomFunctionsAsRelation sessionId conn@(RemoteProcessConnection _) = remoteCall conn (RetrieveAtomFunctionSummary sessionId)
920+atomFunctionsAsRelation sessionId conn@(RemoteConnection _) = remoteCall conn (RetrieveAtomFunctionSummary sessionId)
979921
980922databaseContextFunctionsAsRelation :: SessionId -> Connection -> IO (Either RelationalError Relation)
981923databaseContextFunctionsAsRelation sessionId (InProcessConnection conf) = do
@@ -987,7 +929,7 @@ databaseContextFunctionsAsRelation sessionId (InProcessConnection conf) = do
987929 Right (session, _) ->
988930 pure (DCF.databaseContextFunctionsAsRelation (dbcFunctions (concreteDatabaseContext session)))
989931
990−databaseContextFunctionsAsRelation sessionId conn@(RemoteProcessConnection _) = remoteCall conn (RetrieveDatabaseContextFunctionSummary sessionId)
932+databaseContextFunctionsAsRelation sessionId conn@(RemoteConnection _) = remoteCall conn (RetrieveDatabaseContextFunctionSummary sessionId)
991933
992934-- | Returns the transaction id for the connection's disconnected transaction committed parent transaction.
993935headTransactionId :: SessionId -> Connection -> IO (Either RelationalError TransactionId)
@@ -998,7 +940,7 @@ headTransactionId sessionId (InProcessConnection conf) = do
998940 case eSession of
999941 Left err -> pure (Left err)
1000942 Right session -> pure $ Right (Sess.parentId session)
1001−headTransactionId sessionId conn@(RemoteProcessConnection _) = remoteCall conn (RetrieveHeadTransactionId sessionId)
943+headTransactionId sessionId conn@(RemoteConnection _) = remoteCall conn (RetrieveHeadTransactionId sessionId)
1002944
1003945headNameSTM_ :: SessionId -> Sessions -> TVar TransactionGraph -> STM (Either RelationalError HeadName)
1004946headNameSTM_ sessionId sessions graphTvar = do
@@ -1018,7 +960,7 @@ headName sessionId (InProcessConnection conf) = do
1018960 let sessions = ipSessions conf
1019961 graphTvar = ipTransactionGraph conf
1020962 atomically (headNameSTM_ sessionId sessions graphTvar)
1021−headName sessionId conn@(RemoteProcessConnection _) = remoteCall conn (ExecuteHeadName sessionId)
963+headName sessionId conn@(RemoteConnection _) = remoteCall conn (ExecuteHeadName sessionId)
1022964
1023965-- | Returns a listing of all available atom types.
1024966atomTypesAsRelation :: SessionId -> Connection -> IO (Either RelationalError Relation)
@@ -1032,7 +974,7 @@ atomTypesAsRelation sessionId (InProcessConnection conf) = do
1032974 case typesAsRelation (B.typeConstructorMapping (Sess.concreteDatabaseContext session)) of
1033975 Left err -> pure (Left err)
1034976 Right rel -> pure (Right rel)
1035−atomTypesAsRelation sessionId conn@(RemoteProcessConnection _) = remoteCall conn (RetrieveAtomTypesAsRelation sessionId)
977+atomTypesAsRelation sessionId conn@(RemoteConnection _) = remoteCall conn (RetrieveAtomTypesAsRelation sessionId)
1036978
1037979disconnectedTransactionIsDirty :: SessionId -> Connection -> IO (Either RelationalError Bool)
1038980disconnectedTransactionIsDirty sessionId (InProcessConnection conf) = do
@@ -1043,12 +985,12 @@ disconnectedTransactionIsDirty sessionId (InProcessConnection conf) = do
1043985 Left err -> pure (Left err)
1044986 Right session ->
1045987 pure (Right (isDirty session))
1046−disconnectedTransactionIsDirty sessionId conn@(RemoteProcessConnection _) = remoteCall conn (RetrieveSessionIsDirty sessionId)
988+disconnectedTransactionIsDirty sessionId conn@(RemoteConnection _) = remoteCall conn (RetrieveSessionIsDirty sessionId)
1047989
1048990--used only for testing- we expect this to throw an exception
1049991callTestTimeout_ :: SessionId -> Connection -> IO Bool
1050992callTestTimeout_ _ (InProcessConnection _) = error "bad testing call"
1051−callTestTimeout_ sessionId conn@(RemoteProcessConnection _) = remoteCall conn (TestTimeout sessionId)
993+callTestTimeout_ sessionId conn@(RemoteConnection _) = remoteCall conn (TestTimeout sessionId)
1052994
1053995--used in tests only
1054996transactionGraph_ :: Connection -> IO TransactionGraph
@@ -1133,7 +1075,7 @@ commitLock_ sessionId conf stmBlock = do
11331075 Right (notsToFire, nodesToNotify, newGraph, transactionIdsToPersist) -> do
11341076 --update filesystem database, if necessary
11351077 processTransactionGraphPersistence strat transactionIdsToPersist newGraph
1136− sendNotifications nodesToNotify (ipLocalNode conf) notsToFire
1078+ --sendNotifications nodesToNotify (ipLocalNode conf) notsToFire
11371079 pure (Right ())
11381080{-
11391081writeDisconAndGraph_ :: TVar TransactionGraph -> SessionId -> Session -> Sessions -> DisconnectedTransaction -> TransactionGraph -> STM ()
@@ -1193,7 +1135,7 @@ executeDataFrameExpr sessionId conn@(InProcessConnection _) dfExpr = do
11931135 let dFrame' = maybe dFrame (`DF.drop'` dFrame) (DF.offset dfExpr)
11941136 dFrame'' = maybe dFrame' (`DF.take'` dFrame') (DF.limit dfExpr)
11951137 pure (Right dFrame'')
1196−executeDataFrameExpr sessionId conn@(RemoteProcessConnection _) dfExpr = remoteCall conn (ExecuteDataFrameExpr sessionId dfExpr)
1138+executeDataFrameExpr sessionId conn@(RemoteConnection _) dfExpr = remoteCall conn (ExecuteDataFrameExpr sessionId dfExpr)
11971139
11981140validateMerkleHashes :: SessionId -> Connection -> IO (Either RelationalError ())
11991141validateMerkleHashes sessionId (InProcessConnection conf) = do
@@ -1207,4 +1149,4 @@ validateMerkleHashes sessionId (InProcessConnection conf) = do
12071149 case Graph.validateMerkleHashes graph of
12081150 Left merkleErrs -> pure $ Left $ someErrors (map (\(MerkleValidationError tid expected actual) -> MerkleHashValidationError tid expected actual) merkleErrs)
12091151 Right () -> pure (Right ())
1210−validateMerkleHashes sessionId conn@RemoteProcessConnection{} = remoteCall conn (ExecuteValidateMerkleHashes sessionId)
1152+validateMerkleHashes sessionId conn@RemoteConnection{} = remoteCall conn (ExecuteValidateMerkleHashes sessionId)