Stream._extend 

Append

Creates a new Stream with given element appended to this Stream

(1 to 5).all.append(99).append(100).lp

// Output
~(1, 2, 3, 4, 5, 99, 100)

Note. append has operator alias '+'

Append all

Creates a new Stream with that Stream appended to this Stream

('1' to '9').all.appendAll('a' to 'd').appendAll('A' to 'D').lp

// Output
~(1, 2, 3, 4, 5, 6, 7, 8, 9, a, b, c, d, A, B, C, D)

Filter and converter

Only lets elements for which given PartialFinction is defined

The elements are converted to returned type

def all = ~~[Any] + "ABC" + 1 + 22.0 + "DE" + 333F + "F"

all.lp                                     // Prints ~(ABC, 1, 22.0, DE, 333.0, F)

all.collect{
    case s: String if (s.length > 1) => s
}.lp                                       // Prints ~(ABC, DE)

Note. collect always requires double evaluation for each element, so letMap is preferred in many cases

Default element

If this Stream is empty, the given element will be appended

Otherwise this Stream elements will not change

(1 until 1).all.default(99).lp // Prints: ~(99)

(1 until 5).all.default(99).lp // Prints: ~(1, 2, 3, 4)

Reversed filter

Discards all the elements satisfying the filter

(1 to 10).all.drop(_ % 2 == 0).lp // Prints: ~(1, 3, 5, 7, 9)

Group reversed filter

Discards all elements equal to the found in that stream

val idx = (1 to 5).all.flatMap(i => Seq(i, i, i)).to[Idx]

idx.all.lp                         // Prints: ~(1, 1, 1, 2, 2, 2, 3, 3, 3, 4, 4, 4, 5, 5, 5)

idx.all.dropAll(3 to 7).lp         // Prints: ~(1, 1, 1, 2, 2, 2)

idx.all.dropAll(List(2, 5, 11)).lp // Prints: ~(1, 1, 1, 3, 3, 3, 4, 4, 4)

Note. The operation is very efficient if streams are sorted

Property group reversed filter

Discards elements, which produce property value equal to the found in that stream

~~("abc", "d", "e", "", "fg", "hijk").dropAllBy(_.length, Seq(0, 3)).lp

// Output
~(d, e, fg, hijk)

Discarding everything

With a single test lets to drop the entire pipeline in favor of void instance

Note: This can also be done with 'if-else' outside the pipeline, however it proved to be useful with really long statements

(1 to 10).all.dropEverythingIf(true).lp  // Prints ~()

Sequence tail reversed filter

Discards the last consecutive number of elements

(1 to 10).all.dropLast(3).lp // Prints ~(1, 2, 3, 4, 5, 6, 7)

Sequence head reversed filter

Discards the first consecutive number of elements

(1 to 10).all.dropNext(3).lp // Prints ~(4, 5, 6, 7, 8, 9, 10)

Sequence range reversed filter

Discards elements, which fall within the specified range

(1 to 10).all.dropRange(3 to 7).lp  // Prints ~(1, 2, 3, 9, 10)

Sequence head reversed filter

Discards first consecutive elements satisfying the filter

Note, everything starting from the first non compliant element will be allowed (including later compliant elements)

def all = (1 to 5).all +~ (1 to 5)

all.lp                     // Prints ~(1, 2, 3, 4, 5, 1, 2, 3, 4, 5)

all.dropWhile(_ <= 3).lp   // Prints ~(4, 5, 1, 2, 3, 4, 5)

Same as let

filter operation is left for compatibility

let is the canonical filter

Map multiplier

For every existing element, a mapped stream of elements is inserted into the pipeline

Note. The mapping can return an empty stream, in which case total number of elements might even be reduced

~~(1, 2, 3).flatMap(i => Seq(i * 10, i * 100, i * 1000)).lp

// Output
~(10, 100, 1000, 20, 200, 2000, 30, 300, 3000)

Converts a stream of streams into a flat stream

val stream: ~[~[Char]] = ~~(
  'a' to 'd',
  List('x', 'y', 'z'),
  Vector('v', 'e', 'c', 't', 'o', 'r'))

stream.flatten.lp // Prints: ~(a, b, c, d, x, y, z, v, e, c, t, o, r)

Group by test

Puts elements in the same group based on a function test for every two consecutive elements

// Putting Ints into groups of 3

(0 to 20).all.group(_ / 3 == _ / 3).tp

// Output
---------------
?
---------------
~(0, 1, 2)
~(3, 4, 5)
~(6, 7, 8)
~(9, 10, 11)
~(12, 13, 14)
~(15, 16, 17)
~(18, 19, 20)
---------------

Simple grouping

Puts consecutive elements in the same group if they are equal

Note: Non consecutive equal elements will end up in different groups. Prior ordering might be needed

def all =  ~~(1, 2, 3).flatMap(i => Seq(i, i, i))

all.lp         // Prints ~(1, 1, 1, 2, 2, 2, 3, 3, 3)

all.group.tp   // Prints  ------------
                          ?
                          ------------
                          ~(1, 1, 1)
                          ~(2, 2, 2)
                          ~(3, 3, 3)
                          ------------

Grouping on properties

Puts consecutive elements in the same group if all the specified properties are equal

When properties change, a new group is started

 ('#' to '|').all.groupBy(_.isLetter, _.isDigit).tp

// Output
---------------------------------------------------------------------------------
?
---------------------------------------------------------------------------------
~(#, $, %, &, ', (, ), *, +, ,, -, ., /)
~(0, 1, 2, 3, 4, 5, 6, 7, 8, 9)
~(:, ;, <, =, >, ?, @)
~(A, B, C, D, E, F, G, H, I, J, K, L, M, N, O, P, Q, R, S, T, U, V, W, X, Y, Z)
~([, \, ], ^, _, `)
~(a, b, c, d, e, f, g, h, i, j, k, l, m, n, o, p, q, r, s, t, u, v, w, x, y, z)
~({, |)
---------------------------------------------------------------------------------

Fixed size groups

Puts consecutive elements into fixed size groups

('a' to 'z').all.groupBySize(8).tp

// Output
-------------------------
?
-------------------------
~(a, b, c, d, e, f, g, h)
~(i, j, k, l, m, n, o, p)
~(q, r, s, t, u, v, w, x)
~(y, z)
-------------------------

Insert at

Creates a new Stream with given element inserted into this Stream at given index

If index is out of range, the element is prepended or appended

('a' to 'd').all.insert(2, 'X').lp

 // Output
 ~(a, b, X, c, d)

Insert stream at

Creates a new Stream with that Stream inserted into this Stream at given index

If index is out of range, the elements are prepended or appended

Main filter

Only lets elements satisfying the filter

let is the main filtering method ('filter' is supported for compatibility only)

(1 to 10).all.let(_ % 2 == 0).lp // Prints: ~(2, 4, 6, 8, 10)

Group filter

Only lets elements equal to the found in that stream

val idx = (1 to 5).all.flatMap(i => Seq(i,i,i)).to[Idx]

idx.all.lp                        // Prints: ~(1, 1, 1, 2, 2, 2, 3, 3, 3, 4, 4, 4, 5, 5, 5)

idx.all.letAll(3 to 7).lp         // Prints: ~(3, 3, 3, 4, 4, 4, 5, 5, 5)

idx.all.letAll(List(2, 5, 11)).lp // Prints: ~(2, 2, 2, 5, 5, 5)

Note. The operation is very efficient if streams are sorted

Property group filter

Only lets elements, which produce property value equal to the found in that stream

~~("abc", "d", "e", "", "fg", "hijk").letAllBy(_.length, Seq(0, 1, 2)).lp

// Output
~(d, e, , fg)

Indexed filter

Only lets elements satisfying the filter

('a' to 'z').all.letIdx((i, v) => i >= 2 && i <= 7, 1).lp

// Output
~(b, c, d, e, f, g)

Sequence tail filter

Only lets specified number of last elements

(1 to 10).all.letLast(3).lp  // Prints  ~(8, 9, 10)

Filter and converter

Only lets elements for which given function returns non empty Opt

The elements are converted to the new type

def all = "ABC" ~+ "1" + "22" + "D" + "333" + "E"

all.letMap(v => if (v.length < 2) \/ else v).lp // Prints: ~(ABC, 22, 333)

all.letMap({
  case s if (s.length >= 2) => s
  case _                    => \/
}).lp                                           // Prints: ~(ABC, 22, 333)

Note: letMap is often a faster alternative to collect with PartialFunction, because it is evaluated just once for each element

Sequence head filter

Only lets specified number of next elements

Note: This might look like take, but it is more efficient (take immediately iterates the elements creating a new stream)

(1 to 10).all.letNext(3).lp  // Prints  ~(1, 2, 3)

Sequence range filter

Only lets elements within given index range

('a' to 'z').all.letRange(1 to 7).lp

// Output
~(b, c, d, e, f, g, h)

Filter and type converter

Only lets elements, which are instances of the given type

Note, the result is mapped to the specified type

def all = ~~[Any] + "1" + 2 + 3.0 + 4l + "5"

all.lp                  // Prints ~(1, 2, 3.0, 4, 5)

all.letType[String].lp  // Prints ~(1, 5)

Sequence head filter

Only lets first consecutive elements satisfying the filter

Note, everything starting from the first non compliant element will be discarded (including later compliant elements)

def all = (1 to 5).all +~ (1 to 5)

all.lp                     // Prints ~(1, 2, 3, 4, 5, 1, 2, 3, 4, 5)

all.letWhile(_ <= 3).lp    // Prints ~(1, 2, 3)

Element conversion

Converts every element in the pipeline with given function

(1 to 5).all.map( _ / 2.0).lp  // Prints ~(0.5, 1.0, 1.5, 2.0, 2.5)

Element cast

Casts the pipeline elements into the type specified

def all: ~[Any] = ~~[Any] + 1 + 2 + 3

all.mapCast[Int].lp   // Prints: ~(1, 2, 3)

Note. The cast will not fail immediately, and if there is a problem, it will come up later during pumping action

Indexed element conversion

Converts every element in the pipeline with given function

('A' to 'G').all.mapIdx(_ + "=" + _, 1).lp  // Prints ~(1=A, 2=B, 3=C, 4=D, 5=E, 6=F, 7=G)

Loose indexing optimizations

Many streams from indexed sources (like IndexedSeq, Array, Vector, etc) are special. They know their size and can read elements without iteration. They can optimize operations like take, dropNext, letLast, and many others

noindex turns this privileged Streams into regular and it is needed occasionally for debugging and testing

Loose size information

Many streams return sizeOpt, knowing their current size

nosize drops sizing information, so some optimizations will not be available

This is primarily for testing and debgging

Parallel

Returns Stream.Flow with parallel execution

Each consecutive element will be sent to a new thread for processing

(1 to 5).all.parallel.map("Value: " + _ + "\t" + Thread.currentThread.getName).peek(println).drain

// Output
Value: 1    ForkJoinPool.commonPool-worker-9
Value: 3    ForkJoinPool.commonPool-worker-11
Value: 2    main
Value: 4    ForkJoinPool.commonPool-worker-2
Value: 5    ForkJoinPool.commonPool-worker-4

Conditionally parallel

Switches to parallel execution if boolean parameter == true

Returns Stream.Flow, which could be implemented as sequential Stream or parallel Stream.Flow

(1 to 50).all.parallelIf(true).isParallel   // Returns true

(1 to 50).all.parallelIf(false).isParallep  // Returns false

Conditionally parallel

Switches to parallel execution if number of elements exceeds threshold

Returns Stream.Flow, which could be implemented as sequential Stream or parallel Stream.Flow

(1 to 50).all.parallelIfOver(100).isParallel   // Returns false

(1 to 200).all.parallelIfOver(100).isParallel  // Returns true

Multi-filter grouping

All stream elements are grouped by specified filters

Filters are applied in sequence, thus if an element is accepted into a group, it will not be evaluated by the rest of the filters

If there are elements, which are left without a group, one extra group is created

// Age groups
(1 to 80).all.partition(_ <= 12, 13 to 19, _ < 30, 30 to 40, _ < 50, _ < 65).tp

// Output
-------------------------------------------------------------------
?
-------------------------------------------------------------------
~(1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12)
~(13, 14, 15, 16, 17, 18, 19)
~(20, 21, 22, 23, 24, 25, 26, 27, 28, 29)
~(30, 31, 32, 33, 34, 35, 36, 37, 38, 39, 40)
~(41, 42, 43, 44, 45, 46, 47, 48, 49)
~(50, 51, 52, 53, 54, 55, 56, 57, 58, 59, 60, 61, 62, 63, 64)
~(65, 66, 67, 68, 69, 70, 71, 72, 73, 74, 75, 76, 77, 78, 79, 80)
-------------------------------------------------------------------

Element access

Provides access to passing pipeline elements

This method does not change pipeline type or composition in any way

('A' to 'C').all.peek(v => println("Passing: " + v)).count  // Returns 3

// Output
Passing: A
Passing: B
Passing: C

Indexed element access

Provides access to passing pipeline elements with their index in sequence

('A' to 'C').all.peekIdx((i, v) => println("Peek " + i + " = " + v), 1).count // Returns 3

// Output
Peek 1 = A
Peek 2 = B
Peek 3 = C

Adds preview capabilities

Returns Interface.Preview, which allows to pre-load and inspect elements, even before they go through Stream

Reverse order

Re-arranges elements is reverse order

Note: this operation is not suitable for large streams

Reverse order in segments

Reverses order of elements within segments of fixed size

Use Case: Predefined Shuffle

For testing it is often needed to get elements in random order. However it cannot be completely random, if we want to replicate the bug

reverseSized can shuffle elements in a predefined order, given same group size

(1 to 15).all.reverseSized(5).lp // Prints ~(5, 4, 3, 2, 1, 10, 9, 8, 7, 6, 15, 14, 13, 12, 11)

Restore Flow to Stream

Restores potentially parallel Flow back to Stream

If this is already a Stream, the operation is instant, returning this

Otherwise the operation is quite expensive

In many cases it is advisable to consume pipeline as Flow instead of converting to Stream

val (count, millis) = (1 to 1000).all
  .parallel                           // Switching to parallel Stream.Flow
  .peek(_ => Thread.sleep(1))         // Expensive operation
  .sequential                         // Back to Stream
  .countAndMillis

println("Count = " + count + ", done in " + millis / 1000F + " secs")

// Output
Count = 1000, done in 0.224 secs

// Note: We have 1000 elements each pausing for 1 millis.
//       Without parallel processing total time would be over 1 second

Randomize order

Re-arranges elements is random order

Note: this operation requires full buffering and is not suitable for large streams

Sliding group view

Example: group size 3 with step 1

('a' to 'g').all.sliding(3).tp

// Output
----------
?
----------
~(a, b, c)
~(b, c, d)
~(c, d, e)
~(d, e, f)
~(e, f, g)
----------

Example: group size 4 with step 2

('a' to 'g').all.sliding(4,2).tp

// Output
-------------
?
-------------
~(a, b, c, d)
~(c, d, e, f)
~(e, f, g)
-------------

Sort

Sorts stream elements with provided Ordering

~~(5, 1, 4, 2, 3).sort.lp  // Prints ~(1, 2, 3, 4, 5)

Sort by three properties

Sorts pipeline on first property, then if indeterminate on second, etc...

Sort by two properties

Sorts pipeline on first property, and then, if indeterminate on second

Sort by property

Sorts pipeline on element's property, which is provided by given function

~~("aaaa", "bb", "ccc", "d").sortBy(_.length).lp

// Output
~(d, bb, ccc, aaaa)

Sort reversed

Reverse sorts stream elements with provided Ordering

~~(5, 1, 4, 2, 3).sort.lp  // Prints ~(5, 4, 3, 2, 1)

Transpose

Transposes matrix where rows become columns

def all: ~[~[Int]] = ~~(11 to 15, List(21, 22, 23, 24, 25), Vector(31, 32, 33, 34, 35))

all.tp

all.transpose.tp

// Output
---------------------
?
---------------------
~(11, 12, 13, 14, 15)
~(21, 22, 23, 24, 25)
~(31, 32, 33, 34, 35)
---------------------

-------------
?
-------------
~(11, 21, 31)
~(12, 22, 32)
~(13, 23, 33)
~(14, 24, 34)
~(15, 25, 35)
-------------

Run for empty

Runs given function if pipeline is empty

Note: This method might not execute for some pumping methods, if they determine that pipeline is empty from metadata, and would not even attempt to pump elements. Please, test

Run on timer

Runs given function every specified time period in fractional seconds, while elements are being pumped

Note, it will not run even once if all elements pumped in less than the given seconds

Runs before first

Runs given statement when the first element pumped, but before it is passed down the pipeline

This will not run for empty pipeline

Runs after last

Runs given function when pumping for the next element brings nothing

This will not run for empty pipeline

Lazy infinite stream

Lazily unfolds next value with a function taking all prior values

 // Unfoldifg Fibonacci Sequence

(0 to 1).all.unfold(_.letLast(2).sum).letNext(20).lp

 // Output
 ~(0, 1, 1, 2, 3, 5, 8, 13, 21, 34, 55, 89, 144, 233, 377, 610, 987, 1597, 2584, 4181)

Unzips stream in two

val pairs = ('a' to 'g').all.zipMap(_.upper).to[Idx]

pairs.all.lp  // Prints ~((a,A), (b,B), (c,C), (d,D), (e,E), (f,F), (g,G))

val (left, right) = pairs.all.unzip

left.all.lp   // Prints ~(a, b, c, d, e, f, g)

right.all.lp  // Prints ~(G, F, E, D, C, B, A)

Zip that

Merges this and that streams, creating Tuples for corresponding elements

If that stream runs out of elements to merge, the rest of this stream will be discarded too

(1 to 100).all.zip('A' to 'D').lp  // Prints ~((1,A), (2,B), (3,C), (4,D))

Merge stream

Merges this and that streams, creating Tuples for corresponding elements

('a' to 'f').all.zip('A' to 'H', '?', '?').lp

// Output
~((a,A), (b,B), (c,C), (d,D), (e,E), (f,F), (?,G), (?,H))

Merges current folding value

(1 to 7).all.zipFoldAs(0L)(_ + _).tp

// "Running Total" Output
-- --
?  ?
-- --
1  1
2  3
3  6
4  10
5  15
6  21
7  28

Merge index

Creates a new Stream with elements paired with their sequential index

Note: Idx is the first element in the resulting tuples.

('A' to 'F').all.zipIdx('A'.toInt) lp  // Prints ~((65,A), (66,B), (67,C), (68,D), (69,E), (70,F))

Merge index

Creates a new Stream with elements paired with their sequential index, starting at 0

Note: Idx is the first element in the resulting tuples

('A' to 'F').all.zipIdx.lp // Prints ~((0,A), (1,B), (2,C), (3,D), (4,E), (5,F))

Merge property

Creates a new Stream with elements paired with their property, defined by given function

('A' to 'F').all.zipMap(_.toInt).lp  // Prints ~((A,65), (B,66), (C,67), (D,68), (E,69), (F,70))

Merge with next

Creates new Stream with elements paired with the optional next element

(1 to 5).all.zipNext.lp  // Prints ~((1,Opt(2)), (2,Opt(3)), (3,Opt(4)), (4,Opt(5)), (5,Opt.Void))

Merge with prior

Creates new Stream with elements paired with the optional prior element

(1 to 5).all.zipPrior.lp  // Prints ~((Opt.Void,1), (Opt(1),2), (Opt(2),3), (Opt(3),4), (Opt(4),5))

Append

Same as append

Creates a new Stream with given element appended to this Stream

(1 to 5).all + 99 + 100 lp

// Output
~(1, 2, 3, 4, 5, 99, 100)

Append all

Same as appendAll

Creates a new Stream with that Stream appended to this Stream

('1' to '9').all +~ ('a' to 'd') +~ ('A' to 'D') lp

// Output
~(1, 2, 3, 4, 5, 6, 7, 8, 9, a, b, c, d, A, B, C, D)