WindmillStateReader { public <T> Future<T> valueFuture(ByteString encodedTag, String stateFamily, Coder<T> coder) { return stateFuture(new StateTag(StateTag.Kind.VALUE, encodedTag, stateFamily), coder); } WindmillStateReader(
      MetricTrackingWindmillServerStub server,
      String computation,
      ByteString key,
      long shardingKey,
      long workToken); Future<Instant> watermarkFuture(ByteString encodedTag, String stateFamily); Future<T> valueFuture(ByteString encodedTag, String stateFamily, Coder<T> coder); Future<Iterable<T>> bagFuture(
      ByteString encodedTag, String stateFamily, Coder<T> elemCoder); void startBatchAndBlock(); long getBytesRead(); static final long MAX_BAG_BYTES; static final long MAX_KEY_BYTES; }
@Test public void testReadValue() throws Exception { Future<Integer> future = underTest.valueFuture(STATE_KEY_1, STATE_FAMILY, INT_CODER); Mockito.verifyNoMoreInteractions(mockWindmill); Windmill.KeyedGetDataRequest.Builder expectedRequest = Windmill.KeyedGetDataRequest.newBuilder() .setKey(DATA_KEY) .setShardingKey(SHARDING_KEY) .setWorkToken(WORK_TOKEN) .setMaxBytes(WindmillStateReader.MAX_KEY_BYTES) .addValuesToFetch( Windmill.TagValue.newBuilder() .setTag(STATE_KEY_1) .setStateFamily(STATE_FAMILY) .build()); Windmill.KeyedGetDataResponse.Builder response = Windmill.KeyedGetDataResponse.newBuilder() .setKey(DATA_KEY) .addValues( Windmill.TagValue.newBuilder() .setTag(STATE_KEY_1) .setStateFamily(STATE_FAMILY) .setValue(intValue(8))); Mockito.when(mockWindmill.getStateData(COMPUTATION, expectedRequest.build())) .thenReturn(response.build()); Integer result = future.get(); Mockito.verify(mockWindmill).getStateData(COMPUTATION, expectedRequest.build()); Mockito.verifyNoMoreInteractions(mockWindmill); assertThat(result, Matchers.equalTo(8)); assertNoReader(future); }
@Test public void testNoStateFamily() throws Exception { Future<Integer> future = underTest.valueFuture(STATE_KEY_1, "", INT_CODER); Mockito.verifyNoMoreInteractions(mockWindmill); Windmill.KeyedGetDataRequest.Builder expectedRequest = Windmill.KeyedGetDataRequest.newBuilder() .setKey(DATA_KEY) .setShardingKey(SHARDING_KEY) .setMaxBytes(WindmillStateReader.MAX_KEY_BYTES) .setWorkToken(WORK_TOKEN) .addValuesToFetch( Windmill.TagValue.newBuilder().setTag(STATE_KEY_1).setStateFamily("").build()); Windmill.KeyedGetDataResponse.Builder response = Windmill.KeyedGetDataResponse.newBuilder() .setKey(DATA_KEY) .addValues( Windmill.TagValue.newBuilder() .setTag(STATE_KEY_1) .setStateFamily("") .setValue(intValue(8))); Mockito.when(mockWindmill.getStateData(COMPUTATION, expectedRequest.build())) .thenReturn(response.build()); Integer result = future.get(); Mockito.verify(mockWindmill).getStateData(COMPUTATION, expectedRequest.build()); Mockito.verifyNoMoreInteractions(mockWindmill); assertThat(result, Matchers.equalTo(8)); assertNoReader(future); }