WindmillStateInternals implements StateInternals { @Override public <T extends State> T state(StateNamespace namespace, StateTag<T> address) { return workItemState.get(namespace, address, StateContexts.nullContext()); } WindmillStateInternals(
      @Nullable K key,
      String stateFamily,
      WindmillStateReader reader,
      boolean isNewKey,
      WindmillStateCache.ForKey cache,
      Supplier<Closeable> scopedReadStateSupplier); @Override @Nullable K getKey(); void persist(final Windmill.WorkItemCommitRequest.Builder commitBuilder); @Override T state(StateNamespace namespace, StateTag<T> address); @Override T state(
      StateNamespace namespace, StateTag<T> address, StateContext<?> c);  }
@Test public void testBagAddBeforeRead() throws Exception { StateTag<BagState<String>> addr = StateTags.bag("bag", StringUtf8Coder.of()); BagState<String> bag = underTest.state(NAMESPACE, addr); SettableFuture<Iterable<String>> future = SettableFuture.create(); when(mockReader.bagFuture(key(NAMESPACE, "bag"), STATE_FAMILY, StringUtf8Coder.of())) .thenReturn(future); bag.readLater(); bag.add("hello"); waitAndSet(future, Arrays.asList("world"), 200); assertThat(bag.read(), Matchers.containsInAnyOrder("hello", "world")); bag.add("goodbye"); assertThat(bag.read(), Matchers.containsInAnyOrder("hello", "world", "goodbye")); }
@Test public void testBagClearBeforeRead() throws Exception { StateTag<BagState<String>> addr = StateTags.bag("bag", StringUtf8Coder.of()); BagState<String> bag = underTest.state(NAMESPACE, addr); bag.clear(); bag.add("hello"); assertThat(bag.read(), Matchers.containsInAnyOrder("hello")); Mockito.verifyZeroInteractions(mockReader); }
@Test public void testBagIsEmptyFalse() throws Exception { StateTag<BagState<String>> addr = StateTags.bag("bag", StringUtf8Coder.of()); BagState<String> bag = underTest.state(NAMESPACE, addr); SettableFuture<Iterable<String>> future = SettableFuture.create(); when(mockReader.bagFuture(key(NAMESPACE, "bag"), STATE_FAMILY, StringUtf8Coder.of())) .thenReturn(future); ReadableState<Boolean> result = bag.isEmpty().readLater(); Mockito.verify(mockReader).bagFuture(key(NAMESPACE, "bag"), STATE_FAMILY, StringUtf8Coder.of()); waitAndSet(future, Arrays.asList("world"), 200); assertThat(result.read(), Matchers.is(false)); }
@Test public void testBagIsEmptyTrue() throws Exception { StateTag<BagState<String>> addr = StateTags.bag("bag", StringUtf8Coder.of()); BagState<String> bag = underTest.state(NAMESPACE, addr); SettableFuture<Iterable<String>> future = SettableFuture.create(); when(mockReader.bagFuture(key(NAMESPACE, "bag"), STATE_FAMILY, StringUtf8Coder.of())) .thenReturn(future); ReadableState<Boolean> result = bag.isEmpty().readLater(); Mockito.verify(mockReader).bagFuture(key(NAMESPACE, "bag"), STATE_FAMILY, StringUtf8Coder.of()); waitAndSet(future, Arrays.<String>asList(), 200); assertThat(result.read(), Matchers.is(true)); }
@Test public void testBagIsEmptyAfterClear() throws Exception { StateTag<BagState<String>> addr = StateTags.bag("bag", StringUtf8Coder.of()); BagState<String> bag = underTest.state(NAMESPACE, addr); bag.clear(); ReadableState<Boolean> result = bag.isEmpty(); Mockito.verify(mockReader, never()) .bagFuture(key(NAMESPACE, "bag"), STATE_FAMILY, StringUtf8Coder.of()); assertThat(result.read(), Matchers.is(true)); bag.add("hello"); assertThat(result.read(), Matchers.is(false)); }
@Test public void testNewBagNoFetch() throws Exception { StateTag<BagState<String>> addr = StateTags.bag("bag", StringUtf8Coder.of()); BagState<String> bag = underTestNewKey.state(NAMESPACE, addr); assertThat(bag.read(), Matchers.emptyIterable()); Mockito.verifyZeroInteractions(mockReader); }
@Test @SuppressWarnings("ArraysAsListPrimitiveArray") public void testCombiningAddBeforeRead() throws Exception { GroupingState<Integer, Integer> value = underTest.state(NAMESPACE, COMBINING_ADDR); SettableFuture<Iterable<int[]>> future = SettableFuture.create(); when(mockReader.bagFuture(eq(COMBINING_KEY), eq(STATE_FAMILY), Mockito.<Coder<int[]>>any())) .thenReturn(future); value.readLater(); value.add(5); value.add(6); waitAndSet(future, Arrays.asList(new int[] {8}, new int[] {10}), 200); assertThat(value.read(), Matchers.equalTo(29)); future.set(Arrays.asList(new int[] {29})); value.add(2); assertThat(value.read(), Matchers.equalTo(31)); }
@Test public void testCombiningClearBeforeRead() throws Exception { GroupingState<Integer, Integer> value = underTest.state(NAMESPACE, COMBINING_ADDR); value.clear(); value.readLater(); value.add(5); value.add(6); assertThat(value.read(), Matchers.equalTo(11)); value.add(2); assertThat(value.read(), Matchers.equalTo(13)); Mockito.verifyZeroInteractions(mockReader); }
@Test @SuppressWarnings("ArraysAsListPrimitiveArray") public void testCombiningIsEmpty() throws Exception { GroupingState<Integer, Integer> value = underTest.state(NAMESPACE, COMBINING_ADDR); SettableFuture<Iterable<int[]>> future = SettableFuture.create(); when(mockReader.bagFuture(eq(COMBINING_KEY), eq(STATE_FAMILY), Mockito.<Coder<int[]>>any())) .thenReturn(future); ReadableState<Boolean> result = value.isEmpty().readLater(); ArgumentCaptor<ByteString> byteString = ArgumentCaptor.forClass(ByteString.class); Mockito.verify(mockReader) .bagFuture(byteString.capture(), eq(STATE_FAMILY), Mockito.<Coder<int[]>>any()); assertThat(byteString.getValue(), byteStringEq(COMBINING_KEY)); waitAndSet(future, Arrays.asList(new int[] {29}), 200); assertThat(result.read(), Matchers.is(false)); }
@Test public void testCombiningIsEmptyAfterClear() throws Exception { GroupingState<Integer, Integer> value = underTest.state(NAMESPACE, COMBINING_ADDR); value.clear(); ReadableState<Boolean> result = value.isEmpty(); Mockito.verify(mockReader, never()).bagFuture(COMBINING_KEY, STATE_FAMILY, accumCoder); assertThat(result.read(), Matchers.is(true)); value.add(87); assertThat(result.read(), Matchers.is(false)); }
@Test public void testNewCombiningNoFetch() throws Exception { GroupingState<Integer, Integer> value = underTestNewKey.state(NAMESPACE, COMBINING_ADDR); assertThat(value.isEmpty().read(), Matchers.is(true)); assertThat(value.read(), Matchers.is(Sum.ofIntegers().identity())); assertThat(value.isEmpty().read(), Matchers.is(false)); Mockito.verifyZeroInteractions(mockReader); }
@Test public void testWatermarkAddBeforeReadEarliest() throws Exception { StateTag<WatermarkHoldState> addr = StateTags.watermarkStateInternal("watermark", TimestampCombiner.EARLIEST); WatermarkHoldState bag = underTest.state(NAMESPACE, addr); SettableFuture<Instant> future = SettableFuture.create(); when(mockReader.watermarkFuture(key(NAMESPACE, "watermark"), STATE_FAMILY)).thenReturn(future); bag.readLater(); bag.add(new Instant(3000)); waitAndSet(future, new Instant(2000), 200); assertThat(bag.read(), Matchers.equalTo(new Instant(2000))); Mockito.verify(mockReader, times(2)).watermarkFuture(key(NAMESPACE, "watermark"), STATE_FAMILY); Mockito.verifyNoMoreInteractions(mockReader); bag.add(new Instant(1000)); assertThat(bag.read(), Matchers.equalTo(new Instant(1000))); Mockito.verifyNoMoreInteractions(mockReader); }
@Test public void testWatermarkAddBeforeReadLatest() throws Exception { StateTag<WatermarkHoldState> addr = StateTags.watermarkStateInternal("watermark", TimestampCombiner.LATEST); WatermarkHoldState bag = underTest.state(NAMESPACE, addr); SettableFuture<Instant> future = SettableFuture.create(); when(mockReader.watermarkFuture(key(NAMESPACE, "watermark"), STATE_FAMILY)).thenReturn(future); bag.readLater(); bag.add(new Instant(3000)); waitAndSet(future, new Instant(2000), 200); assertThat(bag.read(), Matchers.equalTo(new Instant(3000))); Mockito.verify(mockReader, times(2)).watermarkFuture(key(NAMESPACE, "watermark"), STATE_FAMILY); Mockito.verifyNoMoreInteractions(mockReader); bag.add(new Instant(3000)); assertThat(bag.read(), Matchers.equalTo(new Instant(3000))); Mockito.verifyNoMoreInteractions(mockReader); }
@Test public void testWatermarkAddBeforeReadEndOfWindow() throws Exception { StateTag<WatermarkHoldState> addr = StateTags.watermarkStateInternal("watermark", TimestampCombiner.END_OF_WINDOW); WatermarkHoldState bag = underTest.state(NAMESPACE, addr); SettableFuture<Instant> future = SettableFuture.create(); when(mockReader.watermarkFuture(key(NAMESPACE, "watermark"), STATE_FAMILY)).thenReturn(future); bag.readLater(); bag.add(new Instant(3000)); waitAndSet(future, new Instant(3000), 200); assertThat(bag.read(), Matchers.equalTo(new Instant(3000))); Mockito.verify(mockReader, times(2)).watermarkFuture(key(NAMESPACE, "watermark"), STATE_FAMILY); Mockito.verifyNoMoreInteractions(mockReader); bag.add(new Instant(3000)); assertThat(bag.read(), Matchers.equalTo(new Instant(3000))); Mockito.verifyNoMoreInteractions(mockReader); }
@Test public void testWatermarkClearBeforeRead() throws Exception { StateTag<WatermarkHoldState> addr = StateTags.watermarkStateInternal("watermark", TimestampCombiner.EARLIEST); WatermarkHoldState bag = underTest.state(NAMESPACE, addr); bag.clear(); assertThat(bag.read(), Matchers.nullValue()); bag.add(new Instant(300)); assertThat(bag.read(), Matchers.equalTo(new Instant(300))); Mockito.verifyNoMoreInteractions(mockReader); }
@Test public void testNewWatermarkNoFetch() throws Exception { StateTag<WatermarkHoldState> addr = StateTags.watermarkStateInternal("watermark", TimestampCombiner.EARLIEST); WatermarkHoldState bag = underTestNewKey.state(NAMESPACE, addr); assertThat(bag.read(), Matchers.nullValue()); Mockito.verifyZeroInteractions(mockReader); }
@Test public void testValueSetBeforeRead() throws Exception { StateTag<ValueState<String>> addr = StateTags.value("value", StringUtf8Coder.of()); ValueState<String> value = underTest.state(NAMESPACE, addr); value.write("Hello"); assertEquals("Hello", value.read()); Mockito.verifyNoMoreInteractions(mockReader); }
@Test public void testValueClearBeforeRead() throws Exception { StateTag<ValueState<String>> addr = StateTags.value("value", StringUtf8Coder.of()); ValueState<String> value = underTest.state(NAMESPACE, addr); value.clear(); assertEquals(null, value.read()); Mockito.verifyNoMoreInteractions(mockReader); }
@Test public void testValueRead() throws Exception { StateTag<ValueState<String>> addr = StateTags.value("value", StringUtf8Coder.of()); ValueState<String> value = underTest.state(NAMESPACE, addr); SettableFuture<String> future = SettableFuture.create(); when(mockReader.valueFuture(key(NAMESPACE, "value"), STATE_FAMILY, StringUtf8Coder.of())) .thenReturn(future); waitAndSet(future, "World", 200); assertEquals("World", value.read()); }
@Test public void testNewValueNoFetch() throws Exception { StateTag<ValueState<String>> addr = StateTags.value("value", StringUtf8Coder.of()); ValueState<String> value = underTestNewKey.state(NAMESPACE, addr); assertEquals(null, value.read()); Mockito.verifyZeroInteractions(mockReader); }