Use mapped stream item downstream

Question

I want to do something like this:

servicesList.parallelStream()
    .map(service-> pluginService.fetchSentinelFileAsString(projectModel, branch))
    .filter(Optional::isPresent)
    .map(Optional::get)
    .map(model -> service.fetch(model));

Note that the service parameter in the first lambda is reused in the foreach lambda. This isn't legal but perhaps there's a way to do it.

Let me describe what that code is doing in case it helps:

• I have a list of references to services
• I want to use the service to get an Optional
• If the Optional is present I want to use the service again to process the content of the Optional

Put another way, I want to sorta reuse a value that was mapped in a previous stream item.

I really think of this as being one stream, but obviously each stream operation passes exactly ONE item to the next stage of the pipeline and this exceeds that limitations.

Answer 1

You should simply pass your service in a tuple (AbstractMap.SimpleEntry) to the downstream:

     servicesList.parallelStream()
    .map(service-> new SimpleEntry<>(service, pluginService.fetchSentinelFileAsString(projectModel, branch)))
    .filter(e -> e.getValue().isPresent())
    .map(e -> e.getKey().fetch(e.getValue()));

As you can see, the key of the entry will be the service, and the result of the fetchSentinelFileAsString call will be the value. At least I would do it in this way.

Answer 2

Another solution may be possible based on handling non-empty Optional inside the first map:

servicesList.parallelStream()
    .map(service -> pluginService
        .fetchSentinelFileAsString(projectModel, branch) // Optional
        .map(model -> service.fetch(model))
    )
    .forEach(fetchedModel -> doSomething(fetchedModel));

Or, taking into account that the result of fetchSentinelFileAsString does not depend on the services in the list which should be applied only to non-null result:

pluginService.fetchSentinelFileAsString(projectModel, branch)
    .ifPresent(model -> serviceList.parallelStream()
        .forEach(service -> service.fetch(model))
    );

Answer 3

Your service will only be available inside of your lambda. You can compare this to a parameter only being available inside of the body of a method

In this case, you can leverage Optional's fluent API to directly invoke the service like in the following which will output a Stream<Optional<T>>

servicesList.parallelStream()
            .map(service-> pluginService.fetchSentinelFileAsString(projectModel, branch)
                                        .map(service::fetch)
            )
            ...

There is also the misusage of get() that should be avoided at all costs. You can refer to this answer from Brian Goetz