public <T> Future<Iterable<TimestampedValue<T>>> orderedListFuture(
      Range<Long> range, ByteString encodedTag, String stateFamily, Coder<T> elemCoder) {
    // First request has no continuation position.
    StateTag<ByteString> stateTag =
        StateTag.<ByteString>of(StateTag.Kind.ORDERED_LIST, encodedTag, stateFamily)
            .toBuilder()
            .setSortedListRange(Preconditions.checkNotNull(range))
            .build();
    return valuesToPagingIterableFuture(stateTag, elemCoder, this.stateFuture(stateTag, elemCoder));
  }
@Test
  public void testReadSortedList() throws Exception {
    long beginning = SortedListRange.getDefaultInstance().getStart();
    long end = SortedListRange.getDefaultInstance().getLimit();
    Future<Iterable<TimestampedValue<Integer>>> future =
        underTest.orderedListFuture(
            Range.closedOpen(beginning, end), STATE_KEY_1, STATE_FAMILY, INT_CODER);
    Mockito.verifyNoMoreInteractions(mockWindmill);

    // Fetch the entire list.
    Windmill.KeyedGetDataRequest.Builder expectedRequest =
        Windmill.KeyedGetDataRequest.newBuilder()
            .setKey(DATA_KEY)
            .setShardingKey(SHARDING_KEY)
            .setWorkToken(WORK_TOKEN)
            .setMaxBytes(WindmillStateReader.MAX_KEY_BYTES)
            .addSortedListsToFetch(
                Windmill.TagSortedListFetchRequest.newBuilder()
                    .setTag(STATE_KEY_1)
                    .setStateFamily(STATE_FAMILY)
                    .addFetchRanges(SortedListRange.newBuilder().setStart(beginning).setLimit(end))
                    .setFetchMaxBytes(WindmillStateReader.MAX_ORDERED_LIST_BYTES));

    Windmill.KeyedGetDataResponse.Builder response =
        Windmill.KeyedGetDataResponse.newBuilder()
            .setKey(DATA_KEY)
            .addTagSortedLists(
                Windmill.TagSortedListFetchResponse.newBuilder()
                    .setTag(STATE_KEY_1)
                    .setStateFamily(STATE_FAMILY)
                    .addEntries(
                        SortedListEntry.newBuilder().setValue(intData(5)).setSortKey(5000).setId(5))
                    .addEntries(
                        SortedListEntry.newBuilder().setValue(intData(6)).setSortKey(6000).setId(5))
                    .addEntries(
                        SortedListEntry.newBuilder().setValue(intData(7)).setSortKey(7000).setId(7))
                    .addEntries(
                        SortedListEntry.newBuilder().setValue(intData(8)).setSortKey(8000).setId(8))
                    .addFetchRanges(
                        SortedListRange.newBuilder().setStart(beginning).setLimit(end)));

    Mockito.when(mockWindmill.getStateData(COMPUTATION, expectedRequest.build()))
        .thenReturn(response.build());

    Iterable<TimestampedValue<Integer>> results = future.get();
    Mockito.verify(mockWindmill).getStateData(COMPUTATION, expectedRequest.build());
    for (TimestampedValue<Integer> unused : results) {
      // Iterate over the results to force loading all the pages.
    }
    Mockito.verifyNoMoreInteractions(mockWindmill);

    assertThat(
        results,
        Matchers.contains(
            TimestampedValue.of(5, Instant.ofEpochMilli(5)),
            TimestampedValue.of(6, Instant.ofEpochMilli(6)),
            TimestampedValue.of(7, Instant.ofEpochMilli(7)),
            TimestampedValue.of(8, Instant.ofEpochMilli(8))));
    assertNoReader(future);
  }