<commit_before>// Copyright 2013 <NAME>. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file or
// at http://opensource.org/licenses/BSD-3-Clause.

// Package functional provides functional programming constructs.
package functional

import (
  "bufio"
  "container/heap"
  "errors"
  "io"
  "reflect"
)

var (
  // Done indicates that the end of a Stream has been reached
  Done = errors.New("functional: End of Stream reached.")
  // Filters return Skipped to indicate that the current value should be
  // skipped.
  Skipped = errors.New("functional: Value skipped.")
)

var (
  nilM = nilMapper{}
  nilPieceL = []compositeMapperPiece{{mapper: nilM}}
  nilS = nilStream{}
  trueF = trueFilterer{}
  falseF = falseFilterer{}
)

// Stream is a sequence emitted values.
// Each call to Next() emits the next value in the stream.
// A Stream that emits values of type T is a Stream of T.
type Stream interface {
  // Next emits the next value in this Stream of T.
  // If Next returns nil, the next value is stored at ptr.
  // If Next returns Done, then the end of the Stream has been reached,
  // and the value ptr points to is unspecified. ptr must be a *T.
  // Once Next returns Done, it should continue to return Done.
  Next(ptr interface{}) error
  // Caller calls Close when it is finished with this Stream. 
  // The result of calling Next after Close is unspecified.
  io.Closer
}

// Tuple represents a tuple of values that ReadRows emits
type Tuple interface {
  // Ptrs returns a pointer to each field in the tuple.
  Ptrs() []interface{}
}

// Filterer of T filters values in a Stream of T.
type Filterer interface {
  // Filter returns nil if value ptr points to should be included or Skipped
  // if value should be skipped. ptr must be a *T.
  Filter(ptr interface{}) error
}

// Mapper maps a type T value to a type U value in a Stream.
type Mapper interface {
  // Map does the mapping storing the mapped value at destPtr.
  // If Mapper returns Skipped, then no mapped value is stored at destPtr.
  // Map may return other errors. srcPtr is a *T; destPtr is a *U
  Map(srcPtr interface{}, destPtr interface{}) error
}

// CompositeMapper represents Mappers composed together e.g f(g(x)).
// Programs using CompositeMapper should typically store and pass them as
// values, not pointers. A CompositeMapper can be used by multiple goroutines
// simultaneously if its underlying Mappers can be used by multiple goroutines
// simultaneously. The zero value for CompositeMapper is a Mapper that maps
// nothing (the Map method always returns Skipped).
type CompositeMapper struct {
  _pieces []compositeMapperPiece
}

func (c CompositeMapper) Map(srcPtr interface{}, destPtr interface{}) error {
  return c.Fast().Map(srcPtr, destPtr)
}

// Fast returns a quicker version of this CompositeMapper that cannot be
// used by multiple goroutines simultaneously as if FastCompose were used.
func (c CompositeMapper) Fast() Mapper {
  pieces := c.pieces()
  fastPieces := make([]fastMapperPiece, len(pieces))
  for i := range fastPieces {
    fastPieces[i].setFromCompositePiece(&pieces[i])
  }
  return fastCompositeMapper{fastPieces}
}

func (c CompositeMapper) pieces() []compositeMapperPiece {
  if len(c._pieces) == 0 {
    return nilPieceL
  }
  return c._pieces
}

// Creater of T creates a new, pre-initialized, T and returns a pointer to it.
type Creater func() interface {}

// Copier of T copies the value at src to the value at dest.
// src and dest are of type *T and both point to pre-initialized T.
type Copier func(src, dest interface{})

// Rows represents rows in a database table. Most database API already have
// a type that implements this interface
type Rows interface {
  // Next advances to the next row. Next returns false if there is no next row.
  // Every call to Scan, even the first one, must be preceded by a call to Next.
  Next() bool
  // Reads the values out of the current row. args are pointer types.
  Scan(args ...interface{}) error
}

// NilStream returns a Stream that emits no values. Calling Close on
// returned Stream is a no-op.
func NilStream() Stream {
  return nilS
}

// Map applies f, which maps a type T value to a type U value, to a Stream
// of T producing a new Stream of U. If s is
// (x1, x2, x3, ...), Map returns the Stream (f(x1), f(x2), f(x3), ...).
// If f returns Skipped for a T value, then the corresponding U value is
// left out of the returned stream. ptr is a *T providing storage for
// emitted values from s. If f is a CompositeMapper, Fast() is called on
// it automatically.
// Calling Close on returned Stream closes s.
func Map(f Mapper, s Stream, ptr interface{}) Stream {
  ms, ok := s.(*mapStream)
  if ok {
    return &mapStream{FastCompose(f, ms.mapper, ptr), ms.Stream, ms.ptr}
  }
  cm, ok := f.(CompositeMapper)
  if ok {
    return &mapStream{cm.Fast(), s, ptr}
  }
  return &mapStream{f, s, ptr}
}

// Filter filters values from s, returning a new Stream of T. The returned
// Stream's Next method reports any errors besides Skipped that the Filter
// method of f returns. 
// f is a Filterer of T; s is a Stream of T.
// Calling Close on returned Stream closes s.
func Filter(f Filterer, s Stream) Stream {
  fs, ok := s.(*filterStream)
  if ok {
    return &filterStream{All(fs.filterer, f), fs.Stream}
  }
  return &filterStream{f, s}
}

// Count returns an infinite Stream of int which emits all values beginning
// at 0.
// Calling Close on returned Stream is a no-op.
func Count() Stream {
  return &count{start: 0, step: 1}
}

// CountFrom returns an infinite Stream of int emitting values beginning at
// start and increasing by step.
// Calling Close on returned Stream is a no-op.
func CountFrom(start, step int) Stream {
  return &count{start: start, step: step}
}

// Slice returns a Stream that will emit elements in s starting at index start
// and continuing to but not including index end. Indexes are 0 based. If end
// is negative, it means go to the end of s.
// Calling Close on returned Stream
// closes s.
func Slice(s Stream, start int, end int) Stream {
  return &sliceStream{Stream: s, start: start, end: end}
}

// ReadRows returns the rows in a database table as a Stream of Tuple.
// Calling Close on returned stream does nothing.
func ReadRows(r Rows) Stream {
  return &rowStream{rows: r}
}

// ReadLines returns the lines of text in r separated by either "\n" or "\r\n"
// as a Stream of string. The emitted string types do not contain the
// end of line characters.
// Calling Close on returned Stream does nothing.
func ReadLines(r io.Reader) Stream {
  return &lineStream{bufio: bufio.NewReader(r)}
}

// ReadLinesAndClose works just like ReadLines except that calling Close on
// returned Stream closes r.
func ReadLinesAndClose(r io.ReadCloser) Stream {
  return &closeStream{Stream: ReadLines(r), Closer: r}
}

// NewStreamFromStreamFunc creates a Stream of Streams by repeatedly calling
// f. Calling Close on returned Stream is a no-op.
func NewStreamFromStreamFunc(f func() Stream) Stream {
  return streamFromStreamFunc{f: f}
}

// Deferred(f) is equivalent to Flatten(Slice(NewStreamFromStreamFunc(f), 0, 1))
func Deferred(f func() Stream) Stream {
  return Flatten(Slice(NewStreamFromStreamFunc(f), 0, 1))
}

// Cycle(f) is equivalent to Flatten(NewStreamFromStreamFunc(f))
func Cycle(f func() Stream) Stream {
  return Flatten(NewStreamFromStreamFunc(f))
}

// Merge merges multiple streams that emit their elments in order into a
// single stream that emits all the elements in order.
// Calling Close on returned Stream closes all underlying streams.
// If caller passes a slice to Merge, no copy is made of it.
// Each Stream in streams must emit elements in order according
// to before, and the resulting Stream emits all elements in order according
// to before. before returns true if the T element at lhs comes before the T
// element at rhs. lhs and rhs are *T.
// Merge is draft API, it may change in incompatible ways.
func Merge(
    creater Creater,
    copier Copier, 
    before func(lhs, rhs interface{}) bool,
    streams ...Stream) Stream {
  if len(streams) == 0 {
    return NilStream()
  }
  if len(streams) == 1 {
    return streams[0]
  }
  if copier == nil {
    copier = assignCopier
  }
  h := &streamHeapWithLess{
      streamHeap: make(streamHeap, len(streams)), before: before}
  for i := range streams {
    h.streamHeap[i] = &item{stream: streams[i], current: creater()}
    h.streamHeap[i].pop()
  }
  heap.Init(h)
  return &mergeStream{orig: streams, sh: h, copier: copier}
}<commit_msg>Stabilize draft API<commit_after>// Copyright 2013 <NAME>. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file or
// at http://opensource.org/licenses/BSD-3-Clause.

// Package functional provides functional programming constructs.
package functional

import (
  "bufio"
  "container/heap"
  "errors"
  "io"
  "reflect"
)

var (
  // Done indicates that the end of a Stream has been reached
  Done = errors.New("functional: End of Stream reached.")
  // Filters return Skipped to indicate that the current value should be
  // skipped.
  Skipped = errors.New("functional: Value skipped.")
)

var (
  nilM = nilMapper{}
  nilPieceL = []compositeMapperPiece{{mapper: nilM}}
  nilS = nilStream{}
  trueF = trueFilterer{}
  falseF = falseFilterer{}
)

// Stream is a sequence emitted values.
// Each call to Next() emits the next value in the stream.
// A Stream that emits values of type T is a Stream of T.
type Stream interface {
  // Next emits the next value in this Stream of T.
  // If Next returns nil, the next value is stored at ptr.
  // If Next returns Done, then the end of the Stream has been reached,
  // and the value ptr points to is unspecified. ptr must be a *T.
  // Once Next returns Done, it should continue to return Done.
  Next(ptr interface{}) error
  // Caller calls Close when it is finished with this Stream. 
  // The result of calling Next after Close is unspecified.
  io.Closer
}

// Tuple represents a tuple of values that ReadRows emits
type Tuple interface {
  // Ptrs returns a pointer to each field in the tuple.
  Ptrs() []interface{}
}

// Filterer of T filters values in a Stream of T.
type Filterer interface {
  // Filter returns nil if value ptr points to should be included or Skipped
  // if value should be skipped. ptr must be a *T.
  Filter(ptr interface{}) error
}

// Mapper maps a type T value to a type U value in a Stream.
type Mapper interface {
  // Map does the mapping storing the mapped value at destPtr.
  // If Mapper returns Skipped, then no mapped value is stored at destPtr.
  // Map may return other errors. srcPtr is a *T; destPtr is a *U
  Map(srcPtr interface{}, destPtr interface{}) error
}

// CompositeMapper represents Mappers composed together e.g f(g(x)).
// Programs using CompositeMapper should typically store and pass them as
// values, not pointers. A CompositeMapper can be used by multiple goroutines
// simultaneously if its underlying Mappers can be used by multiple goroutines
// simultaneously. The zero value for CompositeMapper is a Mapper that maps
// nothing (the Map method always returns Skipped).
type CompositeMapper struct {
  _pieces []compositeMapperPiece
}

func (c CompositeMapper) Map(srcPtr interface{}, destPtr interface{}) error {
  return c.Fast().Map(srcPtr, destPtr)
}

// Fast returns a quicker version of this CompositeMapper that cannot be
// used by multiple goroutines simultaneously as if FastCompose were used.
func (c CompositeMapper) Fast() Mapper {
  pieces := c.pieces()
  fastPieces := make([]fastMapperPiece, len(pieces))
  for i := range fastPieces {
    fastPieces[i].setFromCompositePiece(&pieces[i])
  }
  return fastCompositeMapper{fastPieces}
}

func (c CompositeMapper) pieces() []compositeMapperPiece {
  if len(c._pieces) == 0 {
    return nilPieceL
  }
  return c._pieces
}

// Creater of T creates a new, pre-initialized, T and returns a pointer to it.
type Creater func() interface {}

// Copier of T copies the value at src to the value at dest.
// src and dest are of type *T and both point to pre-initialized T.
type Copier func(src, dest interface{})

// Rows represents rows in a database table. Most database API already have
// a type that implements this interface
type Rows interface {
  // Next advances to the next row. Next returns false if there is no next row.
  // Every call to Scan, even the first one, must be preceded by a call to Next.
  Next() bool
  // Reads the values out of the current row. args are pointer types.
  Scan(args ...interface{}) error
}

// NilStream returns a Stream that emits no values. Calling Close on
// returned Stream is a no-op.
func NilStream() Stream {
  return nilS
}

// Map applies f, which maps a type T value to a type U value, to a Stream
// of T producing a new Stream of U. If s is
// (x1, x2, x3, ...), Map returns the Stream (f(x1), f(x2), f(x3), ...).
// If f returns Skipped for a T value, then the corresponding U value is
// left out of the returned stream. ptr is a *T providing storage for
// emitted values from s. If f is a CompositeMapper, Fast() is called on
// it automatically.
// Calling Close on returned Stream closes s.
func Map(f Mapper, s Stream, ptr interface{}) Stream {
  ms, ok := s.(*mapStream)
  if ok {
    return &mapStream{FastCompose(f, ms.mapper, ptr), ms.Stream, ms.ptr}
  }
  cm, ok := f.(CompositeMapper)
  if ok {
    return &mapStream{cm.Fast(), s, ptr}
  }
  return &mapStream{f, s, ptr}
}

// Filter filters values from s, returning a new Stream of T. The returned
// Stream's Next method reports any errors besides Skipped that the Filter
// method of f returns. 
// f is a Filterer of T; s is a Stream of T.
// Calling Close on returned Stream closes s.
func Filter(f Filterer, s Stream) Stream {
  fs, ok := s.(*filterStream)
  if ok {
    return &filterStream{All(fs.filterer, f), fs.Stream}
  }
  return &filterStream{f, s}
}

// Count returns an infinite Stream of int which emits all values beginning
// at 0.
// Calling Close on returned Stream is a no-op.
func Count() Stream {
  return &count{start: 0, step: 1}
}

// CountFrom returns an infinite Stream of int emitting values beginning at
// start and increasing by step.
// Calling Close on returned Stream is a no-op.
func CountFrom(start, step int) Stream {
  return &count{start: start, step: step}
}

// Slice returns a Stream that will emit elements in s starting at index start
// and continuing to but not including index end. Indexes are 0 based. If end
// is negative, it means go to the end of s.
// Calling Close on returned Stream
// closes s.
func Slice(s Stream, start int, end int) Stream {
  return &sliceStream{Stream: s, start: start, end: end}
}

// ReadRows returns the rows in a database table as a Stream of Tuple.
// Calling Close on returned stream does nothing.
func ReadRows(r Rows) Stream {
  return &rowStream{rows: r}
}

// ReadLines returns the lines of text in r separated by either "\n" or "\r\n"
// as a Stream of string. The emitted string types do not contain the
// end of line characters.
// Calling Close on returned Stream does nothing.
func ReadLines(r io.Reader) Stream {
  return &lineStream{bufio: bufio.NewReader(r)}
}

// ReadLinesAndClose works just like ReadLines except that calling Close on
// returned Stream closes r.
func ReadLinesAndClose(r io.ReadCloser) Stream {
  return &closeStream{Stream: ReadLines(r), Closer: r}
}

// NewStreamFromStreamFunc creates a Stream of Streams by repeatedly calling
// f. Calling Close on returned Stream is a no-op.
func NewStreamFromStreamFunc(f func() Stream) Stream {
  return streamFromStreamFunc{f: f}
}

// Deferred(f) is equivalent to Flatten(Slice(NewStreamFromStreamFunc(f), 0, 1))
func Deferred(f func() Stream) Stream {
  return Flatten(Slice(NewStreamFromStreamFunc(f), 0, 1))
}

// Cycle(f) is equivalent to Flatten(NewStreamFromStreamFunc(f))
func Cycle(f func() Stream) Stream {
  return Flatten(NewStreamFromStreamFunc(f))
}

// Merge merges multiple streams that emit their elments in order into a
// single stream that emits all the elements in order.
// Calling Close on returned Stream closes all underlying streams.
// If caller passes a slice to Merge, no copy is made of it.
// Each Stream in streams must emit elements in order according
// to before, and the resulting Stream emits all elements in order according
// to before. before returns true if the T element at lhs comes before the T
// element at rhs. lhs and rhs are *T.
func Merge(
    creater Creater,
    copier Copier, 
    before func(lhs, rhs interface{}) bool,
    streams ...Stream) Stream {
  if len(streams) == 0 {
    return NilStream()
  }
  if len(streams) == 1 {
    return streams[0]
  }
  if copier == nil {
    copier = assignCopier
  }
  h := &streamHeapWithLess{
      streamHeap: make(streamHeap, len(streams)), before: before}
  for i := range streams {
    h.streamHeap[i] = &item{stream: streams[i], current: creater()}
    h.streamHeap[i].pop()
  }
  heap.Init(h)
  return &mergeStream{orig: streams, sh: h, copier: copier}
}
{'fraction_non_alphanumeric': 0.09261518796118762, 'fraction_numerical': 0.002672709313810935, 'mean_word_length': 3.457912457912458, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7285970', 'n_tokens_mistral': 5141, 'n_tokens_neox': 5048, 'n_words': 2380}