Merge pull request #13 from ralmond/mongo-split

Mongo split

Author: ralmond

.gitignore

@@ -1,3 +1,6 @@
 *~
-
-makeDBuri.Rd~
+makeDBuri.Rd~
+.Rproj.user
+.Rhistory
+.DS_Store
+\#*

ChangeLog

@@ -1,3 +1,57 @@
+2023-07-06  Russell Almond  <ralmond@cherry>
+
+	* R/MessageQueue.R ($count): Added $count method to MessageQueue,
+	ListQueue and MongoQueue. 
+
+2023-06-23  Russell Almond  <ralmond@cherry>
+
+	* R/MessageQueue.R (MessageQueue,MongoQueue,ListQueue): New class
+	for a collection of messages.  `MongoQueue` is stored in a
+	database, and `ListQueue` is just a list of queues.
+	(fetchNextMessage): New generic function for fetching the next
+	method from a queue.
+	(markAsProcessed,markAsError): Added methods for message queue.
+	(importMessages): Imports messages into the database.
+	(cleanMessageQueue): Removes unneeded messages from the database.
+	(resetProcessedMessages): Resets `processed` field of selected
+	messages. 
+	(purgeMessageQueue_): Removes unneded messages form the database.
+
+	* R/InjectionListeners.R (listenerDataTable): -- Method which
+	extracts the 
+	contents of a message data table as a data frame.
+	* R/Listeners.R (registerOutput): Logs an output file as available
+	in the outputdatabase.
+	(buildListenerSet): wrapper for building whole
+	listener set from JSON description.
+
+	* DESCRIPTION (Version): Fairly major refactoring, moving a number
+	of things from EABN to here.
+	
+
+2023-05-25  Russell Almond  <ralmond@cherry>
+
+	* R/Listeners.R: ($notifyListeners):  Moved responsibility for setting `sender` field of message to here.
+
+2023-05-24  Russell Almond  <ralmond@cherry>
+
+	* R/InjectionListener.R (InjectionListener): Changed to pass DB to the class.
+
+	* R/UpdateListener.R (UpdateListener): Changed to pass DB to the class.
+
+	* R/UpsertListener.R (UpsertListener): Changed to pass DB to the class.
+	
+
+	* R/Listeners.R: Moved construction of DB link out of constructors.
+	(buildListener):   dburi and ssl_options are set by the caller, not the config.json This function now builds the Mongo object for the listener.
+
+
+2023-05-20  Russell Almond  <ralmond@cherry>
+
+	* R/Listeners.R: Moved various listener types into their own files.
+
+	* DESCRIPTION (Imports): Moved the database and json commands to the `mongo` package.
+
 2022-04-11  Russell Almond  <[email protected]>
 
 	* R/Message.R (parseSimpleData): forced the _id column to always have the name "oid" (necessary for monog).

DESCRIPTION

@@ -1,6 +1,6 @@
 Package: Proc4
-Version: 0.6-1
-Date: 2022/04/25
+Version: 0.8-5
+Date: 2023/06/23
 Title: Four Process Assessment Database and Dispatcher
 Authors@R: 
     person(given = "Russell",
@@ -11,10 +11,22 @@ Authors@R:
 Author: Russell Almond
 Maintainer: Russell Almond <[email protected]>
 Depends: R (>= 3.0), methods,
-Imports: jsonlite, futile.logger
-Suggests:  mongolite, knitr, rmarkdown, tidyr, CPTtools
+Imports: futile.logger, mongo, jsonlite
+Suggests: utils, mongolite, knitr, rmarkdown, tidyr, CPTtools, rlang, bookdown, devtools, withr, testthat (>= 3.0.0), Peanut
 Description: Utilities for working with messages in the four four process architecture as json objects.
+Collate:  ErrorHandling.R Message.R MessageQueue.R Listeners.R CaptureListener.R InjectionListener.R TableListener.R UpdateListener.R UpsertListener.R
 License: Artistic-2.0 
 URL: https://pluto.coe.fsu.edu/Proc4
 VignetteBuilder: knitr
-
+Support: c(
+   'Bill & Melinda Gates Foundation grant
+   "Games as Learning/Assessment:  Stealth Assessment" (#0PP1035331,
+   Val Shute, PI)',
+   'National Science Foundation grant "DIP:
+   Game-based Assessment and Support of STEM-related Competencies"
+   (#1628937, Val Shute, PI)',
+   'National Science Foundation grant "Mathematical Learning via
+   Architectual Design and Modeling Using E-Rebuild." (\#1720533,
+   Fengfeng Ke, PI)',
+   'Institute of Educational Statistics Grant: "Exploring adaptive cognitive and affective learning support for next-generation STEM learning games." (#R305A170376-20, Val Shute and Russell Almond, PIs')	
+Config/testthat/edition: 3

NAMESPACE

@@ -1,39 +1,50 @@
 import(methods)
 import(futile.logger)
-importFrom("jsonlite", "toJSON", "fromJSON", "unbox","base64_enc","base64_dec",
+import(mongo)
+importFrom("jsonlite", "base64_enc","base64_dec",
            "serializeJSON", "unserializeJSON")
 importFrom("utils", "limitedLabels")
 
 
 export(P4Message)
-exportClasses(P4Message)
-export(m_id,"m_id<-",app,uid,
+exportClasses(P4Message,MessageQueue,MongoQueue,ListQueue)
+export(app,"app<-",uid,"uid<-",
        context,"context<-",
        timestamp,"timestamp<-",
-       sender,mess,details,all.equal.P4Message,
+       sender,"sender<-",mess,"mess<-",details,
+       all.equal.P4Message,
        processed, markAsProcessed, processingError, markAsError,
-       makeDBuri)
+       serializeData,fetchNextMessage,
+       resetProcessedMessages,cleanMessageQueue,
+       importMessages)
 S3method(all.equal,P4Message,all.equal.P4Message)
-exportMethods(m_id,"m_id<-",app,uid,context,"context<-",
-              timestamp,"timestamp<-",sender,mess,details,
-              toString,show, processed, processingError)
+exportMethods(app,"app<-",uid,"uid<-",context,"context<-",
+              timestamp,"timestamp<-",sender,"sender<-",
+              mess,"mess<-",details,"details<-",
+              toString,show, processed, processingError,
+              markAsProcessed, markAsError,
+              resetProcessedMessages,cleanMessageQueue,
+              importMessages)
 
-export(as.json,as.jlist,saveRec,buildJQuery,buildJQterm,
-       parseMessage,parseData,unparseData,parseSimpleData,
-       unboxer,ununboxer, cleanMessageJlist,
-       getOneRec,getManyRecs)
-exportMethods(as.json,as.jlist)
+
+export(buildMessage, cleanMessageJlist)
+exportMethods("as.jlist","parse.jlist")
 
 export(receiveMessage,isListener,ListenerSet,notifyListeners,
+       listeningFor, clearMessages,
        InjectionListener,UpdateListener,UpsertListener,CaptureListener,
-       TableListener, buildListener,listenerName, resetListeners)
-exportClasses(ListenerSet,NullListenerSet,MongoDB,CaptureListener,
-       InjectionListener,
+       TableListener, buildListener, listenerName, resetListeners,
+       buildListenerSet, registerOutput, listenerDataTable,
+       updateTable,generateListenerExports)
+exportClasses(ListenerSet,NullListenerSet,RefListener,
+       CaptureListener,InjectionListener,
        UpdateListener,UpsertListener,TableListener)
 exportMethods(isListener,receiveMessage,notifyListeners,listenerName,
-              resetListeners)
+              resetListeners,clearMessages,listeningFor, registerOutput,
+              listenerDataTable)
 
-export(withFlogging)
+exportClasses(mongoAppender)
+export(withFlogging,mongoAppender)
 
 
 

Proc4.Rproj

@@ -0,0 +1,20 @@
+Version: 1.0
+
+RestoreWorkspace: Default
+SaveWorkspace: Default
+AlwaysSaveHistory: Default
+
+EnableCodeIndexing: Yes
+UseSpacesForTab: Yes
+NumSpacesForTab: 2
+Encoding: UTF-8
+
+RnwWeave: Sweave
+LaTeX: pdfLaTeX
+
+AutoAppendNewline: Yes
+StripTrailingWhitespace: Yes
+
+BuildType: Package
+PackageUseDevtools: Yes
+PackageInstallArgs: --no-multiarch --with-keep.source

R/CaptureListener.R

@@ -0,0 +1,36 @@
+
+#################################################
+## CaptureListener
+
+## This a simple listener whose goal is to simply hold the message to
+## it can be checked later.
+
+CaptureListener <-
+  setRefClass("CaptureListener",
+              fields=c(messages="list"),
+              methods=list(
+                  initialize = function(name="Capture",messages=list(),...)
+                    callSuper(name=name,messages=messages,...),
+                  receiveMessage = function (message) {
+                        messages <<- c(message,messages)
+                  },
+                  lastMessage = function() {
+                    if(length(messages)==0L) return(NULL)
+                    messages[[1]]
+                  },
+                  reset = function(app) {
+                    messages <<- list()
+                  }),
+                  contains="RefListener")
+
+CaptureListener <- function (name="Capture",messages=list(),messSet=character(),...) {
+  new("CaptureListener",name=name,messages=messages,messSet=messSet)
+}
+
+setMethod("listenerDataTable","TableListener",
+          function(listener, appid=character())
+            as.data.frame(sapply(messages,attributes))
+)
+
+
+

R/ErrorHandling.R

@@ -122,6 +122,47 @@ withFlogging <- function(expr,...,context=deparse(substitute(expr)),
           invisible(structure(msg, class = "try-error", condition = obj)))
 }
 
+## Breaks futile.logger line into its parts.
+parseline <- function(line) {
+  bracket1 <- regexpr("[",line,fixed=TRUE)
+  bracket2 <- regexpr("]",line,fixed=TRUE)
+  list(level=trimws(substr(line,1,bracket1-1)),
+       time=strptime(substr(line,bracket1+1,bracket2-1),format="%Y-%m-%d %H:%M:%S"),
+       message=trimws(substring(line,bracket2+1)))
+}
+
+## Forces Date into Mongo Format.
+mongoDate <- function (dtime) {
+  jsonlite::fromJSON(jsonlite::toJSON(mongo::unboxer(dtime),POSIXt="mongo"),FALSE)
+}
+
+mongoAppender <-
+  setRefClass("mongoAppender",
+              fields=c(db="JSONDB",
+                       app="character",
+                       engine="character",
+                       tee="character"),
+              methods=c(
+                initialize = function(app="default",
+                                      engine="Unspecified",
+                                      db=mongo::MongoDB("Log",noMongo=TRUE),
+                                      tee=character()) {
+                  callSuper(db=db,app=app,engine=engine,tee=tee)
+                },
+                logit = function(line) {
+                  pline <- parseline(line)
+                  entry <- buildJQuery(app=app,engine=engine,level=pline$level,
+                                       timestamp=pline$time,
+                                       message=pline$message)
+                  mdbInsert(db,entry)
+                  if (length(tee) > 0L) {
+                    cat(line,file=tee,append=TRUE,sep="")
+                  }
+                },
+                logger = function() {
+                  function(line) {.self$logit(line)}
+                }))
+
 
 shinyAppender <-
   setRefClass("shinyAppender",
@@ -136,14 +177,22 @@ shinyAppender <-
                     callSuper(file=file, field=field, steps=steps,
                               messages=messages)
                   },
-                  update = function (line)
+                  update = function (line, output, renderer=function(tab) shiny::renderTable(tab,colname=FALSE))
                   {
                     if (length(file) == 1L && nchar(file)>0L) {
                       cat(line, file = file, append = TRUE, sep = "")
                     }
                     messages$Messages <<- c(messages$Messages,line)
                     if (length(field)==1L && nchar(field)>0L) {
-                      output[[field]] <- renderTable(messages,colnames=FALSE)
+                      # Call the render for a test case.
+                      if (is.null(output)) do.call(renderer,list(messages))
+                    } else {
+                      output[[field]] <- do.call(renderer,list(messages))
+                    }
+                  },
+                  logger = function() {
+                    function (line) {
+                      .self$update(line)
                     }
                   }
               ))

R/InjectionListener.R

@@ -0,0 +1,71 @@
+#################################################
+## InjectionListener
+
+## This a simple listener whose goal is to simply to inject the
+## message into a mongo collection where it can be used as a queue.
+
+InjectionListener <-
+  setRefClass("InjectionListener",
+              list(),
+              methods=list(
+                  initialize=
+                    function(name="Injection",
+                             db=mongo::MongoDB(noMongo=TRUE),
+                             messSet=character(),
+                             ...) {
+                      callSuper(name=name,db=db,
+                                messSet=messSet,
+                                ...)
+                    },
+                  messdb = function() {db},
+                  receiveMessage = function (message) {
+                      flog.debug("Sending message %s",toString(message))
+                      flog.debug(".. from %s",sender(message))
+                      flog.trace("Message:",x=as.jlist(message,
+                                                       attributes(message)),
+                                 capture=TRUE)
+                      mongo::m_id(message) <- NA_character_
+                      mdbInsert(messdb(),as.json(message,serialize=TRUE))
+                  },
+                  reset = function(app) {
+                      mdbRemove(messdb(),buildJQuery(app=app))
+                  }
+              ),
+              contains="RefListener")
+
+InjectionListener <- function (name="Injection",
+                               db=mongo::MongoDB(noMongo=TRUE),
+                               messSet=character(),
+                               ...) {
+  new("InjectionListener",name=name,db=db,messSet=messSet,...)
+}
+
+
+setMethod("listenerDataTable","InjectionListener",
+          function(listener,appid=character()) {
+            stat1 <- mdbFind(listener$messdb(),buildJQuery(app=appid))
+            if (isTRUE(nrow(stat1) > 0L)) {
+              stat1data <- t(sapply(stat1$data,parseData))
+              sdat <- data.frame(stat1[,c("app","uid","context","timestamp")],
+                                 stat1data)
+              sdat$app <- basename(sdat$app)
+              return(sdat)
+            } else {
+              flog.warn("No records in statistics file.")
+              return(NULL)
+            }
+          })
+
+## These are left here as they may be useful later; however, they are
+buildHistMat <- function (col, app, uid) {
+  stat1 <- mdbFind(col,buildJQuery(app=app, uid=uid))
+  stat1d <- lapply(stat1$data,function (d) Peanut::flattenStats(parseData(d)))
+  data.frame(stat1[,c("app","uid","context","timestamp")],
+             do.call(rbind,stat1d))
+}
+
+buildAppHist <- function (col, app) {
+  uids <- col$distinct("uid",buildJQuery(app=app))
+  do.call(rbind,
+          lapply(uids,function(u) buildHistMat(col, app, u)))
+}