### Question:
CryptoCompareService implements CryptoService { public static <T> Flux<T> provideCaching(Flux<T> input) { return input; } CryptoCompareService(); Flux<Map<String, Object>> eventsStream(); static Flux<T> provideResilience(Flux<T> input); static Flux<T> provideCaching(Flux<T> input); static final int CACHE_SIZE; }### Answer:
@Test public void verifyThatSupportMultiSubscribers() { AtomicInteger subscribtions = new AtomicInteger(0); Flux<Object> source = DirectProcessor .create() .doOnSubscribe(s -> subscribtions.incrementAndGet()); Flux<Object> cachedFlux = CryptoCompareService.provideCaching(source); cachedFlux.subscribe(System.out::println); cachedFlux.subscribe(System.out::println); Assert.assertEquals(1, subscribtions.get()); }
@Test public void verifyThatSupportReplayMode() { UnicastProcessor<String> source = UnicastProcessor.create(); ReplayProcessor<String> consumer1 = ReplayProcessor.create(10); ReplayProcessor<String> consumer2 = ReplayProcessor.create(10); Publisher<String> publisher = CryptoCompareService.provideCaching(source); source.onNext("A"); source.onNext("B"); source.onNext("C"); publisher.subscribe(consumer1); source.onNext("D"); source.onNext("E"); source.onNext("F"); publisher.subscribe(consumer2); source.onNext("G"); source.onComplete(); StepVerifier.create(consumer1) .expectSubscription() .expectNext("A", "B", "C", "D", "E", "F", "G") .verifyComplete(); StepVerifier.create(consumer2) .expectSubscription() .expectNext("D", "E", "F", "G") .verifyComplete(); }