### Question:
Adapters { public static <T, R> Consumer<T> toSynchronousConsumer(Function<? super Flux<T>, ? extends Publisher<R>> sender) { return t -> Flux.just(t).publish(sender).ignoreElements().subscribeWith(new BlockableTerminationSubscriber<>()).block(); } private  Adapters(); static Acknowledgeable<T> toLoggingAcknowledgeable(T t); static Subscriber<Acknowledgeable<T>> toAcknowledgeableSubscriber(Subscriber<T> subscriber); static Subscriber<T> toSendingSubscriber(Consumer<? super Flux<T>> sender); static Subscriber<T> toSubscriber(Consumer<T> consumer); static Consumer<T> toConsumer(SubscriberFactory<T> subscriberFactory); static Consumer<T> toSendingConsumerPerThread(Consumer<? super Flux<T>> sender); static Consumer<T> toSendingConsumer(Consumer<? super Flux<T>> sender); static Consumer<T> toSynchronousConsumer(Function<? super Flux<T>, ? extends Publisher<R>> sender); static Consumer<T> toSynchronousConsumer(Function<? super Flux<T>, ? extends Publisher<R>> sender, Duration timeout);  }### Answer:
@Test public void sendingCanBeMadeSynchronous() { AtomicBoolean terminated = new AtomicBoolean(false); Consumer<String> consumer = Adapters.toSynchronousConsumer(flux -> flux.subscribeOn(Schedulers.parallel()).doOnComplete(() -> terminated.set(true))); consumer.accept("Hello"); assertTrue(terminated.get()); }
@Test public void synchronousRuntimeExceptionsArePropagatedFromConsumer() { Consumer<String> consumer = Adapters.toSynchronousConsumer(flux -> flux.subscribeOn(Schedulers.parallel()).handle((string, sink) -> sink.error(new IllegalArgumentException()))); try { consumer.accept("Hello"); fail("Consumer should have thrown an Exception"); } catch (Exception e) { assertTrue(e instanceof IllegalArgumentException); } }
@Test public void synchronousCheckedExceptionsArePropagatedFromConsumer() { Consumer<String> consumer = Adapters.toSynchronousConsumer(flux -> flux.subscribeOn(Schedulers.parallel()).handle((string, sink) -> sink.error(new IOException()))); try { consumer.accept("Hello"); fail("Consumer should have thrown an Exception"); } catch (Exception e) { assertTrue(e instanceof RuntimeException); assertTrue(e.getCause() instanceof IOException); } }
@Test public void synchronousConsumptionTimeoutsThrowException() { Consumer<String> consumer = Adapters.toSynchronousConsumer(flux -> Flux.never().subscribeOn(Schedulers.parallel()), Duration.ofMillis(100L)); try { consumer.accept("Hello"); fail("Consumer should have thrown an Exception"); } catch (Exception e) { assertTrue(e instanceof IllegalStateException); } }
@Test public void synchronousConsumerDoesNotDeadlockOnFailureWithRetry() { Adapters.toSynchronousConsumer(this::handleWithErrorAndRetry).accept("Hello"); }