Prevent Spring AOP from closing my Stream?

Question

I'm using the following construction (simplified):

@Component
class RuleProvider {
    public Stream<String> getRules() {
        return Stream.of("r1", "r2", "r3");
    }
}

@Service
class RuleService {
    @Autowired RuleProvider ruleProvider;
    public void evaluateRules() {
        ruleProvider.getRules().foreach(System.out::println);
    }
}

and I use Spring AOP to perform logging. Now I'd like to log all the rules that are handed to the service. In general, this should do it:

@Aspect
class LoggingAspect {
    @AfterReturning(value="execution(* *..RuleProvider.getRules(*))",
                    returning="rules")
    void logRules(JoinPoint jp, Stream<String> rules) {
        Logger logger = LoggerFactory.getLogger(jp.getTarget().getClass());
        rules.peek(rule -> logger.debug("evaluating {}", rule);
    }
}

This should work - the aspect is evaluated, registers the peek intermediate operation, and once the foreach() terminal op is performed, the logging should also to done.

However, when I run this, the Stream seems to get closed after the aspect has run - I get a "stream has already been operated upon or closed" in the service method.

Why is the Stream being closed? Is there anything I can do against that?

Answer 1

The stream is not closed, but it obviously "has already been operated on" exactly as the error message says. Consider how peek is normally used:

Stream.of(...)
    .peek(...)
    .other_operations(...);

Note that the other_operations() are acting on the Stream returned by peek(), and not on the original stream. What you are doing with the aspect is analogous to this:

Stream s = Stream.of(...);
s.peek(...);
s.other_operations(...);

What you can do to fix this is to replace the return value, e.g. by using an @Around aspect.