Distributed Stream Quality of Life Improvements

As I hope most people reading this already know, since Infinispan 8 you can utilize the entire Java 8 Stream API and have it be distributed across your cluster.  This performs the various intermediate and terminal operations on the data local to the node it lives on, providing for extreme performance.  There are some limitations and things to know as was explained at distributed-streams.

The problem with the API up to now was that, if you wanted to use lambdas, it was quite an ugly scene.  Take for example the following code snippet:

8.0 Distributed Streams Example

	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())));

view raw old.java hosted with ❤ by GitHub

However, for Infinispan 9 we utilize a little syntax feature added with Java 8 [1] to add some much needed quality of life improvements.  This allows the most specific interface to be chosen when a method is overloaded.  This allows for a neat interaction when we add some new interfaces that implement Serializable and the various function interfaces (SerializableFunction, SerializablePredicate, SerializableSupplier, etc).  All of the Stream methods have been overridden on the CacheStream interface to take these arguments.

This allows for the code to be much cleaner as we can see here:

9.0 Distributed Streams Example

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

view raw new.java hosted with ❤ by GitHub

Extra Methods

This is not the only benefit of providing the CacheStream interface: we can also provide new methods that aren't available on the standard Stream interface.  One example is the forEach method which allows the user to more easily provide a Cache that is injected on each node as required.  This way you don't have to use the clumsy CacheAware interface and can directly use lambdas as desired.

Here is an example of the new forEach method in action:

	cache.entrySet().stream()
	.filter(e -> e.getKey().startsWith("RedHat"))
	.forEach((c, e) -> {
	c.getCacheManager().getCache("RedHat").put(e.getKey(), e.getValue());
	});

view raw forEach.java hosted with ❤ by GitHub

In this example we take a cache and, based on the keys in it, write those values into another cache. Since forEach doesn't have to be side effect free, you can do whatever you want inside here.

All in all these improvements should make using Distributed Streams with Infinispan much easier.  The extra methods could be extended further if users have use cases they would love to suggest.  Just let us know, and I hope you enjoy using Infinispan!