iterors extracted to separate package

Author: crowding

.Rbuildignore

@@ -6,6 +6,7 @@
 ^_pkgdown\.yml$
 ^docs$
 ^pkgdown$
+^graphs$
 ^vignettes/README.md$
 ^tests/testthat/graphs
 ^ctz.svg$

DESCRIPTION

@@ -1,6 +1,6 @@
 Package: async
 Title: Coroutines: Generators / Yield, Async / Await, and Streams
-Version: 0.3.1
+Version: 0.3.2
 Date: 2023-03-01
 Authors@R: 
     person(given = "Peter",
@@ -25,6 +25,7 @@ Encoding: UTF-8
 LazyData: true
 Depends: R (>= 4.1)
 Imports:
+        iterors,
         nseval (>= 0.4.3),
         later,
         promises,
@@ -34,14 +35,12 @@ Suggests:
     rmarkdown,
     knitr,
     magrittr,
-    iterators,
     audio,
     profvis,
     microbenchmark
 Collate:
   'async-package.R'
   'util.R'
-  'iteror.R'
   'cps.R'
   'signals.R'
   'syntax.R'

NAMESPACE

@@ -2,13 +2,10 @@
 
 S3method(all_indices,environment)
 S3method(all_indices,list)
-S3method(as.list,iteror)
-S3method(as.vector,iteror)
 S3method(compile,coroutine)
 S3method(debugAsync,coroutine)
 S3method(format,channel)
 S3method(format,coroutine)
-S3method(format,iteror)
 S3method(getEntry,coroutine)
 S3method(getPump,async)
 S3method(getPump,generator)
@@ -23,14 +20,7 @@ S3method(getStop,coroutine)
 S3method(getStop,stream)
 S3method(is.channel,channel)
 S3method(is.channel,default)
-S3method(iterators::nextElem,iteror)
-S3method(iteror,"function")
-S3method(iteror,default)
-S3method(iteror,iter)
-S3method(iteror,iteror)
 S3method(make_graph,coroutine)
-S3method(nextOr,funiteror)
-S3method(nextOr,iter)
 S3method(nextThen,channel)
 S3method(print,channel)
 S3method(print,coroutine)
@@ -46,6 +36,7 @@ S3method(summary,stream)
 export(async)
 export(await)
 export(awaitNext)
+export(collect)
 export(collector)
 export(combine)
 export(debugAsync)
@@ -54,9 +45,7 @@ export(gather)
 export(gen)
 export(goto)
 export(graphAsync)
-export(ilimit)
 export(is.channel)
-export(iseq)
 export(iteror)
 export(nextOr)
 export(nextThen)
@@ -71,3 +60,5 @@ import(nseval)
 import(promises)
 import(testthat)
 import(utils)
+importFrom(iterors,iteror)
+importFrom(iterors,nextOr)

NEWS.md

@@ -1,5 +1,9 @@
 # async 0.3.1
 
+* `iteror` and associated functions have been extracted to a new package [`iterors`](http://github.com/crowding/iterors), which includes 
+
+# async 0.3.1
+
 * `debugAsync` now accepts a `trace` argument to control printing of execution traces to console.
   Execution traces now print node addresses. 
 * Various inspection methods `getNode`, `getState`, `getEnv`, `getOrig`  have been consolidated into the method `summary.coroutine`.

R/all_names.R

@@ -164,29 +164,34 @@ visit_function <- function(fn, yield, nonTail=TRUE, forGraph=FALSE) {
                visit_weird_call(expr, inTail, NULL, yield)
              } else if (exists(head, env, inherits=FALSE)) {
                peek <- get0(head, envir=env, ifnotfound=NULL, inherits=FALSE)
-               if (!all(c("cont") %in% names(formals(peek)))) {
+               if (is.function(peek)) {
+                 if (!all(c("cont") %in% names(formals(peek)))) {
+                   visit_ordinary_call(expr, inTail, orig, yield)
+                 } else {
+                   # A trampoline-indirect call! Register both the
+                   # target and the indirect.
+                   handl <- match.call(peek, expr, expand.dots=FALSE,
+                                       envir=as.environment(list(`...`=NULL)))
+                   if ("winding" %in% names(handl)) {
+                     # windup takes TWO function pointers
+                     woundup <- as.call(list(handl$winding))
+                     windup <- TRUE
+                     handl$winding <- NULL
+                     yield(as.character(woundup[[1]]), "wind")
+                     yield(c(list(woundup, expr), orig), "windup")
+                     visit_weird_call(woundup, inTail, c(list(expr), orig), yield)}
+                   trampoline_args <- names(handl) %in% c("cont", "val")
+                   trampolined <- as.call(handl[trampoline_args])
+                   handl <- handl[!trampoline_args]
+                   yield(as.character(trampolined[[1]]), "tramp")
+                   yield(as.character(expr[[1]]), "hand")
+                   yield(c(list(handl, expr), orig), "handler")
+                   yield(c(list(trampolined, expr), orig), "trampoline")
+                   visit_weird_call(handl, FALSE, c(list(expr), orig), yield)
+                 }
+               } else { # not a function?
+                 #This comes up if it's an ieration variable, which you write to...
                  visit_ordinary_call(expr, inTail, orig, yield)
-               } else {
-                 # A trampoline-indirect call! Register both the
-                 # target and the indirect.
-                 handl <- match.call(peek, expr, expand.dots=FALSE,
-                                     envir=as.environment(list(`...`=NULL)))
-                 if ("winding" %in% names(handl)) {
-                   # windup takes TWO function pointers
-                   woundup <- as.call(list(handl$winding))
-                   windup <- TRUE
-                   handl$winding <- NULL
-                   yield(as.character(woundup[[1]]), "wind")
-                   yield(c(list(woundup, expr), orig), "windup")
-                   visit_weird_call(woundup, inTail, c(list(expr), orig), yield)}
-                 trampoline_args <- names(handl) %in% c("cont", "val")
-                 trampolined <- as.call(handl[trampoline_args])
-                 handl <- handl[!trampoline_args]
-                 yield(as.character(trampolined[[1]]), "tramp")
-                 yield(as.character(expr[[1]]), "hand")
-                 yield(c(list(handl, expr), orig), "handler")
-                 yield(c(list(trampolined, expr), orig), "trampoline")
-                 visit_weird_call(handl, FALSE, c(list(expr), orig), yield)
                }
              } else {
                # something that isn't bound in the immediately

R/async-package.R

@@ -50,10 +50,16 @@ trace_ <- function(x) if(getOption("async.verbose")) cat(x)
   options(async.compileLevel=getOption("async.compileLevel") %||% 0,
           async.paranoid=getOption("async.paranoid") %||% FALSE,
           async.verbose=getOption("async.verbose") %||% FALSE,
-          async.destructive=getOption("async.verbose") %||% TRUE,
-          async.sendLater=getOption("async.verbose") %||% TRUE)
+          async.destructive=getOption("async.destructive") %||% TRUE,
+          async.sendLater=getOption("async.sendLater") %||% TRUE)
 }
 
+#' @export
+iterors::nextOr
+
+#' @export
+iterors::iteror
+
 ## usethis namespace: start
 ## usethis namespace: end
 NULL

R/async.R

@@ -179,10 +179,11 @@ make_async <- function(expr, orig = expr, ...,
   nonce <- (function() function() NULL)()
   state <- "pending" #print method uses this
   value <- nonce
+  err <- NULL
   resolve_ <- NULL
   reject_ <- NULL
 
-  node(getState <- function() state)
+  node(getState <- function() list(state=state, err=err))
 
   node(return_ <- function(val) {
     state <<- "resolved"
@@ -192,7 +193,7 @@ make_async <- function(expr, orig = expr, ...,
   })
 
   node(stop_ <- function(val) {
-    value <<- val
+    err <<- val
     state <<- "rejected"
     reject_(val)
     val
@@ -313,10 +314,8 @@ getStartSet.async <- function(x, ...) {
 #' "rejected".
 #' @exportS3Method
 summary.async <- function(object, ...) {
-  c(list(code=object$orig,
-         state=object$state$getState(),
-         node=environment(object$state$pump)$getCont(),
-         envir=environment(object$state$pump)$targetEnv),
+  c(list(code=object$orig),
+    object$state$getState(),
     NextMethod("summary"))
 }
 

R/channel.R

@@ -220,6 +220,7 @@ channel <- function(impl, max_queue=500L, max_awaiting=500L,
                    break
                  }))
         listener$resolve(val)
+        odo <<- odo+1
       }, error=function(err) {
         warning("Unhandled channel error on send: ", err)
       })
@@ -237,18 +238,17 @@ channel <- function(impl, max_queue=500L, max_awaiting=500L,
     send()
   }
 
-  nextOr <- function(or) {
+  nextOr_ <- function(or) {
     #subscribe and return a promise.
     promise(function(resolve, reject) {
       nextThen(resolve, reject, function() reject("StopIteration"))
     })
   }
 
   impl(emit, reject, close)
-  structure(list(nextThen=nextThen, nextOr=nextOr,
-                 formatChannel=formatChannel),
-            class=c("channel", "funiteror", "iteror", "iter"))
-
+  structure(add_class(iteror(nextOr_), "channel"),
+            methods=list(nextThen=nextThen, nextOr=nextOr,
+                         formatChannel=formatChannel))
 }
 
 #' @exportS3Method
@@ -258,7 +258,7 @@ print.channel <- function(x, ...) {
 
 #' @exportS3Method
 format.channel <- function(x, ...) {
-  x$formatChannel(...)
+  attr(x, "methods")$formatChannel(...)
 }
 
 #' Receive values from channels by callback.
@@ -278,7 +278,9 @@ format.channel <- function(x, ...) {
 #'   emitted value. For [subscribe], a function to be called with each
 #'   emitted value until the stream finishes.
 #' @param onError Function to be called if channel stops with an
-#'   error.
+#'   error. Note that if you call nextThen multiple times to register
+#'   multile callbacks, only the first will receive onError; the rest
+#'   will be called with onClose.
 #' @param onClose Function to be called if the channel finishes normally.
 #' @param ... Undocumented.
 #' @export
@@ -293,7 +295,7 @@ nextThen.channel <- function(x,
                                stop("Unhandled channel error: ", err),
                              onClose = function() NULL,
                              ...) {
-  x$nextThen(onNext, onError, onClose, ...)
+  attr(x, "methods")$nextThen(onNext, onError, onClose, ...)
 }
 
 #' @export
@@ -332,7 +334,7 @@ subscribe.channel <- function(x, onNext, onError, onClose, ...) {
 #'
 #' `combine(...)` takes any number of [promise] or [channel]
 #' objects. It awaits each one, and returns a [channel] object
-#' which re-emits every value from its target promises, in whatever
+#' which re-emits every value from its targets, in whatever
 #' order they are received.
 #' @param ... Each argument should be a [promise] or a [channel].
 #' @return a [channel] object.

R/collect.R

@@ -1,38 +1,4 @@
-#' Gather all values emitted by iterators and channels.
-#'
-#' @description These functions help collect values from generators, streams or
-#' other processes into lists. Note that you can also generate a list
-#' of values using [run].
-#'
-#' @description `as.list.iteror` and `as.vector.iteror` convert iterable
-#' objects into vectors of the given mode.
-#'
-#' @exportS3Method as.list iteror
-#' @param ... Undocumented.
-#' @return `as.list.iteror` returns a `[list]`.
-#' @author Peter Meilstrup
-#' @rdname collect
-#' @examples
-#'
-#' as.list(iseq(1,10, by=3))
-as.list.iteror <- function(x, ...) {
-  collect(function(emit) repeat emit(nextOr(x, break)), type=list())
-}
-
-#' @param x An [iteror]
-#' @param mode The mode of the output; taking the same modes as `vector`.
-#' @return `as.vector.iteror(x, mode)` returns a vector of the given mode.
-#' @exportS3Method as.vector iteror
-#' @rdname collect
-#' @examples
-#'
-#' as.vector(gen(for (i in 1:10) if (i %% 3 != 0) yield(i)), "numeric")
-as.vector.iteror <- function(x, mode) {
-  collect(function(emit)
-    repeat(emit(nextOr(x, break))),
-    vector(mode, 0))
-}
-
+#' Collect iterator / channel items into a vector.
 #' @rdname collect
 #' @description `gather` takes a [channel] as argument and returns a
 #'   [promise]. All values emitted by the channel will be collected
@@ -68,6 +34,7 @@ gather <- function(ch, type=list()) {
 #' @rdname collect
 #' @description Method `as.promise.channel` is a synonym for `gather`.
 #' @exportS3Method promises::as.promise channel
+#' @param x a [channel].
 as.promise.channel <- function(x) gather(x)
 
 #' @rdname collect
@@ -100,6 +67,7 @@ as.promise.channel <- function(x) gather(x)
 #' as.list.iteror <- function(it) {
 #'   collect(\(yield) repeat yield(nextOr(it, break)))
 #' }
+#' @export
 collect <- function(fn, type=list()) {
   collector(function(emit, extract) {fn(emit); extract(TRUE)}, type)
 }

R/coroutine.R

@@ -1,4 +1,3 @@
-
 # not-exported methods
 getPump <- function(x) UseMethod("getPump")
 getEntry <- function(x) UseMethod("getEntry")
@@ -80,7 +79,7 @@ getStartSet.coroutine <- function(x) {
        runPump = environment(getPump(x))$runPump,
        base_winding = environment(getPump(x))$base_winding,
        setDebug = environment(getPump(x))$setDebug,
-       getCont = environment(getPump(x))$getCont)
+       getPumpState = environment(getPump(x))$getPumpState)
 }
 
 #' Query / display coroutine properties and state.
@@ -95,11 +94,11 @@ getStartSet.coroutine <- function(x) {
 format.coroutine <- function(x, ...) {
   s <- summary(x)
   a <- deparse(call(class(x)[[1]], s$code), backtick=TRUE)
-  b <- format(s$envir, ...)
+  b <- format(s$targetEnv, ...)
   c <- paste0(c("<", class(x)[[1]], " [",
                 s$state,
-                " at `", s$node, "`",
-                if (s$state=="stopped")
+                " at `", s$cont, "`",
+                if (s$state %in% c("stopped", "rejected"))
                   c(": ", capture.output(print(s$err))),
                 "]>"), collapse="")
   d <- NextMethod()
@@ -136,12 +135,8 @@ coroutine_function <- function(arg, head, ...) {
 #' * `err`: the error object, if the coroutine caught an error.
 #' @export
 summary.coroutine <- function(object, ...) {
+  s <- environment(getPump(object))$getPumpState()
   d <- debugAsync(object)
-  list(
-    node=environment(getPump(object))$getCont(),
-    envir=environment(getPump(object))$targetEnv,
-    err=environment(getPump(object))$err,
-    debugR=d$R,
-    debugInternal=d$internal,
-    trace=d$trace)
+  c(s,
+    debug=d)
 }