SspDefinitionStore.java

@Service
@Slf4j
@RequiredArgsConstructor
public class SspDefinitionenStore implements SspDefinitionConsumer {

    private CacheMono<VersionedId, SspDefinition, SspDefinition> sspDefinitionCache;
    private FluxSink<SspDefinition> sspDefinitionSink;

    @PostConstruct
    public void initialize() {
        sspDefinitionCache = CacheMono.fromPublisher(
                Flux.create(sink -> sspDefinitionSink = sink),
                SspDefinition::getId);
    }

    @Override
    public void accept(SspDefinition sspDefinition) {
        sspDefinitionSink.next(sspDefinition);
    }

    public Mono<SspDefinition> lookupSspDefinition(VersionedId sspId) {
        return sspDefinitionCache.lookup(sspId)
                .doOnNext(sspTopology -> log.info(
                    "SspDefinition was found by lookup with key {}", sspId))
                .onErrorResume(err -> {
                    log.error("Error on lookup SspDefinition by key {}, message: {}",
                              sspId, err.getMessage());
                    return Mono.empty();
                });
    }

    public Optional<SspDefinition> findSspDefinition(VersionedId sspId) {
        return sspDefinitionCache.getValue(sspId);
    }

    public Flux<SspDefinition> findSspDefinitions() {
        return Flux.fromStream(sspDefinitionCache.getValues().filter(Objects::nonNull));
    }

    ...
}