@@ -1,7 +1,7 @@

 Package: Streamer

 Type: Package

 Title: Enabling stream processing of large files

-Version: 1.3.4

+Version: 1.3.5

 Date: 2010-10-13

 Author: Martin Morgan, Nishant Gopalakrishnan

 Maintainer: Martin Morgan <mtmorgan@fhcrc.org>

@@ -15,8 +15,8 @@ Description: Large data files can be difficult to work with in R,

   package?Streamer for details.

 License: Artistic-2.0

 LazyLoad: yes

-Imports: methods, graph, parallel

-Suggests: BiocGenerics, RUnit, Rsamtools (>= 1.5.53)

+Imports: methods, graph, parallel, BiocGenerics

+Suggests: RUnit, Rsamtools (>= 1.5.53)

 biocViews: Infrastructure, DataImport

 Collate: 

   generics.R OldClass.R

@@ -2,6 +2,8 @@ useDynLib(Streamer, .registration=TRUE)

 import(methods)

+importFrom(BiocGenerics, lapply, sapply)

+

 importFrom(parallel, mcparallel, mccollect)

 importFrom(graph, graphBAM)

@@ -10,6 +12,8 @@ importMethodsFrom(graph, degree)

 exportPattern("^[^\\.]")

+exportMethods(lapply, sapply)

+

 exportClasses(Streamer,

               Producer, RawInput, Seq, 

               Consumer, RawToChar, Rev, Team, Downsample,

@@ -8,3 +8,34 @@ setMethod(stream, "Producer",

     else

         do.call(stream, c(rev(list(..., verbose=verbose)), list(x)))

 })

+

+setMethod(lapply, "Producer",

+    function(X, FUN, ..., .env=parent.frame())

+{

+    FUN <- match.fun(FUN)

+    fun <- function(yield) {

+        y <- tryCatch(yield(), error=function(err) {

+            stop("'lapply,Producer-method' yield() failed: ",

+                 conditionMessage(err))

+        })

+        if (!length(y))

+            return(y)

+        tryCatch(eval(FUN(y, ...), .env), error=function(err) {

+            stop("'lapply,Producer-method' FUN() failed: ",

+                 conditionMessage(err))

+        })

+    }

+    ## avoid S4 dispatch on yield(X)

+    .Call(.lapply_Producer, fun, X$yield, environment())

+})

+

+setMethod(sapply, "Producer",

+    function(X, FUN, ..., simplify=TRUE, USE.NAMES=TRUE)

+{

+    FUN <- match.fun(FUN)

+    .env <- parent.frame()

+    answer <- lapply(X = X, FUN = FUN, ..., .env=.env)

+    if (!identical(simplify, FALSE) && length(answer))

+        simplify2array(answer, higher = (simplify == "array"))

+    else answer

+})

new file mode 100644

@@ -0,0 +1,30 @@

+test_lapply_Producer <- function()

+{

+    vals <- split(1:47, rep(1:7, each=7, length.out=47))

+    exp <- unname(lapply(vals, mean))

+

+    ## function

+    obs <- lapply(Seq(to=47, length.out=7), mean)

+    checkIdentical(exp, obs)

+

+    ## anonymous function

+    obs <- lapply(Seq(to=47, length.out=7), function(x) mean(x))

+    checkIdentical(exp, obs)

+

+    ## ... args

+    obs <- lapply(Seq(to=47, length.out=7), function(x, z) mean(z), z=1:10)

+    checkIdentical(mean(1:10), unique(unlist(obs)))

+

+    ## env

+    ZZZ <- 1:10

+    res <- lapply(Seq(to=47, length.out=7), function(x) mean(ZZZ))

+    checkIdentical(mean(ZZZ), unique(unlist(res)))

+

+    ## error

+    fun <- function(x) if (x == 3) stop("x: ", x) else x

+    checkException(lapply(Seq(to=5), fun), silent=TRUE)

+

+    ## trigger re-allocation

+    ## res <- lapply(Seq(to=4096*4),

+    ##               function(x) { if (x %% 1000 == 0) message(x); x })

+}

@@ -2,7 +2,9 @@

 \Rdversion{1.1}

 \docType{class}

 \alias{Producer-class}

-\alias{show,Producer-class}

+\alias{lapply,Producer-method}

+\alias{sapply,Producer-method}

+\alias{show,Producer-method}

 \title{Class "Producer"}

@@ -18,8 +20,34 @@

 }

+\usage{

+\S4method{lapply}{Producer}(X, FUN, ..., .env=parent.frame())

+\S4method{sapply}{Producer}(X, FUN, ..., simplify=TRUE, USE.NAMES=TRUE)

+}

+

+\arguments{

+  \item{X}{An instance of class \code{Producer}}

+

+  \item{FUN}{A function to be applied to each successful \code{yield()}

+    of \code{X}.}

+

+  \item{...}{Additional arguments to \code{FUN}.}

+

+  \item{simplify}{See \code{?base::sapply}.}

+

+  \item{USE.NAMES}{See \code{?base::sapply} but note that names do not

+    usually make sense for instances of class \code{Producer}.}

+

+  \item{.env}{For internal use}

+}

+

 \section{Methods}{

-  Methods defined on this class include:

+

+  \code{lapply} and \code{sapply} apply \code{FUN} to each result

+  applied to \code{yield()}. Infite producers will of course exhaust

+  memory.

+

+  Inherited methods defined on this class include:

   \describe{

     \item{stream}{\code{signature(x = "Producer", ...)}: see

       \code{?stream}.}

@@ -40,15 +68,18 @@

 \section{Class-Based Methods}{

   \describe{

-    The \code{Producer} class inherits the methods \code{initialize}, \code{msg},

-    \code{reset}, \code{status} and \code{yield} from the \code{Streamer}

-    virtual class. Please refer to the \code{\link{Streamer}} class for more

-    details.

+

+    The \code{Producer} class inherits the methods \code{initialize},

+    \code{msg}, \code{reset}, \code{status} and \code{yield} from the

+    \code{Streamer} virtual class. Please refer to the

+    \code{\link{Streamer}} class for more details.

-    Derived classes should implement an appropriate \code{initialize} method to

-    initialize the fields of the derived class. Additionally, a \code{yield} method

-    should be implemented to return the contents of the current stream. The

-    default method for the base virtual \code{Streamer} class returns a \code{list()} 

+    Derived classes should implement an appropriate \code{initialize}

+    method to initialize the fields of the derived class. Additionally,

+    a \code{yield} method should be implemented to return the contents

+    of the current stream. The default method for the base virtual

+    \code{Streamer} class returns a \code{list()}

+

   }

 }

@@ -61,6 +92,13 @@

 }

-\examples{showClass("Producer")}

+\examples{

+showClass("Producer")

+showMethods(class="Producer", where="package:Streamer")

+

+sapply(Seq(to=47, length.out=7), function(elt) {

+    c(n = length(elt), xbar = mean(elt))

+})

+}

 \keyword{classes}

@@ -1,10 +1,13 @@

 #include <R_ext/Rdynload.h>

 #include "raw_input.h"

+#include "lapply.h"

 static const R_CallMethodDef callMethods[] = {

   /* raw_parse */

   {".raw_parse_count_records", (DL_FUNC) &raw_parse_count_records, 2},

   {".raw_parse", (DL_FUNC) &raw_parse, 3},

+  /* lapply */

+  {".lapply_Producer", (DL_FUNC) &lapply_Producer, 3},

   {NULL, NULL, 0}

 };

new file mode 100644

@@ -0,0 +1,35 @@

+#include "lapply.h"

+

+SEXP

+lapply_Producer(SEXP fun, SEXP X, SEXP rho)

+{

+    SEXP call1, ans1, ans;

+    int iter = 0;

+    R_len_t curr_size = 4096;

+    PROTECT_INDEX px;

+

+    /* FIXME: type checks */

+

+    PROTECT(call1 = lang2(fun, X));

+    PROTECT_WITH_INDEX(ans = Rf_allocVector(VECSXP, curr_size), &px);

+    while (1) {

+	PROTECT(ans1 = Rf_eval(call1, rho));

+	if (0 == Rf_length(ans1)) {

+	    UNPROTECT(1);

+	    break;

+	}

+	if (iter == curr_size) {

+	    if (curr_size == R_LEN_T_MAX)

+		Rf_error("%s cannot create %d-element vector",

+			 "'lapply,Producer-method'", curr_size);

+	    curr_size *= 1.6;

+	    if (curr_size > R_LEN_T_MAX)

+		curr_size = R_LEN_T_MAX;

+	    REPROTECT(ans = Rf_lengthgets(ans, curr_size), px);

+	}

+	SET_VECTOR_ELT(ans, iter++, ans1);

+	UNPROTECT(1);

+    }

+    UNPROTECT(2);

+    return Rf_lengthgets(ans, iter);

+}

new file mode 100644

@@ -0,0 +1,10 @@

+#ifndef LAPPLY_H_

+#define LAPPLY_H

+

+#include <Rdefines.h>

+

+SEXP

+lapply_Producer(SEXP fun, SEXP X, SEXP rho);

+

+#endif

+