# Stopping an infinite stream with collection state

I took up a somewhat interesting challenge when I answered this question. The task is to collect the first occurrences of different types of animals from an infinite stream until you've caught them all.

The requirements given:

1. POJO Animal
2. enum AnimalType
3. method to get AnimalType from Animal
4. infinite stream of animals: Stream<Animal>
5. collect first of each occurrence of each animal type
6. solve this without using state outside of the stream

Every requirement has been satisfied except the "outside stream state" one which I believe to be impossible. Despite that, I've shown how you can still solve the problem using a purely functional method (firstFrom()) that is both deterministic and referentially transparent.

The requested getAnimalType() method exists but is completely unnecessary. Normally wouldn't put a type field in an object at all. This is what happens when implementation details bleed into requirements.

I'm looking for a thorough review of the code. How well it communicates the point, technical soundness, readability, cleanliness, and style. The following code works as one file (for convenience of pasting).

package infinitestream;

import org.junit.Assert;
import org.junit.Test;

import java.util.Arrays;
import java.util.List;
import java.util.Objects;
import java.util.Set;
import java.util.stream.Collectors;
import java.util.stream.Stream;

public class InfiniteStreamTest {
@Test
public void testFirstFrom() {

// .-== Use method being tested ==-. //
Set<Animal> setOfAnimals = InfiniteStream
.firstFrom(
infiniteStreamOfAnimals(),
predictedSize()
)
;
// '-== Use method being tested ==-' //

// .-== Test results ==-. //
Assert.assertTrue( setOfAnimals.size() == predictedSize() );

System.out.println( setOfAnimals );

String actual = setOfAnimals.toString();
String expected = "[first HUMAN animal, first DOG animal, first CAT animal]";
String message =
System.lineSeparator() + expected + " <- expected" +
System.lineSeparator() + actual + " <- actual" +
System.lineSeparator()
;
Assert.assertEquals(message, expected, actual);
// '-== Test results ==-' //
}

// Displays
// [first HUMAN animal, first DOG animal, first CAT animal]

// .-== Construct dependencies ==-. //
private int predictedSize() { return AnimalType.values().length; }

private Stream<Animal> infiniteStreamOfAnimals() {
List<Animal> animals = Arrays
.asList(
new Animal( "first", AnimalType.HUMAN ),
new Animal( "first", AnimalType.DOG ),
new Animal( "second", AnimalType.HUMAN ),
new Animal( "first", AnimalType.CAT )
)
;

Stream<Integer> infiniteStreamOfInts = Stream.iterate( 0, i -> i+1 );

Stream<Animal> infiniteStreamOfAnimals = infiniteStreamOfInts
.map(
i->animals.get(
i % animals.size()
)
)
;

return infiniteStreamOfAnimals;
}
// '-== Construct dependencies ==-' //

}

// .-== System being tested ==-. //
class InfiniteStream {
public static <T> Set<T> firstFrom( Stream<T> infiniteStream, int size ) {

infiniteStream
.takeWhile( x->set.size() < size )
.collect(
Collectors.toCollection(
()->set
)
)
;

return set;
}
}
// '-== System being tested ==-' //

// .-== Required constructs ==-. //
enum AnimalType{ HUMAN, DOG, CAT }

class Animal {

public Animal( String tag, AnimalType type ) {
this.tag = tag;
this.type = type;
}

public AnimalType getAnimalType() { return type; } //TODO: remove unused method

public String toString() { return tag + " " + type + " animal"; }

@Override
public int hashCode() { return Objects.hashCode( type ); }

@Override
public boolean equals( Object that ) {
return that != null
&& that.getClass() == this.getClass()
&& ( (Animal) that ).type == this.type
;
}

private String tag;
private AnimalType type;
}
// '-== Required constructs ==-' //


## Unit tests

String actual = setOfAnimals.toString();
String expected = "[first HUMAN animal, first DOG animal, first CAT animal]";


This is broken... ;-)

EDIT: I encountered this problem after I changed the code to return a Collectors.toSet() - your code has a LinkedHashSet so it is deterministic

Still, in general, the issue is that Sets are unordered, and there's no expectation that setOfAnimals will return a toString() in a deterministic order. Depending on the Set implementation, or even some other random factor, the order of the toString may be different. You need to have a deterministic test. Perhaps returning a List instead of a Set? I actually prefer that option as the List can be in stream order. Note that when I ran your tests in Eclipse with the regular run mechanism, it worked, but when I did code-coverage, the order was different, and the tests failed. This is not just an idle suggestion!

## hashCode and equals hacks.

Your code requires that the equals, and hashCode implementation of the stream object conforms to the required contract. I see this as being a "hack". You should have a better mechanism, a more functional mechanism to accomplish this task.

## More generic and functional

I recommend an extractor function that identifies a key value from the stream members, and then a Set of possible key values as inputs. This should be turned in to a generic method.

For example, your firstFrom method should take three parameters, the stream, a function that extracts the "key", and the possible keys to expect:

public static <S, T> Set<S> firstFrom(
final Stream<S> stream,
final Function<S, T> keyExtractor,
final Set<T> keys) {

.....
}


Note that I have renamed the first parameter to be stream instead of infiniteStream ... the code works for either, so why imply it does not?

Also, I have taken in a set of keys, this parameter contains the "universe" of keys to expect/extract from the stream.

In order for the generics to be improved, they should also take in to account the super and sub types that are possible to support. I settled on the generic signatures:

public static <S, T> List<S> firstFrom(
final Stream<? extends S> stream,
final Function<? super S, T> keyExtractor,
final Set<? extends T> keys) {


Note that the stream can now consist of any Animal or sub-type of animal, and the extractor function can extract from any super-type of Animal too. Further, the keys can be specific subtypes of the extracted keys as well.

The full method I propose is:

public static <S, T> List<S> firstFrom(
final Stream<? extends S> stream,
final Function<? super S, T> keyExtractor,
final Set<? extends T> keys) {

// take a copy to avoid mutating the input parameter
final Set<T> remaining = new HashSet<>(keys);

return stream.takeWhile(s -> !remaining.isEmpty())
.filter(s -> remaining.remove(keyExtractor.apply(s)))
.collect(Collectors.toList());
}


There are a few tricks in here. Firstly, the isEmpty() is generally a much better test than size(). By removing values from the remaining keys set we can stop the stream with isEmpty().

Secondly, we filter the stream to get rid of any subsequent encounters of the key. This is a nice way to prepare the data for the collector in the List.

Finally, I use a trick of the Set.remove() function. It returns true if the set is mutated. So, only the first time the key is encountered will it return true. Thus, only the first stream instance is collected.

## Use Case

The use-case for the firstFrom() function now looks like:

    Set<AnimalType> possibles = new HashSet<>();

// .-== Use method being tested ==-. //
List<Animal> setOfAnimals = InfiniteStream.firstFrom(
infiniteStreamOfAnimals(),
Animal::getAnimalType,
possibles);


See how the Animal::getAnimalType is used to extract the type for the stream members? This beats overriding the hashCode and equals!

Also, the possibles contains the different values in the AnimalType enumeration.

## Full code.

I moved the instances of the animals to the top in to static final values (this was because I messed around with how to get the test assertions right before I settled on the deterministic List output....). I think it is neater now too.

There are other formatting issues, but I've neatened the code with Eclipse's Ctrl-A and Ctrl-F, and then pasted the code here:

import java.util.Arrays;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.function.Function;
import java.util.stream.Collectors;
import java.util.stream.Stream;

import org.junit.Assert;
import org.junit.Test;

public class InfiniteStreamTest {

private static final Animal fHuman = new Animal("first", AnimalType.HUMAN);
private static final Animal fDog = new Animal("first", AnimalType.DOG);
private static final Animal sHuman = new Animal("second", AnimalType.HUMAN);
private static final Animal fCat = new Animal("first", AnimalType.CAT);
private static final List<Animal> animals = Arrays.asList(
fHuman, fDog, sHuman, fCat);

@Test
public void testFirstFrom() {

Set<AnimalType> possibles = new HashSet<>();

// .-== Use method being tested ==-. //
List<Animal> setOfAnimals = InfiniteStream.firstFrom(
infiniteStreamOfAnimals(),
Animal::getAnimalType,
possibles);
// '-== Use method being tested ==-' //

// .-== Test results ==-. //
Assert.assertTrue(setOfAnimals.size() == predictedSize());

System.out.println(setOfAnimals);

String actual = setOfAnimals.toString();
String expected = "[first HUMAN animal, first DOG animal, first CAT animal]";
String message =
System.lineSeparator() + expected + " <- expected" +
System.lineSeparator() + actual + " <- actual" +
System.lineSeparator()
;
Assert.assertEquals(message, expected, actual);
// '-== Test results ==-' //
}

// Displays
// [first HUMAN animal, first DOG animal, first CAT animal]

// .-== Construct dependencies ==-. //
private int predictedSize() {
return AnimalType.values().length;
}

private Stream<Animal> infiniteStreamOfAnimals() {
Stream<Integer> infiniteStreamOfInts = Stream.iterate(0, i -> i + 1);

Stream<Animal> infiniteStreamOfAnimals = infiniteStreamOfInts.map(
i -> animals.get(i % animals.size()));

return infiniteStreamOfAnimals;
}
// '-== Construct dependencies ==-' //

}

// .-== System being tested ==-. //
class InfiniteStream {

public static <S, T> List<S> firstFrom(
final Stream<? extends S> stream,
final Function<? super S, T> keyExtractor,
final Set<? extends T> keys) {

// take a copy to avoid mutating the input parameter
final Set<T> remaining = new HashSet<>(keys);

return stream.takeWhile(s -> !remaining.isEmpty())
.filter(s -> remaining.remove(keyExtractor.apply(s)))
.collect(Collectors.toList());
}

}
// '-== System being tested ==-' //

// .-== Required constructs ==-. //
enum AnimalType {
HUMAN, DOG, CAT
}

class Animal {

public Animal(String tag, AnimalType type) {
this.tag = tag;
this.type = type;
}

public AnimalType getAnimalType() {
return type;
}

public String toString() {
return tag + " " + type + " animal";
}

private String tag;
private AnimalType type;
}
// '-== Required constructs ==-' //

• @Koekje - neither the OP's code nor mine will work with a parallel stream. The access/mutation to the out-of-stream state in the remaining in my code would need to be synchronized, and with a parallel stream the List collected would have a non-deterministic order. So, no, it won't work with a parallel stream. – rolfl Apr 23 '18 at 19:55
• Hmm, I posted too quick and wanted to change it up. But geez, that was a quick response :) As a different question, should the method perhaps check if the stream is ordered? Because else, 'first' would not seem to make much sense. – Koekje Apr 23 '18 at 19:57
• @Koekje - that's a good point, and you would get my upvote if you put that in an answer. – rolfl Apr 23 '18 at 19:59
• I very much appreciate this thorough answer and want to deal with all your good points, but at the moment I'm going crazy trying to reproduce your code coverage test that destroys the ordering. I'm using EclEmma 3.0.1.201711142128. I've ran it over and over with the Alt+Shift+E, T in Eclipse Oxygen.2 Release (4.7.2) (with JUnit 5 as my test runner) and it always works fine. What are you doing? – candied_orange Apr 23 '18 at 21:19
• Give me a second, I'll reproduce it, and we can discuss in the 2nd Monitor – rolfl Apr 23 '18 at 21:21

I find your solution a bit convoluted and rather confusing, for several reasons:

• You are relying on the collector returned by Collectors.toSet() to have the characteristic IDENTITY_FINISH, which, while likely, is only an implementation detail and not guaranteed by the specification of the method.

To explain: The interface Collector defines three type parameters, T, A and R. T is the type of the elements that are collected – in your case, this would be Animal. R is the result type of the collection – a Set<Animal> in your case. A is the accumulation type, i.e. the type of the container that accumulates the elements while they are collected. However, the accumulation type of the collector returned by Collectors.toCollection(Supplier<C>) is ?, meaning that it could be anything. For example, the collector could first collect the animals into a List, and only when all the animals from the stream have been collected into the List, it will create a Set from this List and return the Set. Even if the accumulation type is Set<Integer>, the accumulation container does not necessarily have to be the Set that will be returned.

Now, if a collector has the characteristic IDENTITY_FINISH, then it means that the accumulation container is identical to the result container. Your code only works if the accumulation container is also the result container, because it depends on the size of the result container to grow as the elements from the stream are collected.

• The operation you perform on the stream depends on state that might change during the execution of the operation, which can be dangerous and produces headaches. For all your method firstFrom(Stream<T>, int) knows, the Stream<Animal> might be parallel (even if it is ordered), so the collection operation might be executed in several threads. If this is the case, then every thread will first collect the elements it operates on to its own Set, and only when the threads are done will the results of the parallel collections be merged into the final result container. Oops, the threads will never be done collecting elements, because the size of the final result container never changes.

Interestingly, when I tried a similar approach for collecting the first $n$ integers (distinct or not) from an infinite integer stream to a List, the program did terminate, but the list was much larger than the specified maximum when the stream was parallel.

Of course, it is possible to check whether the collector has the characteristic IDENTITY_FINISH and to manually ascertain that the stream is sequential, but then I would ask the question: Why bother with such a complicated design, when you can just call iterator() on the stream and iterate over the elements in an old-fashioned, but simple and straightforward way? This would be much more to the point, and the risk of headaches would be minimal.

stream.spliterator().hasCharacteristics(Spliterator.ORDERED)

You could throw an IllegalArgumentException when the given stream is not ordered, for example.