==  Streams

Java 8 introduced the concept of a link:https://docs.oracle.com/javase/8/docs/api/java/util/stream/Stream.html[Stream]
which allows functional-style operations on collections rather than having to procedurally
iterate over the data yourself. Stream operations can be implemented in a fashion very
similar to MapReduce.  Streams, just like MapReduce allow you to perform processing
upon the entirety of your cache, possibly a very large data set, but in an efficient way.

Also since we can control how the entries are iterated upon we can more efficiently perform the operations in
a cache that is distributed if you want it to perform all of the operations across the cluster
concurrently.

A stream is retrieved from the link:https://docs.jboss.org/infinispan/{infinispanversion}/apidocs/org/infinispan/Cache.html#entrySet--[entrySet],
link:https://docs.jboss.org/infinispan/{infinispanversion}/apidocs/org/infinispan/Cache.html#keySet--[keySet] or
link:https://docs.jboss.org/infinispan/{infinispanversion}/apidocs/org/infinispan/Cache.html#values--[values] collections returned from the
Cache by invoking the link:https://docs.oracle.com/javase/8/docs/api/java/util/Collection.html#stream--[stream] or
link:https://docs.oracle.com/javase/8/docs/api/java/util/Collection.html#parallelStream--[parallelStream] methods.

=== Common stream operations

This section highlights various options that are present irrespective of what type of underlying cache
you are using.

==== Key filtering

It is possible to filter the stream so that it only operates upon a given subset of keys.  This can be done
by invoking the
link:https://docs.jboss.org/infinispan/{infinispanversion}/apidocs/org/infinispan/CacheStream.html#filterKeys-java.util.Set-[filterKeys]
method on the `CacheStream`.  This should always be used over a Predicate
link:http://docs.oracle.com/javase/8/docs/api/java/util/stream/Stream.html?is-external=true#filter-java.util.function.Predicate-[filter]
and will be faster if the predicate was holding all keys.

If you are familiar with the ``AdvancedCache`` interface you may be wondering why you even use
link:https://docs.jboss.org/infinispan/{infinispanversion}/apidocs/org/infinispan/AdvancedCache.html#getAll-java.util.Set-[getAll]
over this keyFilter.  There are some small benefits (mostly smaller payloads) to using getAll
if you need the entries as is and need them all in memory in the local node.  However if you
need to do processing on these elements a stream is recommended since you will get both
distributed and threaded parallelism for free.

==== Segment based filtering

NOTE: This is an advanced feature and should only be used with deep knowledge of Infinispan segment and hashing techniques.
These segments based filtering can be useful if you need to segment data into separate invocations.
This can be useful when integrating with other tools such as
link:http://spark.apache.org/[Apache Spark].

This option is only supported for replicated and distributed caches.  This allows the user to operate upon
a subset of data at a time as determined by the
link:https://docs.jboss.org/infinispan/8.2/apidocs/org/infinispan/distribution/ch/KeyPartitioner.html[KeyPartitioner].
The segments can be filtered by invoking
link:https://docs.jboss.org/infinispan/{infinispanversion}/apidocs/org/infinispan/CacheStream.html#filterKeySegments-java.util.Set-[filterKeySegments]
method on the `CacheStream`.  This is applied after the key filter but before any intermediate operations are performed.

=== Local/Invalidation

A stream used with a local or invalidation cache can be used just the same way you would use a stream on a
regular collection. Infinispan handles all of the translations if necessary behind the scenes and works with all
of the more interesting options (ie. storeAsBinary, compatibility mode, and a cache loader).

==== Example

The code below takes a cache and returns a map with all the cache entries whose values contain the string "JBoss"
[source,java]
----
Map<Object, String> jbossValues = cache.entrySet().stream()
              .filter(e -> e.getValue().contains("JBoss"))
              .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
----

=== Distribution/Replication

This is where streams come into their stride.  When a stream operation is performed it will
send the various intermediate and terminal operations to each node that has pertinent data.
This allows processing the intermediate values on the nodes owning the data, and only sending
the final results back to the originating nodes, improving performance.


==== Rehash Aware

Internally the data is segmented and each node only performs the operations upon the data it owns as a primary owner.
This allows for data to be processed evenly, assuming segments are granular enough to provide for equal amounts of
data on each node.

When you are utilizing a distributed cache, the data can be reshuffled between nodes when a
new node joins or leaves. Distributed Streams handle this reshuffling of data automatically so you don't
have to worry about monitoring when nodes leave or join the cluster.
Reshuffled entries may be processed a second time, and we keep track of the processed entries at the
key level or at the segment level (depending on the terminal operation) to limit the amount of
duplicate processing.

It is possible but highly discouraged to disable rehash awareness on the stream.  This should only be considered if
your request can handle only seeing a subset of data if a rehash occurs.  This can be done by invoking
link:https://docs.jboss.org/infinispan/{infinispanversion}/apidocs/org/infinispan/CacheStream.html#disableRehashAware--[CacheStream.disableRehashAware()]
The performance gain for most operations when a rehash doesn't occur is completely negligible.
The only exceptions are for iterator and forEach, which will use less memory, since they do not have
to keep track of processed keys.

WARNING: Please rethink disabling rehash awareness unless you really know what you are doing.

==== Serialization

Since the operations are sent across to other nodes they must be serializable by Infinispan marshalling.  This allows the
operations to be sent to the other nodes.

The simplest way is to cast your lambdas as Serializable as below.

In our previous example we used a `Collector` to collect all the results into a `Map`.
Unfortunately the link:https://docs.oracle.com/javase/8/docs/api/java/util/stream/Collectors.html[Collectors]
class doesn't produce Serializable instances.  Thus if you need to use these, you can use the newly provided
link:http://docs.jboss.org/infinispan/{infinispanversion}/apidocs/org/infinispan/stream/CacheCollectors.html[CacheCollectors]
class which allows for a `Supplier<Collector>` to be provided.  This instance could then use the
link:https://docs.oracle.com/javase/8/docs/api/java/util/stream/Collectors.html[Collectors]
to supply a `Collector` which is not serialized. You can read more details about how the
collector peforms in a distributed fashion at link:user_guide.html#_distributed_stream_execution[distributed execution].

[source,java]
----
Map<Object, String> jbossValues = cache.entrySet().stream()
              .filter((Serializable & Predicate<Map.Entry<Object, String>>) e -> e.getValue().contains("Jboss"))
              .collect(CacheCollectors.serializableCollector(() -> Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue)));
----

The recommended way is to use an 
link:user_guide.html#_advanced_externalizers[AdvancedExternalizer]
as this provides the smallest payload.

==== Parallel Computation

Distributed streams by default try to parallelize as much as possible.  It is possible for the end user to control this and
actually they always have to control one of the options.  There are 2 ways these streams are parallelized.

===== Local to each node
When a stream is created from the cache collection the end user can choose between invoking
link:https://docs.oracle.com/javase/8/docs/api/java/util/Collection.html#stream--[stream] or
link:https://docs.oracle.com/javase/8/docs/api/java/util/Collection.html#parallelStream--[parallelStream]
method.  Depending on if the parallel stream was picked will enable multiple threading for
each node locally.  Note that some operations like a rehash aware iterator and forEach operations
will always use a sequential stream locally.  This could be enhanced at some point to allow for
parallel streams locally.


===== Remote requests
When there are multiple nodes it may be desirable to control whether the remote requests are all processed
at the same time concurrently or one at a time.  By default all terminal operations except the iterator
perform concurrent requests.  The iterator, method to reduce overall memory pressure on the local node,
only performs sequential requests which actually performs slightly better.

If a user wishes to change this default however they can do so by invoking the
link:https://docs.jboss.org/infinispan/{infinispanversion}/apidocs/org/infinispan/CacheStream.html#sequentialDistribution--[sequentialDistribution]
or link:https://docs.jboss.org/infinispan/{infinispanversion}/apidocs/org/infinispan/CacheStream.html#parallelDistribution--[parallelDistribution]
methods on the `CacheStream`.   Note that currently intermediate operations return a `Stream` instance
so you must make sure to invoke these methods before calling another intermediate operation.


==== Task timeout

It is possible to set a timeout value for the operation requests. This timeout is used only for remote requests timing out and
it is on a per request basis. The former means the local execution will not timeout and the latter means if you have a failover
scenario as described above the subsequent requests each have a new timeout.  If no timeout is specified it uses the
replication timeout as a default timeout. You can set the timeout in your task by doing the following: 

[source,java]
----
CacheStream<Object, String> stream = cache.entrySet().stream();
stream.timeout(1, TimeUnit.MINUTES);
----

For more information about this, please check the java doc in
link:https://docs.jboss.org/infinispan/{infinispanversion}/apidocs/org/infinispan/CacheStream.html#timeout-long-java.util.concurrent.TimeUnit-[timeout]
javadoc.

==== Injection

The link:https://docs.oracle.com/javase/8/docs/api/java/util/stream/Stream.html[Stream]
has a terminal operation called
link:https://docs.oracle.com/javase/8/docs/api/java/util/stream/Stream.html#forEach-java.util.function.Consumer-[forEach]
which allows for running some sort of side effect operation on the data.  In this case it may be desirable to get a reference to
the `Cache` that is backing this Stream.  If your `Consumer` implements the
link:http://docs.jboss.org/infinispan/{infinispanversion}/apidocs/org/infinispan/stream/CacheAware.html[CacheAware]
interface the `injectCache` method be invoked before the accept method from the `Consumer` interface.

==== Distributed Stream execution

Distributed streams execution works in a fashion very similiar to map reduce.  Except in this case we are sending zero to many intermediate operations
(map, filter etc.) and a single terminal operation to the various nodes.  The operation basically comes down to the following:

. The desired segments are grouped by which node is the primary owner of the given segment

. A request is generated to send to each remote node that contains the intermediate and terminal operations including which segments it should process

.. The terminal operation will be performed locally if necessary

.. Each remote node will receive this request and run the operations and subsequently send the response back

. The local node will then gather the local response and remote responses together performing any kind of reduction required by the operations themselves.

. Final reduced response is then returned to the user

In most cases all operations are fully distributed, as in the operations are all fully applied on each remote node and usually only the last operation or something related may be
reapplied to reduce the results from multiple nodes.  One important note is that intermediate values do not actually have to be serializable, it is the last value
sent back that is the part desired (exceptions for various operations will be highlighted below).

===== Terminal operator distributed result reductions
The following paragraphs describe how the distributed reductions work for the various terminal operators.  Some of these are special in that an intermediate value may
be required to be serializable instead of the final result.

allMatch noneMatch anyMatch::
The link:https://docs.oracle.com/javase/8/docs/api/java/util/stream/Stream.html#allMatch-java.util.function.Predicate-[allMatch]
operation is ran on each node and then all the results are logically anded together locally
to get the appropriate value.  The
link:https://docs.oracle.com/javase/8/docs/api/java/util/stream/Stream.html#noneMatch-java.util.function.Predicate-[noneMatch]
and
link:https://docs.oracle.com/javase/8/docs/api/java/util/stream/Stream.html#anyMatch-java.util.function.Predicate-[anyMatch]
operations use a logical or instead. These methods also have early termination support,
stopping remote and local operations once the final result is known.

collect::
The link:https://docs.oracle.com/javase/8/docs/api/java/util/stream/Stream.html#collect-java.util.stream.Collector-[collect]
method is interesting in that it can do a few extra steps.  The remote node performs
everything as normal except it doesn't perform the final
link:https://docs.oracle.com/javase/8/docs/api/java/util/stream/Collector.html#finisher--[finisher]
upon the result and instead sends back the fully combined results.  The local thread
then link:https://docs.oracle.com/javase/8/docs/api/java/util/stream/Collector.html#combiner--[combines]
the remote and local result into a value which is then finally finished.  The key
here to remember is that the final value doesn't have to be serializable but rather
the values produced from the link:https://docs.oracle.com/javase/8/docs/api/java/util/stream/Collector.html#supplier--[supplier]
and link:https://docs.oracle.com/javase/8/docs/api/java/util/stream/Collector.html#combiner--[combiner]
methods.

count::
The link:https://docs.oracle.com/javase/8/docs/api/java/util/stream/Stream.html#count--[count]
method just adds the numbers together from each node.

findAny findFirst::
The link:https://docs.oracle.com/javase/8/docs/api/java/util/stream/Stream.html#findAny--[findAny]
operation returns just the first value they find, whether it was from a remote node
or locally.  Note this supports early termination in that once a value is found it
will not process others.  Note the findFirst method is special since it requires a sorted
intermediate operation, which is detailed in the
link:user_guide.html#_intermediate_operation_exceptions[exceptions] section.

max min::
The link:https://docs.oracle.com/javase/8/docs/api/java/util/stream/Stream.html#max-java.util.Comparator-[max] and
link:https://docs.oracle.com/javase/8/docs/api/java/util/stream/Stream.html#min-java.util.Comparator-[min] methods find the respective min or max value on each node then a final
reduction is performed locally to ensure only the min or max across all nodes is returned.

reduce::
The various reduce methods link:https://docs.oracle.com/javase/8/docs/api/java/util/stream/Stream.html#reduce-java.util.function.BinaryOperator-[1] , 
link:https://docs.oracle.com/javase/8/docs/api/java/util/stream/Stream.html#reduce-T-java.util.function.BinaryOperator-[2] , 
link:https://docs.oracle.com/javase/8/docs/api/java/util/stream/Stream.html#reduce-U-java.util.function.BiFunction-java.util.function.BinaryOperator-[3] will end up serializing
the result as much as the accumulator can do.  Then it will accumulate the local and remote results together locally, before combining if you have provided that.  Note this means
a value coming from the combiner doesn't have to be Serializable.

==== Key based rehash aware operators

The link:https://docs.jboss.org/infinispan/{infinispanversion}/apidocs/org/infinispan/CacheStream.html#iterator--[iterator],
link:https://docs.jboss.org/infinispan/{infinispanversion}/apidocs/org/infinispan/CacheStream.html#spliterator--[spliterator]
and link:https://docs.jboss.org/infinispan/{infinispanversion}/apidocs/org/infinispan/CacheStream.html#forEach-java.util.function.Consumer-[forEach]
are unlike the other terminal operators in that the rehash awareness has to keep
track of what keys per segment have been processed instead of just segments.  This is
to guarantee an exactly once (iterator & spliterator) or at least once behavior (forEach)
even under cluster membership changes.

The `iterator` and `spliterator` operators when invoked on a remote node will return back batches
of entries, where the next batch is only sent back after the last has been fully consumed.  This
batching is done to limit how many entries are in memory at a given time.  The user node will hold
onto which keys it has processed and when a given segment is completed it will release those keys from
memory.  This is why sequential processing is preferred for the iterator method, so only a subset of segment
keys are held in memory at once, instead of from all nodes.

The forEach method also returns batches, but it returns a batch of keys after it has finished processing
at least a batch worth of keys.  This way the originating node can know what keys have been processed
already to reduce chances of processing the same entry again.  Unfortunately this means it is possible
to have an at least once behavior when a node goes down unexpectedly.  In this case that node could have
been processing a batch and not yet completed one and those entries that were processed but not
in a completed batch will be ran again when the rehash failure operation occurs.  Note that adding a
node will not cause this issue as the rehash failover doesn't occur until all responses are received.

These operations batch sizes are both controlled by the same value which can be configured by invoking
link:https://docs.jboss.org/infinispan/8.2/apidocs/org/infinispan/CacheStream.html#distributedBatchSize-int-[distributedBatchSize]
method on the `CacheStream`.  This value will default to the `chunkSize` configured in state transfer.
Unfortunately this value is a tradeoff with memory usage vs performance vs at least once and your
mileage may vary.

===== Using `iterator` with a replication cache

Currently if you are using a replicated cache the `iterator` or `spliterator`
terminal operations will not perform any of the operations remotely
and will instead perform everythign on the local node. This is for performance as doing a
remote iteration process is very costly.

==== Intermediate operation exceptions

There are some intermediate operations that have special exceptions, these are
link:https://docs.oracle.com/javase/8/docs/api/java/util/stream/Stream.html#skip-long-[skip],
link:https://docs.oracle.com/javase/8/docs/api/java/util/stream/Stream.html#peek-java.util.function.Consumer-[peek],
sorted link:https://docs.oracle.com/javase/8/docs/api/java/util/stream/Stream.html#sorted-java.util.Comparator-[1]
link:https://docs.oracle.com/javase/8/docs/api/java/util/stream/Stream.html#sorted--[2].
& link:https://docs.oracle.com/javase/8/docs/api/java/util/stream/Stream.html#distinct--[distinct].
All of these methods have some sort of artificial iterator implanted in the stream
processing to guarantee correctness, they are documented as below.  Note this means
these operations may cause possibly severe performance degradation.

Skip::
An artificial iterator is implanted up to the intermediate skip operation.
Then results are brought locally so it can skip the appropriate amount of elements.
Peek::
An artificial iterator is implanted up to the intermediate peek operation.
Only up to the number of peeked elements is returned a remote node.  Then results
are brought locally so it can peek at only the amount desired.
Sorted::
WARNING: This operation requires having all entries in memory on the local node.
An artificial iterator is implanted up to the intermediate sorted operation.
All results are sorted locally.  There are possible plans to have a distributed sort which
returns batches of elements, but this is not yet implemented.
Distinct::
WARNING: This operation requires having all or nearly all entries in memory on the local node.
Distinct is performed on each remote node and then an artificial iterator returns those distinct values.
Then finally all of those results have a distinct operation performed upon them.

The rest of the intermediate operations are fully distributed as one would expect.

=== Examples
Word count is a classic, if overused, example
of map/reduce paradigm. Assume we have a mapping of key -> sentence stored on
Infinispan nodes. Key is a String, each sentence is also a String, and we have
to count occurrence of all words in all sentences available. The implementation
of such a distributed task could be defined as follows:

[source,java]
----
public class WordCountExample {

   /**
    * In this example replace c1 and c2 with
    * real Cache references
    *
    * @param args
    */
   public static void main(String[] args) {
      Cache<String, String> c1 = ...;
      Cache<String, String> c2 = ...;

      c1.put("1", "Hello world here I am");
      c2.put("2", "Infinispan rules the world");
      c1.put("3", "JUDCon is in Boston");
      c2.put("4", "JBoss World is in Boston as well");
      c1.put("12","JBoss Application Server");
      c2.put("15", "Hello world");
      c1.put("14", "Infinispan community");
      c2.put("15", "Hello world");

      c1.put("111", "Infinispan open source");
      c2.put("112", "Boston is close to Toronto");
      c1.put("113", "Toronto is a capital of Ontario");
      c2.put("114", "JUDCon is cool");
      c1.put("211", "JBoss World is awesome");
      c2.put("212", "JBoss rules");
      c1.put("213", "JBoss division of RedHat ");
      c2.put("214", "RedHat community");

      Map<String, Integer> wordCountMap = c1.entrySet().parallelStream()
         .map((Serializable & Function<Map.Entry<String, String>, String[]>) e -> e.getValue().split("\\s"))
         .flatMap((Serializable & Function<String[], Stream<String>>) Arrays::stream)
         .collect(CacheCollectors.serializableCollector(() -> Collectors.groupingBy(Function.identity(), Collectors.counting())));
   }
}

----

In this case it is pretty simple to do the word count from the previous example.

However what if we want to find the most frequent word in the example?  If you take a second
to think about this case you will realize you need to have all words counted  and available
locally first. Thus we actually have a few options.

We could use a finisher on the collector, which is invoked on the user thread
after all the results have been collected.
Some redundant lines have been removed from the previous example.

[source,java]
----
public class WordCountExample {
   public static void main(String[] args) {
      // Lines removed

      String mostFrequentWord = c1.entrySet().parallelStream()
         .map((Serializable & Function<Map.Entry<String, String>, String[]>) e -> e.getValue().split("\\s"))
         .flatMap((Serializable & Function<String[], Stream<String>>) Arrays::stream)
         .collect(CacheCollectors.serializableCollector(() -> Collectors.collectingAndThen(
            Collectors.groupingBy(Function.identity(), Collectors.counting()),
               wordCountMap -> {
                  String mostFrequent = null;
                  long maxCount = 0;
                     for (Map.Entry<String, Long> e : wordCountMap.entrySet()) {
                        int count = e.getValue().intValue();
                        if (count > maxCount) {
                           maxCount = count;
                           mostFrequent = e.getKey();
                        }
                     }
                     return mostFrequent;
               })));

}

----

Unfortunately the last step is only going to be ran in a single thread, which if we have a lot of
words could be quite slow.  Maybe there is another way to parallelize this with Streams.

We mentioned before we are in the local node after processing, so we could actually use
a stream on the map results.  We can therefore use a parallel stream on the results.

[source,java]
----
public class WordFrequencyExample {
   public static void main(String[] args) {
      // Lines removed

      Map<String, Long> wordCount = c1.entrySet().parallelStream()
              .map((Serializable & Function<Map.Entry<String, String>, String[]>) e -> e.getValue().split("\\s"))
              .flatMap((Serializable & Function<String[], Stream<String>>) Arrays::stream)
              .collect(CacheCollectors.serializableCollector(() -> Collectors.groupingBy(Function.identity(), Collectors.counting())));
      Optional<Map.Entry<String, Long>> mostFrequent = wordCount.entrySet().parallelStream().reduce(
              (e1, e2) -> e1.getValue() > e2.getValue() ? e1 : e2);
----

This way you can still utilize all of the cores locally when calculating the most frequent element.

Also remember that `Streams` are a JRE tool now and there are a multitude of examples that can
be found all over.  Just remember that your operations need to be Serializable in some fashion!
