Stream Gatherers - let's get to know each other better!, Marcin Chrost

This talk focuses on the concept of stream gatherers and their role in enhancing Java Stream API's functionality. The speaker begins by discussing a common scenario of selecting distinct elements from a collection of cars, illustrating the complexities that arise when trying to implement this using traditional methods. He introduces the stream gatherers, a feature added in JDK 22, which simplifies processing by allowing custom operations without altering existing code. The speaker explains how gatherers enable developers to manage state and implement advanced operations like filtering and pattern matching more effectively than the original Stream API. He provides practical examples of built-in gatherers, such as scan and sliding window, before diving into how developers can create custom gatherers to solve specific problems while adhering to design principles like open/closed principle. The session concludes with a call to explore the GitHub repository containing example implementations.

I think we can start with the talk. It's about stream gatherers. We will try to get know each other a little bit better. And the first question to the audience, is anyone using them already? Uh hands up. Uh I do not see a lot to be honest. So, another question is was anyone present at the workshops conducted by Kai Horstmann 3 hours ago when he was talking

yeah, fine. This you definitely didn't regret it. I I can also I was also present there and also the last few slides impressed me a little bit and I added something to my talk at the end. Okay, it's always it's always good to to hear some someone who is wise in this topic. But okay, let's not waste the time. Let's go further and we will start from

the very simple example. Okay. You have a collection of cars and you need to just select distinct of them. It's relatively something very simple. It's just piece of cake. As we can see at this slide of code, we are just having this and now we have this record car and we are have the list of the car and we are just using to select the distinct of

them and as you can see this distinct operator you can apply to the stream does the trick and we do not need to add anything above that. So, if this is so simple, then let's try to complicate things a add something different. Let's try to select cars with distinct color and if you try to do it exactly the same way as the opposite slide, you probably would

start with mapping this car to the color enum, and then trying to select this distinct one, and then you realize that because you have just lost the information about the car and only got the the color, there is no point of return to coming back again to the car from this distinct colors. definitely, this is something not simple as it may seem. And if you you try

to find some solution, you can also use AI of course no everybody is using now, you will finally get to something like that. You need to implement your custom collector, and in the collector using this map which allows you to just map everything by the color, then the value, and then merge this one to into this car, and and return from the collector. And of course, this

means that of course you are able to do this task, but the complexity of the code is surprisingly high, and definitely something seems to be missing in this in this area that we need for such simple task to write some absurdity amount of code, which is what is worse, it's not so easy to understand, is a little bit complicated as you as it can be. So, now

when we are just facing the problem with the such hard code, exactly with this one, we say, "What? That should it be so hard to achieve with them?" And this is the point when I wanted to introduce myself and say how can we overcome this problem, and what this Stream Gatherers play the important role in this area to solve us such puzzles without a lot of troubles.

My name is Marcin Krust. I came here from Poland. I live in Katowice and work in as a software developer in C and Allegro because both both companies because we are I'm working in the outsourcing, but the Allegro is the main one because is the Polish biggest marketplace. we can even say that we are in Poland we are proud of it because non not nor eBay neither

Amazon was able to overcome our market. Allegro is the something like the Polish monopolies at this place, so that's fine that we have something like And except that then working Allegro, I'm also a trainer. I cooperate with two companies in Poland that organize IT trainings Bottega IT Minds and J-Systems. If anyone is here from Poland probably at least it should be conscious about Bottega. If someone is

interested with contacting me, of course this is my webpage, my GitHub, my LinkedIn, so feel free to connect me on this social media or look into my repositories. One of them will be presented the end of the talk as a QR code when we finish all of the examples, let's not so waste so many time on my humble person. Let's go to this area of architecture. So,

definitely it seems that we need something elastic that allows us to introduce some operations that are not predicted by the creators of Java Streams API and exactly is this this is something that is compliant with the open close principle. What is this the second rule of this solid rules from object-oriented programming better ideas and it means if you want to extend something that already exist, this shouldn't

include and require you to change existing code. The code should be written as such that you you should be able to add something new by just adding new classes, implementing adding some implementations, but you shouldn't modify existing code. And this exactly is the opposite of what we have just observed because it seems that we had a lot of intermediate operations in streams, but we have we did

do do not have something like distinct by key or something like that. So, it means okay, this is missing, we should add it to the library. And I will try to provide you that this approach makes no sense and the gatherers one, in my opinion, it's better better fits the bill. But, let's go to the another slide. So, seems that gatherers it's something that should allow us

to perform these tasks a little bit easier than it was previously. And, of course, this is exactly how they work. They have been introduced in JDK 22 as a preview feature and they quickly came as a permanent one, which is not the rule in the nowadays, especially taking for for consideration the context of not virtual threads, but structured concurrency, which is still in the preview version. It's

about six or seven preview. Unfortunately, I don't I I don't know I don't have any idea when this will be added as a permanent one. And also another example are just these string templates because we were waiting for it because string string interpolation is still missing in Java. And unfortunately, it hadn't been included at all. They just vanished uh as a smoke. Okay, but this is a

permanent feature. We have it already added in JDK 24, so you can use it without any obstacles or limits. And it allows advanced grouping and processing stream items. And in my opinion, this is the missing part of the JDK 8 stream API. Let's see on the picture how does it look like. So, if you have the stream API in Java, usually stream processing consists of three parts.

On this on this picture are four, but definitely only three distinct as to be honest. So, we have stream creation, inter- intermediate operation or operations, and the terminal operations, which really runs the stream because Java streams are lazy as such. So, nothing happens until you really run this terminal operations and just execute the stream. And if we are talking about this creation of the stream, of course,

we have a lot of methods dot dot stream on all existing collections and so on. However, you are not limited to do them if you have just your own collection, your own set of data, or something like that. You can always use this stream support or spliterator classes, and you are able to construct your own custom stream without any problems. So, definitely, this is elastic as such.

If you are talking about this terminal operations, there are a lot of them. However, if any of them does not fit your needs, still you have the collect operations that allows you to write your own collector and execute your own custom logic. So, also, you are elastic here. And then, we are going to this intermediate operation. And of course, there are a lot of them, especially this

distinct, filter, map, reducer. Sorry, that's this the terminal one. >> [snorts] >> But, something is missing that uh doesn't allow us to implement any custom operation. And then, in Java 24, we have the gator operator that that's exactly works the same way as collect in the end and the stream support on spliterator on the end. So, allows you to implement your own custom business logic you need

in this intermediate operations. And in my opinion, this exactly perfectly fits the hole that we have in this API. So, the question is how does it look like because uh it may seem that uh implementing such interface that allows take number of items that is not uh presumed at the beginning and also emits the not specified number of items downstream may be a little bit tricky. However,

if you see how does this uh API is implemented, it doesn't it's not so tricky as it makes my seem at at the beginning. So, we have this gatherer we into which we are pushing these elements because uh the very important thing is that that uh gatherer does not work as a pull model. It's it's just receives the elements pushed from the upstream and exactly this is

the way is this stream API works because it's also it's not so doesn't also pull the elements. It's only when you are run the terminal uh the upstream is being notified and pushes the elements to these all the intermediate operations and they are pushing in downstream. And the second missing piece of puzzle is that that behind besides pushing these elements, you should also be notified from the

downstream whether you are accepting them more or not because if not you need to stop the process and just terminate this through this this thing. We will cover this topic later when we will talking about greedy and picky integrators, but let's assume that pushing all the elements is not so automatically. You need to be conscious whether you are they are being accepted or not. Do we have

any building gatherers? Yes, they are and we are trying to show two of them because in my opinion they will somehow shows us how does the gatherer test us and what they allows us to do. Then I also provide you the existing rest and the some libraries that are and then we will go to this custom one because we will be also implementing custom gatherers. So, let's

start from the scan and scan seems that uh This is something like the reduce. it's just works as a reduce operator. However, emits every on the path. So, we are starting from the from the stream, then we are using this mapping of mapping function that just takes the initial element, combines with another, and emits in the result stream as a as a such. And the question is

except how does it work, what's the appliance of such operator? And now we will see the example of current min and max operations. So, let's assume that we have the stream of some values and we want to check what is the current minimum and maximum value. We are not looking only for the global maximum and minimum value, but we want to be conscious what is the minimum

and maximum values at the moment we are checking this stream. So, definitely we should emit something on when every of the items is being emitted in the original stream. And this is exactly the place when the scan gatherer does the trick. So, we should just prepare some structure for storing this local minimum and maximum and the record exactly fits the bill because it's a in my opinion

very great invention in modern Java. Of course, taking into consideration that modern is Java 16 because now we 26 or 27. So, sorry, but it's the 10 10 versions ago, but still. And then we are using this this scan gatherer. We are implementing two things. The first one is that uh we need to initialize the structure with this max and min value. Definitely, this is the because

we need something like the sentinels that doesn't allows you to just jump out of the scale. So, definitely, the max and min value fits the bill. And then every item is being emitted, we are calculating the current minimum and just pushing in into the scanner. And finally, this gatherer does the trick. We are emitting all of them. As you can see, exactly, we have this local or

for each emitted item. So, it could be written without this scan gatherer. However, such gatherer makes it much more easier to to write than it was The another one that I would like you to from the built-in gatherers is the sliding window. And sliding window means that we are taking our stream. We are deciding how many items we want to process at once. And then we are

just taking this amount at the beginning, process it, and then slide the window with this constant size one element further, and we are processing another three [snorts] another and elements and so on and so on and so on. when does this when does this algorithm can be applied? And [clears throat] usually comes into the mind is pattern matching. You are If you are trying to find some

pattern in the streams, the sliding window is exactly what we need because we know which pattern we are looking for. We know the length of the pattern. So, we should just take this window, apply to the first n elements, and check whether the pattern is exactly the same we are looking for, then slide it one one element further, also compare and compare and compare and find all

the occurrences of the pattern. As exactly on this code was can see. So, we have this pattern of three elements. We have this list of the elements we are apply looking for the pattern, and we are just applying the gatherer window sliding with the size of the pattern. It couldn't be much more simpler. Of course, after applying this gatherer, we need to filter whether the pattern is

really really fits the find that sliding window, and then we are just printing that okay, we found the pattern. So, of course, this code works, and you and it informs you that the pattern has been found, that's fine. However, there is one thing that might come into your mind. Okay, we were able to find the pattern, that's I want the indexes. I want to know where exactly

the pattern is located, and this comes a little bit tricky, but this can be obtained with chaining gatherers because gatherer, the gatherer function as such, intermediate operation, yeah? So, intermediate operations can be changed. So, if we employ two gatherers together, we would be able to do that. And what's what's wrong with with this [snorts] approach? Unfortunately, to achieve that, we need to make this a little bit

complicated because firstly, we need to decorate our stream with the index. So, we need to transform the list of our items with the indexes of each of them. And And this kind of code could be written a little bit better because Kai has presented his operator during the workshops with which was more generic but allows you to decorate the streams with that. I decided to leave the

code as it is. However, I warmly recommend also to look at Kai's slides because in my opinion, his implementation of this decorating indexing is better and also probably the such operator is already present if in the one of the libraries I will present you on the next slides because the building gatherers let's find there are that they are but there's it's their number is too too less

in my opinion. However, after decorating with the index, we are now able to apply the second gatherer, window sliding, and then we are just taking the three elements per such, then we are find whether they are matching the And after that, we are able to Sorry. To find not only the pattern but also index. I think that this code could be a little bit rewritten but definitely

it fits the bill and shows you that just chaining two gatherers allows you to find pattern and the index on that. Are there any other building? Yes, they are. This is something like fold. Works exactly as a reduce but retains a stream so that allows you to get one value but for example, chaining with the other operators, it's something it's fine because usually using reduce somehow terminates

your stream and it doesn't allow you to work with this value further. With default, you are able to to continue working. It's maybe very helpful in some situations. Window feeds divides items into batches of given size. Uh so, it just takes, for example, the first three, then the next three, and so on and so on. They are not overlapping as in in the case of window sliding.

So, for example, this gatherer can be used for just for sampling, for taking average version of the every five items and so on. Definitely, just to reduce the number of items to being processed immediately, to reduce the overload, whatever and never. Map concurrent, the third one, it's spouse concurrent mapping of items. And in such using this gatherer, I have a lot of doubts because it has a

lot of pros. Just uses visual threads, allows you to max concurrency, but the there is one feature that in my opinion makes using it very tricky because it preserves order. And then, does it does mean that if you are just paralyzing the operations for mapping, and they are not the equal if the according to the velocity, then it means that if you do it wrongly, the items

that are being processed quickly may then need and wait until the slower mapping operations would be done. In my opinion, this did crash the performance a little bit. Except that we have also libraries that can also recommend you. I just copied these links from this Sky Horseman slides. Especially, my heart is very sparkling in the second one because it has been created by Grzegorz Piwowarek, my colleague

from Bottega IT Minds. So, this is also the Polish insert [snorts] into this gatherer operations. So, if Grzegorz has written something, I can I [snorts] can recommend from my heart. So, definitely, don't don't hesitate to use all of them and probably all of them has this indexing operator or decorating operations I have done. Okay. Now, if we have already presented this building gatherers, the most interesting part

starts to appear. Because are we able to write some custom one? Of And in my opinion, this is the most important thing because custom gatherers is is exactly what we need because we need to be able to create our custom elastic intermediate operations. So, let's start from some theory because we need to in introduce the interfaces and then we are try to present some implementations of them.

let's start from the interface gatherer because this is the clue of the of the all operations as we need to add. As as you can see, it doesn't look very friendly at the first glance because this is a generic one. It contains of three types of the generics. It's to be honest, it's something very very common as a as a thing collector, but definitely who usually looks

into the collector how it's implemented and what's the interface, but exactly this is the same the same thing with the gatherer. And this interface contains three things. This is the type of the emitted items. R is A is a state and the T is the integrator. That's T are the items emitted into the gatherer. >> Okay. But, what we see here is that this consists of four

methods, which three of them have the default implementations are and are not the mandatory ones. And the only key operation that needs to be implemented when you want to specify your own custom gatherer, it's integrator. So, definitely we should jump inside this integrator interface and see what does it consist of. the custom gatherer important parts. I'm also I'm only talking about the sequential one. The parallel without

of the scope of this talk is more also more interesting topic, but let's let's focus only on the sequential one. And as you can see, we have the integrator and integrator consumes one up to n items and also the state because we are also provide some state that is being memo- memorized by the integrator because usually when we have some more comp- sophisticated logic basing only on

actually pushed items is not enough to be allowed to push something downstream. And as you can see, the the number of items and the state that are being pushed into the integrator from outside is exactly the same state you are just pushing from integrator to downstream. So, I would say that even this this API is a symmetric one. What you put is what you get. Of course,

only the number can be different. So, for example, if you is not being said that for example, if you pushed three items into the integrator, you will just you are obliged to emit exactly three. No, it's the completely same. You can even just emit nothing and it's still and it's fine. >> And this is the main part of of the gatherer. We have also something like initializer

because if you want to have this memory and the state, you need to somehow initialize it before the gatherer runs before the first item is implemented. This is the optional initializer and also you can have something like finisher that allows you to emit some additional items because sometimes you are not able to emit anything until you know that the stream has really completed. And we also see

later. But let's focus on the integrator as such. This is because this is the key this interface has only one method. And if you look at uh signature, it starts to seem pretty obvious what does it mean. So we have A, T, and R and it seemed that the T is the element that is just being pushed into this integrator. A is a state that can be

somehow state stored internally and used by this by this integrator. And the last thing because we have only have the internal state, we have something that is being pushed so we have the input. We need the output and the output needs to be something that represents us to possibility to push down the elements and let's also we need to be able to determine whether we should push

the elements or the downstream is just rejecting them. And as you can see this downstreaming API as presented on the below the slide just consists one of the method. You are just pushing the element and you will get the result of boolean and true means that you are able to push the further ones. False means that story stop. I cannot accept it anymore. The the time is

over. And you can also see that because the downstream returns the information about where you are allowed to push them or not, you you should also be able to return the informations above to your upstream. And exactly this is the The why the integrate method also returns the boolean value exactly of the same reason. False means story stop. Don't push me any other elements. We we are

we are up. So, let's start from something simple because I said that the state is an optional thing in integrator. And we are we are able to implement some of them without any state. We are just accepting elements, doing something with them, and if they are fitting our view we are pushing downstream or not and the state makes no sense in in such case. And what I

want to present you is the operation map not null elements because sometimes it's it's very hard to to process such operations because let's assume that we have the stream of stream of elements and some of them may be null. More or less. Of course, we can use optional to get avoid such situation, but optional makes an API a little bit cluttered and it's also additional level of

generics. So, maybe sometimes it's easier to use null elements, but if you want to safely map them in the mapping functions, you need to add some if in the lambda and so also not so easy as it may seems and we are trying to being as much concise as it possible. So, we are using this not null gatherer and in in the case of map not null,

it seems that the state is completely out of the scope because we need we don't need them. We are just getting the element, mapping it, and just pushing downstream of course if it if it if it's not null. And the implementation is exactly very simple. We are using a void the the big void type as a type of the state because we the state of the gatherer

and we are just taking the element of T, checking whether it's null because if not we are just going further and if it's not now, we are just mapping it and pushing to the downstream. And that's all. And and the the other as you can see, we are just allows us to just combine these operations into one single lambda. We are just adding the gather. And then,

if you run it, you can see the result. It's two, four, and six. And the null elements is completely vanishing. It is exactly what we needed to achieve our target. Okay. if this works exactly that way, now let's go to the question of greedy and non-greedy picky because I have I was looking for a good word describing the integrator if it's not greedy. And also, Kari said,

"Okay, the picky is the better one." So, I decided to check that. And the question is, what's the difference between them because it hasn't been stated already. usually, the greedy integrator, and what is more funny, it's being used it should be it should be used as a default one because using greedy integrators allows the stream API beneath to optimize some operations. Integrator integrator greedy just means that

it may return false and stop accepting elements only that if the downstream returns exactly false. So, it It cannot decide that he stops emitting elements and notifying upstream about that in case when when the downstream does not stop accepting elements. And this is the it's it's on this question. So, in in such case, we need to stop the elements at the level of our integrator. We need

to use the picky one. And the only difference between creating them is just using the other functions because usually we are using the greedy integrator is of greedy we are using picky integrator is off definitely. And what is more interesting is because this one seems that mean not this the default but usually it should be as the default one. But the API has been created as such

when you have another options. So, the question is when can we use this picky integrator and why when does it solves our problems? So, let's go to this code that we are using here. We are trying to implement the gatherer that find first element of the streams that uh fit the predicate. And in such case we are using does the integrator that also does not keeps the

state because it makes no sense to use it. But in the case of this integrator we need to test each element when we are using when we are being pushed by them. And if it fits the bill, we are just pushing this item and return false and stop uh emitting anything uh to the upstream. In any other case that the item does not fit it, it means

that we are uh looking for the another one. So, we are waiting for another item. So, we need to return true and then we are just pushing everything upwards. As you can see in the example we have these numbers of from one to five. And we are checking looking for the first one that is uh that it's even. And as you can see, you can imagine that

in in case of this uh gatherer and this all these conditions, we should emit only the element of two. And that's true. And the question is what would happen if we are not using greedy integrator here picky integrator here. So, if you are trying to optimize the stream and use the greedy integrator, which usually people are doing when you're using implementers, would finish in the situations that

even taking the the the the the condition that we should not push any items further, we would get two and four in the result stream. So, seems that using the greedy integrators does forces this elementer from uh uh upstream being pushed down and downstream unfortunately consumes all of them. So, definitely if your integra- if your integrator and then gatherer needs to stop emitting items at its discretion,

you need to use this uh picky integrator because the greedy one wouldn't do the trick. because we have only 16 minutes left, the question is, do we need stateful And of course, we need them because more complex solutions need to use to keep the memory of items being emitted. And let's go initial problem that we had at the beginning of the talk because we wanted to be

able to select distinct element from the stream, but based on some key. So, we need to be able to perform some mapping operations of the each item in the stream, check whether the key retrieved by this operation fits our bill, stay and only emit that items that are being distinct along this key. So, how does it would like? Definitely, this cannot be achieved without any state because

you should be conscious about what already items according to this key have been emitted because we only need to you want to preserve the distinct of them. So, you need to have the state and somehow contain already emitted keys. So, in such case, we need to add the this another part which is usually optional taking the in the case of gatherers is the initializer. And this initializer,

we creating a hash set of the of the keys that would [snorts] be somehow extracted from the items in the original original stream. And as you can see, at the beginning we have this empty hash set, that's fine because we have we haven't already emitted anything. now we are able to consume the elements being pushed from the upstream. So, our integrator is just type of TNT, which

means that we are consuming element T of T pushing element of T. And this set of P would allow us to keep the track of the already emitted elements. Okay. How does it look like inside the implementation? Sorry. in the case of the greedy one, we are just for each item we are applying the selectors who are extracting the key. In our case, would be just getting

the color of the car. And if the hash set of the colors does not has this key already, so we have we hadn't emitted any car of the given color, we are of course adding this color to the to the to the state because the state needs to be maintained. And then we are just pushing this element downstream and return the value of of the of the

pushing how does it push whether or not. And in the case when this key has been already emitted, we just don't emit anything and just return after that, we get to the point that this allows us to solve the initial puzzle that we have uh presented at the beginning of the in a very easy way. So, we have only just creating this distinct by selector gatherer, we

are just getting the selector function just car get color. And after applying that, the solution is pretty so straightforward and just consists one liner. In my opinion, this this is a game changer. Definitely in comparison to the original solution without gatherers, it's something really better that allows us to you solve the the problems of uh very simple and you can also see that we with this approach,

we are not adding anything new to the stream API. We are just using this gather function and what we only change is the new implementation new class that implements the gatherer interface. So, definitely this approach is exactly what is compliant with this open close principle. You do not need you do not modify the Java stream API. You are implementing your own class and pushing it into the

in the the place when it can be plugged in. So, you are the the the Java stream API is stable. You can just extend it by your own implementations. Nice. Okay. So, if we had this gatherer already presented, we know how this this integrator works either picking or greedy. We are we know how the initializer works. We we do we know how we can apply the state.

The what what's else we need we we can add to this process as at the original picture, we have three items. It was the integrator, it was initializer, we have already presented it, and the third one is the finisher. when this last guy can be used? What we need finisher for? And this is exactly the another problem I want to present you with this gatherers. So, have

very simple situations. We have some collection stream of items. We have some extractor functions exactly with the same one as with the cars and extracting colors. And we want uh max element of the original stream, but not basing on the elements as such, but on the value extracted by the key. So, for example, we have the we have cars, we have the prices of mileage, and for

example, we want to find the car which has the lowest price in the whole collection. finisher jumps into because if you are looking about emitting one value that has the max that has for example the max or minimum value we're looking for, we are not able to emit it until we are 100% sure that all of the items has been already pushed from outside because if we

if we are still expecting some items, we cannot ensure that any of them wouldn't fit the wouldn't be the maximum or minimum one, and emitting any of them previously could completely falsify the result of Now, let's go to this implementation. Of course, we need the state because we need to have the reference to this maximum or minimum We are using max, so the maximum value. And of

course, the best way to keep some reference in Java that can be changed during the execution and it's not final one is using atomic reference. Don't do at home. Definitely, it should be some class that wraps the value of T, but of if of course if Java has already something like that, that's exactly the trick that is very often used just when we're when we're talking about

mutable reference. Now, if you have this initializer, let's go to this integrator and integrator of greedy we are just taking this function. We are checking whether this element is empty with already have this max item or not. If it is present, we are just extracting this current property from from item. And if this property is bigger than the max property of the of the item stored in

this atomic reference, we are just updating it. And of course, if there's already nothing being pushed, we are just setting this reference of the first item. That's fine. It It can be also a little bit shortened, but definitely fits the bill and also somehow and prevents you from the nullability, so it should be everything fine. But the most important part of this integrator is that that after

emitting all of the in the max item ref, you have this element that is that fits the that that conforms the condition that it should be the max by the key. So, after emitting all of the elements, you have this reference. Of course, it can be especially if there was no items emitted in the stream, and we have also should protect ourselves be before that. It means

that we need to check whether the reference is nullable or not, and if it's not, we are just pushing this element downstream. And that's all. Everything is fine. We are emitting this one element, and the code that are using this gatherer is present on this slide. So, we are just using this text Alice has a cat and the candles Alice. And we are trying to find the

length the more the the longest word in this in this text. We are just looking for the first one, so we are just taking this this text. We are changing into the stream by splitting with the regex. We are just skipping all white color white characters. And then, after that, we are just using our gatherer. We are just as extractor we are using this length function of

the string. And finally, we will of course get the list of one element, and this one element is exactly Alice, what we already presented. And that's fine. Everything is okay. You can also see gatherer returns the stream as such, so we need to somehow terminate the operations and make the to list. And of course, it's exactly what we needed. As a outro, because we are 6 minutes

left, but probably I would finish a little bit earlier. The pros and cons of the gatherers, and they are very flexible because you can somehow implement anything you need in intermediate operation. And also, they simplify your code because as you can see with this distinct by key, you are just writing one gatherer enclosing the logic inside it, and just run the gather operations, put this gatherer inside

it, and that's fine. Okay, so are there any cons? Of course, because they are they do not have any support from from primitive types as exactly was it is the Java stream API. And another thing is that they do not have any access to stream characteristics, which somehow blocks you for further optimizations. You are not enable to check for example whether the stream is finished or not

or something like that. So, you cannot rely on any stream characteristics. You need to only rely on the emitted elements on or and the downstream reactions. Is it enough? In my opinion it is, but of course this can be a little bit questionable. If you want to try this code at home, this is the QR code with the repository. You can just scan it and they are

all all of this examples are being presented there. So, definitely I encourage you to take it and check how does it look like and what you can achieve for that. that's all. I thank you very much for the for the presents. Enjoy and I wish you to enjoy the rest of the conference. There are still a lot of >> interesting talks beforehand us and have a nice

afternoon. >> [applause]