WindmillKeyedWorkItem implements KeyedWorkItem<K, ElemT> { @Override public Iterable<WindowedValue<ElemT>> elementsIterable() { return FluentIterable.from(workItem.getMessageBundlesList()) .transformAndConcat(Windmill.InputMessageBundle::getMessagesList) .transform( message -> { try { Instant timestamp = WindmillTimeUtils.windmillToHarnessTimestamp(message.getTimestamp()); Collection<? extends BoundedWindow> windows = WindmillSink.decodeMetadataWindows(windowsCoder, message.getMetadata()); PaneInfo pane = WindmillSink.decodeMetadataPane(message.getMetadata()); InputStream inputStream = message.getData().newInput(); ElemT value = valueCoder.decode(inputStream, Coder.Context.OUTER); return WindowedValue.of(value, timestamp, windows, pane); } catch (IOException e) { throw new RuntimeException(e); } }); } WindmillKeyedWorkItem(
      K key,
      Windmill.WorkItem workItem,
      Coder<? extends BoundedWindow> windowCoder,
      Coder<Collection<? extends BoundedWindow>> windowsCoder,
      Coder<ElemT> valueCoder); @Override K key(); @Override Iterable<TimerData> timersIterable(); @Override Iterable<WindowedValue<ElemT>> elementsIterable(); @Override boolean equals(@Nullable Object other); @Override int hashCode(); @Override String toString();  }
@Test public void testElementIteration() throws Exception { Windmill.WorkItem.Builder workItem = Windmill.WorkItem.newBuilder().setKey(SERIALIZED_KEY).setWorkToken(17); Windmill.InputMessageBundle.Builder chunk1 = workItem.addMessageBundlesBuilder(); chunk1.setSourceComputationId("computation"); addElement(chunk1, 5, "hello", WINDOW_1, paneInfo(0)); addElement(chunk1, 7, "world", WINDOW_2, paneInfo(2)); Windmill.InputMessageBundle.Builder chunk2 = workItem.addMessageBundlesBuilder(); chunk2.setSourceComputationId("computation"); addElement(chunk2, 6, "earth", WINDOW_1, paneInfo(1)); KeyedWorkItem<String, String> keyedWorkItem = new WindmillKeyedWorkItem<>( KEY, workItem.build(), WINDOW_CODER, WINDOWS_CODER, VALUE_CODER); assertThat( keyedWorkItem.elementsIterable(), Matchers.contains( WindowedValue.of("hello", new Instant(5), WINDOW_1, paneInfo(0)), WindowedValue.of("world", new Instant(7), WINDOW_2, paneInfo(2)), WindowedValue.of("earth", new Instant(6), WINDOW_1, paneInfo(1)))); }