public synchronized Row poll(final Duration timeout) {
    if (closed || failed) {
      return null;
    }
    final long timeoutNs = timeout.toNanos();
    final long end;
    long remainingTime;
    if (timeoutNs > 0) {
      end = System.nanoTime() + timeoutNs;
      remainingTime = timeoutNs;
    } else {
      end = Long.MAX_VALUE;
      remainingTime = Long.MAX_VALUE;
    }
    do {
      // Poll in smaller units so we can exit on close
      final long pollTime = Math.min(remainingTime, MAX_POLL_NANOS);
      try {
        final Row row = queue.poll(pollTime, TimeUnit.NANOSECONDS);
        if (row != null) {
          tokens.decrementAndGet();
          context.runOnContext(v -> checkRequestTokens());
          return row;
        } else if (complete) {
          // If complete, close once the queue has been emptied
          close();
        }
      } catch (InterruptedException e) {
        return null;
      }
      remainingTime = end - System.nanoTime();
    } while (!closed && !failed && remainingTime > 0);
    return null;
  }
@Test
  public void shouldSetError() {
    publisher = new BufferedPublisher<Row>(context, ImmutableList.of()) {
      @Override
      protected void maybeSend() {
        sendError(new RuntimeException("Error!"));
      }
    };

    publisher.subscribe(pollableSubscriber);

    Row row = pollableSubscriber.poll(POLL_DURATION);
    assertThat(row, is(nullValue()));
    assertThat(throwable, is(notNullValue()));
    assertThat(throwable.getMessage(), is("Error!"));
  }