### Question:
ProgressTracker { public BodyPublisher tracking(BodyPublisher upstream, Listener listener) { requireNonNull(upstream, "upstream"); requireNonNull(listener, "listener"); if (upstream instanceof MultipartBodyPublisher) { return new MultipartTrackingBodyPublisher( options, new MultipartListenerAdapter(listener), (MultipartBodyPublisher) upstream); } return new UnipartTrackingBodyPublisher(options, listener, upstream); } private  ProgressTracker(Builder builder); long bytesTransferredThreshold(); Optional<Duration> timePassedThreshold(); boolean enclosedProgress(); Optional<Executor> executor(); BodyPublisher tracking(BodyPublisher upstream, Listener listener); BodyPublisher trackingMultipart(
      MultipartBodyPublisher upstream, MultipartListener listener); BodySubscriber<T> tracking(
      BodySubscriber<T> downstream, Listener listener, long contentLengthIfKnown); BodyHandler<T> tracking(BodyHandler<T> handler, Listener listener); static ProgressTracker create(); static Builder newBuilder();  }### Answer:
@Test void trackUploadProgressNoThreshold() { int batchSize = 64; int count = 20; var tracker = withVirtualClock().build(); var listener = new TestListener(); var trackedUpstream = tracker.tracking(bodyPublisher(batchSize, count), listener); var downstream = new TestSubscriber<ByteBuffer>(); trackedUpstream.subscribe(downstream); assertBodyForwarded(downstream, batchSize, count); testProgressSequenceNoThreshold(listener, batchSize, count); }
@Test void trackUploadProgressWithError() { var tracker = withExecutor().build(); var listener = new TestListener(); var trackedUpstream = tracker.tracking( BodyPublishers.fromPublisher(new FailedPublisher<>(TestException::new)), listener); var downstream = new TestSubscriber<ByteBuffer>(); upstreamExecutor.execute( () -> trackedUpstream.subscribe(downstream)); downstream.awaitError(); listener.awaitComplete(); assertEquals(1, listener.errors); }
@Test void trackDownloadProgressNoThresholds() { int batchSize = 64; int count = 20; int length = batchSize * count; var tracker = withVirtualClock().build(); var listener = new TestListener(); var downstream = new TestSubscriber<List<ByteBuffer>>(); var trackedDownstream = tracker.tracking( BodySubscribers.fromSubscriber(downstream, d -> countBytes(d.items)), listener, length); var publisher = listPublisher(batchSize, count); publisher.subscribe(trackedDownstream); downstream.awaitComplete(); assertEquals(count, downstream.nexts); assertEquals(length, trackedDownstream.getBody().toCompletableFuture().join()); testProgressSequenceNoThreshold(listener, batchSize, count); }
@Test void trackDownloadProgressWithError() { var tracker = withExecutor().build(); var listener = new TestListener(); var downstream = new TestSubscriber<List<ByteBuffer>>(); var trackedDownstream = tracker.tracking( BodySubscribers.fromSubscriber(downstream), listener, -1); var publisher = new FailedPublisher<List<ByteBuffer>>(TestException::new); upstreamExecutor.execute(() -> publisher.subscribe(trackedDownstream)); downstream.awaitError(); listener.awaitComplete(); assertEquals(1, listener.errors); }