feat: session management for Opal

DESCRIPTION

@@ -17,9 +17,10 @@ Authors@R:
              email="info@obiba.org"))
 Depends:
     R (>= 3.5),
-    opalr (>= 3.0),
-    DSI (>= 1.5),
-    methods
+    opalr (>= 3.5),
+    DSI (>= 1.8),
+    methods,
+    utils
 Description: 'DataSHIELD' is an infrastructure and series of R packages that 
     enables the remote and 'non-disclosive' analysis of sensitive research data.
     This package is the 'DataSHIELD' interface implementation for 'Opal', which is
@@ -37,14 +38,15 @@ URL: https://github.com/datashield/DSOpal/,
     https://academic.oup.com/ije/article/43/6/1929/707730,
     https://journals.plos.org/ploscompbiol/article?id=10.1371/journal.pcbi.1008880
 BugReports: https://github.com/datashield/DSOpal/issues/
-RoxygenNote: 7.3.2
+RoxygenNote: 7.3.3
 Roxygen: list(markdown = TRUE)
 Encoding: UTF-8
 Collate: 
     'DSOpal-package.R'
     'OpalDriver.R'
     'OpalConnection.R'
     'OpalResult.R'
+    'OpalSession.R'
     'datashield.aggregate.r'
     'datashield.assign.r'
     'datashield.command.r'

NAMESPACE

@@ -4,6 +4,7 @@ export(Opal)
 exportClasses(OpalConnection)
 exportClasses(OpalDriver)
 exportClasses(OpalResult)
+exportClasses(OpalSession)
 exportMethods(dsAggregate)
 exportMethods(dsAssignExpr)
 exportMethods(dsAssignResource)
@@ -13,9 +14,11 @@ exportMethods(dsDisconnect)
 exportMethods(dsFetch)
 exportMethods(dsGetInfo)
 exportMethods(dsHasResource)
+exportMethods(dsHasSession)
 exportMethods(dsHasTable)
 exportMethods(dsIsAsync)
 exportMethods(dsIsCompleted)
+exportMethods(dsIsReady)
 exportMethods(dsKeepAlive)
 exportMethods(dsListMethods)
 exportMethods(dsListPackages)
@@ -28,6 +31,9 @@ exportMethods(dsRestoreWorkspace)
 exportMethods(dsRmSymbol)
 exportMethods(dsRmWorkspace)
 exportMethods(dsSaveWorkspace)
+exportMethods(dsSession)
+exportMethods(dsStateMessage)
 import(DSI)
 import(methods)
 import(opalr)
+import(utils)

R/OpalConnection.R

@@ -44,7 +44,7 @@ setClass("OpalConnection", contains = "DSConnection", slots = list(name = "chara
 #' @export
 setMethod("dsConnect", "OpalDriver", 
           function(drv, name, restore = NULL, username = NULL, password = NULL, token = NULL, url = NULL, opts = list(), profile = NULL, ...) {
-            o <- opalr::opal.login(username, password, token, url, opts, profile=profile, restore=restore)
+            o <- opalr::opal.login(username, password, token, url, opts, profile=profile, restore=restore, context="datashield")
             o$name <- name
             con <- new("OpalConnection", name = name, opal = o)
             con
@@ -251,6 +251,56 @@ setMethod("dsHasResource", "OpalConnection", function(conn, resource) {
   }
 })
 
+#' Check remote R session exists
+#'
+#' Check if a remote R session exists (not necessarily running and ready to accept 
+#' R commands submissions).
+#'
+#' @param conn An object that inherits from \code{\link{OpalConnection-class}}.
+#' @return A logical indicating if a remote R session exists accessible through this connection.
+#'
+#' @examples
+#' \dontrun{
+#' con <- dsConnect(DSOpal::Opal(), "server1",
+#'   username = "administrator", password = "password", url = "https://opal-demo.obiba.org")
+#' dsHasSession(con)
+#' dsDisconnect(con)
+#' }
+#' @import opalr
+#' @import methods
+#' @export
+setMethod("dsHasSession", "OpalConnection", function(conn) {
+  o <- conn@opal
+  !is.null(o$rid)
+})
+
+#' Create a remote R session
+#' 
+#' Create a remote R session if none exists. If a remote R session already exists,
+#' it will be reused. Returns an object of class \code{\link{OpalSession-class}} representing
+#' the remote R session accessible through this connection.
+#' 
+#' @param conn An object that inherits from \code{\link{OpalConnection-class}}.
+#' @param async Whether the result of the call should be retrieved asynchronously. When TRUE (default) 
+#' the calls are parallelized over the connections, when the connection supports 
+#' that feature, with an extra overhead of requests.
+#' @return An object of class \code{\link{OpalSession-class}} representing the remote R session.
+#' 
+#' @examples
+#' \dontrun{
+#' con <- dsConnect(DSOpal::Opal(), "server1",
+#'   username = "dsuser", password = "password", url = "https://opal-demo.obiba.org")
+#' session <- dsSession(con, async=TRUE)
+#' dsDisconnect(con)
+#' }
+#' @import opalr
+#' @export
+setMethod("dsSession", "OpalConnection", function(conn, async=TRUE) {
+  o <- conn@opal
+  opalr::opal.session(o, wait = !async)
+  new("OpalSession", conn = conn)
+})
+
 #' Opal asynchronous support 
 #' 
 #' List that Opal supports asynchronicity on all DataSHIELD operations.
@@ -270,7 +320,7 @@ setMethod("dsHasResource", "OpalConnection", function(conn, resource) {
 #' @import methods
 #' @export
 setMethod("dsIsAsync", "OpalConnection", function(conn) {
-  list(aggregate = TRUE, assignTable = TRUE, assignResource = TRUE, assignExpr = TRUE)
+  list(session = TRUE, aggregate = TRUE, assignTable = TRUE, assignResource = TRUE, assignExpr = TRUE)
 })
 
 #' List R symbols

R/OpalSession.R

@@ -0,0 +1,94 @@
+#' @include OpalDriver.R OpalConnection.R
+NULL
+
+#' Class OpalSession.
+#'
+#' An Opal session implementing the DataSHIELD Interface (DSI)  \code{\link[DSI]{DSSession-class}}.
+#' 
+#' @import methods
+#' @import DSI
+#' @export
+#' @keywords internal
+setClass("OpalSession", contains = "DSSession", slots = list(
+  conn = "OpalConnection",
+  rval = "list"))
+
+#' Get whether the remote R session is up and running
+#' 
+#' Get the state of the remote R session and return TRUE if
+#' the state is RUNNING. Always TRUE for synchronous
+#' operations.
+#' 
+#' @param session \code{\link{OpalSession-class}} object.
+#' @return A logical indicating the readiness of the session.
+#' 
+#' @examples
+#' \dontrun{
+#' con <- dbConnect(DSOpal::Opal(), "server1",
+#'   "administrator", "password", "https://opal-demo.obiba.org")
+#' session <- dsSession(con, async = TRUE)
+#' ready <- dsIsReady(session)
+#' while (!ready) {
+#'   Sys.sleep(1)
+#'   ready <- dsIsReady(session)
+#'   cat(".")
+#' }
+#' dsDisconnect(con)
+#' }
+#' 
+#' @import opalr
+#' @import methods
+#' @export
+setMethod("dsIsReady", "OpalSession", function(session) {
+  if (is.null(session)) {
+    FALSE
+  } else {
+    o <- session@conn@opal
+    opalr::opal.session_running(o)
+  }
+})
+
+#' Get the remote R session state message
+#' 
+#' Explain the remote R session state as a human-readable message.
+#' 
+#' @param session \code{\link{OpalSession-class}} object.
+#' @return A character string
+#' 
+#' @examples
+#' \dontrun{
+#' con <- dbConnect(DSOpal::Opal(), "server1",
+#'   "administrator", "password", "https://opal-demo.obiba.org")
+#' session <- dsSession(con, async = TRUE)
+#' ready <- dsIsReady(session)
+#' while (!ready) {
+#'   Sys.sleep(1)
+#'   ready <- dsIsReady(session)
+#'   cat(dsStateMessage(session), "\n")
+#' }
+#' dsDisconnect(con)
+#' }
+#' 
+#' @import opalr
+#' @import utils
+#' @import methods
+#' @export
+setMethod("dsStateMessage", "OpalSession", function(session) {
+  if (is.null(session)) {
+    "No session"
+  } else {
+    o <- session@conn@opal
+    last_message <- utils::tail(opalr::opal.session_events(o)$message, 1)[[1]]
+    if (is.null(last_message) || is.na(last_message) || last_message == "") {
+      if (opalr::opal.version_compare(o, "5.3.0") >= 0) {
+        "No recent events"
+      } else {
+        "Ready"
+      }
+    } else {
+      last_message
+    }
+  }
+})
+
+

inst/examples/datashield-api.R

@@ -1,46 +1,58 @@
+options(verbose=FALSE)
+url <- "https://opal-demo.obiba.org"
+#url <- "http://localhost:8080"
 library(DSOpal)
-o <- dsConnect(DSOpal::Opal(), name="server1", username="administrator", password="password", url="http://localhost:8080")
-#o <- dsConnect(DSOpal::Opal(), name="server1", token="f9thEkhtXpZMoS8UEbsF09F7A8zJ1iJC", url="http://localhost:8080")
+o <- dsConnect(DSOpal::Opal(), name="server1", username="administrator", password="password", url=url, profile="default")
 o
+o@opal$context
 dsListTables(o)
-dsHasTable(o, "datashield.CNSIM1")
-dsHasTable(o, "datashield.CNSIM1xx")
+dsHasTable(o, "CNSIM.CNSIM1")
+dsHasTable(o, "CNSIM.CNSIM1xx")
 
 dsListResources(o)
 dsHasResource(o, "test.CNSIM1")
 dsHasResource(o, "test.CNSIM1xx")
 
 dsIsAsync(o)
+
+dsHasSession(o)
+session <- dsSession(o, async = TRUE)
+dsIsReady(session)
+while (!dsIsReady(session)) {
+  Sys.sleep(1)
+  cat(dsStateMessage(session), "\n")
+}
+
 rbind(dsListMethods(o, type = "aggregate"), dsListMethods(o, type = "assign"))
 dsListPackages(o)
 
-res <- dsAssignTable(o, "D", "datashield.CNSIM1", async = TRUE)
+res <- dsAssignTable(o, "D", "CNSIM.CNSIM1", async = TRUE)
 dsGetInfo(res)
 dsFetch(res)
 
 dsListResources(o)
-dsHasResource(o, "datashield.cnsim3")
-res <- dsAssignResource(o, "R1", "datashield.cnsim3", async = TRUE)
+dsHasResource(o, "CNSIM.cnsim3")
+res <- dsAssignResource(o, "R1", "CNSIM.cnsim3", async = TRUE)
 dsGetInfo(res)
 dsFetch(res)
 
-res <- dsAggregate(o, "colnames(D)", async = TRUE)
+res <- dsAggregate(o, "colnamesDS(D)", async = TRUE)
 dsGetInfo(res)
 dsFetch(res)
 
-res <- dsAggregate(o, "class(D)", async = TRUE)
+res <- dsAggregate(o, "classDS(D)", async = TRUE)
 dsGetInfo(res)
 dsFetch(res)
 
-res <- dsAggregate(o, "class(R1)", async = TRUE)
+res <- dsAggregate(o, "classDS(R1)", async = TRUE)
 dsGetInfo(res)
 dsFetch(res)
 
-res <- dsAssignTable(o, "D", "datashield.CNSIM1", id.name="id", async = FALSE)
+res <- dsAssignTable(o, "D", "CNSIM.CNSIM1", id.name="id", async = FALSE)
 dsGetInfo(res)
 dsFetch(res)
 
-res <- dsAggregate(o, "colnames(D)", async = FALSE)
+res <- dsAggregate(o, "colnamesDS(D)", async = FALSE)
 dsGetInfo(res)
 dsFetch(res)
 
@@ -52,8 +64,7 @@ dsListWorkspaces(o)
 dsRmWorkspace(o, "server1:cnsim1")
 dsDisconnect(o, save = "server1:xxx")
 
-o <- dsConnect(DSOpal::Opal(), name="server1", username="administrator", password="password", url="http://localhost:8080", restore="server1:xxx")
-#o <- dsConnect(DSOpal::Opal(), name="server1", token="f9thEkhtXpZMoS8UEbsF09F7A8zJ1iJC", url="http://localhost:8080", restore="server1:xxx")
+o <- dsConnect(DSOpal::Opal(), name="server1", username="administrator", password="password", url=url, profile="default", restore="server1:xxx")
 dsListWorkspaces(o)
 dsListSymbols(o)
 dsDisconnect(o)

inst/examples/datashield.R

@@ -7,45 +7,53 @@
 library(DSOpal)
 
 # datashield logins and assignments
-data("logindata.opal.demo")
-login.data <- logindata.opal.demo
-login.data$url <- rep("http://localhost:8080", 3)
-opals <- datashield.login(login.data, assign=T, variables=c("GENDER","PM_BMI_CONTINUOUS"))
-print(opals)
+url <- "https://opal-demo.obiba.org"
+#url <- "http://localhost:8080"
+builder <- DSI::newDSLoginBuilder()
+builder$append("server1", user="administrator", password="password", url=url, profile="default")
+builder$append("server2", user="administrator", password="password", url=url, profile="default")
+builder$append("server3", user="administrator", password="password", url=url, profile="default")
+login.data <- builder$build()
+conns <- datashield.login(login.data)
+print(conns)
+
+# start remote R sessions (optional)
+datashield.sessions(conns)
+
 # check assigned variables
-datashield.symbols(opals)
+datashield.symbols(conns)
 
 # table assignment can also happen later
-datashield.assign(opals, "T", "CNSIM.CNSIM1", variables=c("GENDER","PM_BMI_CONTINUOUS"))
-datashield.aggregate(opals,'classDS(T)')
+datashield.assign.table(conns, "D", list(server1="CNSIM.CNSIM1", server2="CNSIM.CNSIM2", server3="CNSIM.CNSIM3"))
+datashield.aggregate(conns, quote(classDS("D")))
 
-# execute some aggregate calls (if these methods are available in the opals)
-datashield.aggregate(opals,'colnamesDS(D)')
-datashield.aggregate(opals,quote(lengthDS(D$GENDER)))
+# execute some aggregate calls (if these methods are available in the conns)
+datashield.aggregate(conns, quote(colnamesDS("D")))
+datashield.aggregate(conns, quote(lengthDS("D$GENDER")))
 
 # clean symbols
-datashield.rm(opals,'D')
-datashield.symbols(opals)
+datashield.rm(conns, "D")
+datashield.symbols(conns)
 
 # assign and aggregate arbitrary values
-datashield.assign(opals, "x", quote(c("1", "2", "3")))
-datashield.aggregate(opals,quote(lengthDS(x)))
-datashield.aggregate(opals,'classDS(x)')
-
-datashield.methods(opals, type="aggregate")
-datashield.methods(opals$server1, type="aggregate")
-datashield.method_status(opals, type="assign")
-datashield.pkg_status(opals)
-datashield.table_status(opals, list(server1="CNSIM.CNSIM1", server2="CNSIM.CNSIM2", server3="CNSIM.CNSIM3"))
-
-datashield.logout(opals, save = "test")
-
-opals <- datashield.login(logindata.opal.demo, assign=FALSE, restore = "test")
-datashield.symbols(opals)
-datashield.workspaces(opals)
-datashield.workspace_save(opals, "toto")
-datashield.workspaces(opals)
-datashield.workspace_rm(opals, "toto")
-datashield.workspaces(opals)
-datashield.logout(opals)
+datashield.assign(conns, "x", quote(c("1", "2", "3")))
+datashield.aggregate(conns, quote(lengthDS("x")))
+datashield.aggregate(conns, quote(classDS(x)))
+
+datashield.methods(conns, type="aggregate")
+datashield.methods(conns$server1, type="aggregate")
+datashield.method_status(conns, type="assign")
+datashield.pkg_status(conns)
+datashield.table_status(conns, list(server1="CNSIM.CNSIM1", server2="CNSIM.CNSIM2", server3="CNSIM.CNSIM3"))
+
+datashield.logout(conns, save = "test")
+
+conns <- datashield.login(login.data, assign=FALSE, restore = "test")
+datashield.symbols(conns)
+datashield.workspaces(conns)
+datashield.workspace_save(conns, "toto")
+datashield.workspaces(conns)
+datashield.workspace_rm(conns, "toto")
+datashield.workspaces(conns)
+datashield.logout(conns)