### Question:
RxFailedWorkRecycler implements RxFailureConsumer<W> { @Override public Publisher<?> apply(W work, Throwable error) { if (isRecyclable(work, error)) { return Mono.just(work) .map(toRecycle -> recycle(toRecycle, error)) .doOnNext(recycled -> hookOnRecycle(work, recycled, error)) .flatMapMany(recyclePublisher); } else { return Mono.empty().doOnSubscribe(subscription -> hookOnDrop(work, error)); } } RxFailedWorkRecycler(WorkRecycleConfig config, Function<? super R, ? extends Publisher<?>> recyclePublisher); @Override Publisher<?> apply(W work, Throwable error);  }### Answer:
@Test public void failedWorkIsNotRecycledIfExpired() { TestWork work = TestWork.create(WorkType.INTENT, "SOME/URL", Instant.now().minus(RECYCLE_CONFIG.getRecycleExpiration()).minusMillis(1).toEpochMilli()); Flux.from(recycler.apply(work, new RuntimeException())).blockLast(); assertNull(recycleHolder.get()); }
@Test public void failedWorkIsRecycledUntilMaxCountIsReached() { TestWork work = TestWork.create(WorkType.INTENT, "SOME/URL", Instant.now().toEpochMilli()); for (long recycleCount = 0; recycleCount <= RECYCLE_CONFIG.getMaxRecycleCount(); recycleCount++) { Flux.from(recycler.apply(work, new RuntimeException())).blockLast(); assertNotNull(recycleHolder.get()); work = recycleHolder.get(); } recycleHolder.set(null); Flux.from(recycler.apply(work, new RuntimeException())).blockLast(); assertNull(recycleHolder.get()); }
@Test public void failedWorkIsNotRecycledForUnrecyclableException() { TestWork work = TestWork.create(WorkType.INTENT, "SOME/URL", Instant.now().toEpochMilli()); Flux.from(recycler.apply(work, new IllegalStateException())).blockLast(); assertNull(recycleHolder.get()); Flux.from(recycler.apply(work, new RuntimeException())).blockLast(); assertNotNull(recycleHolder.get()); }