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

 Package: Streamer

 Type: Package

 Title: Enabling stream processing of large files

-Version: 1.3.10

+Version: 1.3.11

 Author: Martin Morgan, Nishant Gopalakrishnan

 Maintainer: Martin Morgan <mtmorgan@fhcrc.org>

 Description: Large data files can be difficult to work with in R,

@@ -22,7 +22,7 @@ Collate:

   Streamer-class.R Producer-class.R Consumer-class.R Stream-class.R

   ConnectionProducer-classes.R RawInput-class.R Seq-class.R

   Downsample-class.R FunctionProducerConsumer-classes.R

-  ParallelParam-classes.R Team-class.R Team-methods.R

+  ParallelParam-classes.R Team-class.R Team-methods.R Reducer-class.R

   Utility-classes.R ParallelConnector-class.R TConnector-class.R

   YConnector-class.R

   lapply-methods.R stream-methods.R

@@ -19,6 +19,6 @@ exportMethods(lapply, sapply)

 exportClasses(Streamer,

               Producer, RawInput, Seq, 

-              Consumer, RawToChar, Rev, Team, Downsample,

+              Consumer, RawToChar, Rev, Reducer, Team, Downsample,

               Stream, ConnectionProducer, ParallelConnector,

               TConnector, YConnector)

new file mode 100644

@@ -0,0 +1,69 @@

+.Reducer <- setRefClass("Reducer",

+    fields = list(

+      FUN = "function",

+      init = "ANY",

+      yieldNth = "integer",

+      .hasInit = "logical",

+      .curr = "ANY",

+      .ith = "integer",

+      .done = "logical"),

+    contains = "Consumer",

+    methods = list(

+

+      .nth = function() length(.buf),

+

+      done = function() .done,

+

+      reset = function() {

+          callSuper()

+          .self$.curr <- NULL

+          .self$.ith <- 0L

+          .self$.done <- FALSE

+      },

+

+      yield = function() {

+          if (verbose)

+              message("Reducer$yield")

+          if (done())

+              return(.curr)

+          repeat {

+              val <- callSuper()

+              if (!length(val)) {

+                  ret <- if (.ith != 0) .self$.curr else val

+                  .self$.done <- TRUE

+                  .self$.curr <- val

+                  return(ret)

+              }

+              .self$.curr <- 

+                  if (.ith == 0L)

+                      if  (.hasInit)

+                          FUN(init, val)

+                      else

+                          val

+                  else

+                      FUN(.curr, val)

+              .self$.ith <- .ith + 1L

+              if (!is.na(yieldNth) && (.ith %% yieldNth) == 0) {

+                  .self$.ith <- 0L

+                  break

+              }

+          }

+          .curr

+      },

+

+      show = function() {

+          cat("yieldNth: ", yieldNth, " (current: ", .ith, ")\n",

+              sep="")

+          cat("has init:", .hasInit, "\n")

+      }))

+

+Reducer <-

+    function(FUN, init,  ..., yieldNth=NA_integer_)

+{

+    FUN <- match.fun(FUN)

+    hasInit <- !missing(init)

+    if (!hasInit)

+        init <- NULL

+    .Reducer$new(FUN=FUN, init=init, yieldNth = as.integer(yieldNth),

+                 ..., .done = FALSE, .ith = 0L, .hasInit = hasInit)

+}

new file mode 100644

@@ -0,0 +1,22 @@

+test_Reducer <- function()

+{

+    s <- stream(Seq(to=10), Reducer("+"))

+    checkIdentical(sum(1:10), yield(s))

+    checkIdentical(numeric(), yield(s))

+    s <- stream(Seq(to=10, yieldSize=5L), Reducer("+"))

+    checkIdentical(1:5 + 6:10, yield(s))

+    checkIdentical(numeric(), yield(s))

+    ## init

+    s <- stream(Seq(to=10), Reducer("+", init=10L))

+    checkIdentical(10L + sum(1:10), yield(s))

+    checkIdentical(numeric(), yield(s))

+    ## yieldNth

+    s <- stream(Seq(to=10), Reducer("+", yieldNth=5))

+    checkIdentical(c(sum(1:5), sum(6:10)), sapply(s, c))

+    checkIdentical(numeric(), yield(s))

+    ## reset

+    s <- stream(Seq(to=10), Reducer("+", init=10L)); yield(s)

+    reset(s)

+    checkIdentical(10L + sum(1:10), yield(s))

+    checkIdentical(numeric(), yield(s))

+}

new file mode 100644

@@ -0,0 +1,67 @@

+\name{Reducer-class}

+\Rdversion{1.1}

+\docType{class}

+\alias{Reducer}

+\alias{Reducer-class}

+

+\title{Consumer class to combine successive records}

+

+\description{

+  

+  A \code{\linkS4class{Consumer}}-class to reduce N successive records

+  into a single yield.

+

+}

+

+\usage{

+Reducer(FUN, init, ..., yieldNth = NA_integer_)

+}

+

+\arguments{

+

+  \item{FUN}{A function of two arguments, where the first argument is

+    the result of the previous reduction (or \code{init}, if specified,

+    for the first record) and the second argument is the current

+    record.}

+

+  \item{init}{An optional initial value to initiate the reduction. When

+    present, \code{init} is used to initial each yield.}

+

+  \item{...}{Additional arguments, passed to the \code{$new} method of

+    the underlying reference class. Currently unused.}

+

+  \item{yieldNth}{A positive integer indicating how many upstream yields

+    are combined before the Reducer yields. A value of

+    \code{NA_integer_} indicates reduction of all records in the input

+    stream.}

+

+}

+  

+\section{Methods}{See \code{\link{Consumer}} Methods.}

+

+\section{Internal Class Fields and Methods}{

+

+  Internal fields of this class are are described with, e.g.,

+  \code{getRefClass("Reducer")$fields}.

+

+  Internal methods of this class are described with

+  \code{getRefClass("Reducer")$methods()} and

+  \code{getRefClass("Reducer")$help()}.

+

+}

+

+\author{Martin Morgan \url{mtmorgan@fhcrc.org}}

+

+\seealso{\code{\link{stream}}}

+

+\examples{

+s <- stream(Seq(to=10), Reducer("+"))

+yield(s)     ## sum(1:10)

+s <- stream(Seq(to=10), Reducer("+", yieldNth=5))

+yield(s)     ## sum(1:5)

+yield(s)     ## sum(6:10)

+s <- stream(Seq(to=10), Reducer("+", init=10, yieldNth=5))

+sapply(s, c) ## 10 + c(sum(1:5), sum(6:10))

+}

+

+\keyword{classes}