KeyedWorkItemCoder extends StructuredCoder<KeyedWorkItem<K, ElemT>> { public static <K, ElemT> KeyedWorkItemCoder<K, ElemT> of( Coder<K> keyCoder, Coder<ElemT> elemCoder, Coder<? extends BoundedWindow> windowCoder) { return new KeyedWorkItemCoder<>(keyCoder, elemCoder, windowCoder); } private  KeyedWorkItemCoder(
      Coder<K> keyCoder, Coder<ElemT> elemCoder, Coder<? extends BoundedWindow> windowCoder); static KeyedWorkItemCoder<K, ElemT> of(
      Coder<K> keyCoder, Coder<ElemT> elemCoder, Coder<? extends BoundedWindow> windowCoder); Coder<K> getKeyCoder(); Coder<ElemT> getElementCoder(); @Override void encode(KeyedWorkItem<K, ElemT> value, OutputStream outStream); @Override KeyedWorkItem<K, ElemT> decode(InputStream inStream); @Override List<? extends Coder<?>> getCoderArguments(); @Override void verifyDeterministic(); @Override boolean consistentWithEquals();  }
@Test public void testCoderProperties() throws Exception { CoderProperties.coderSerializable( KeyedWorkItemCoder.of(StringUtf8Coder.of(), VarIntCoder.of(), GlobalWindow.Coder.INSTANCE)); }
@Test public void testEncodeDecodeEqual() throws Exception { Iterable<TimerData> timers = ImmutableList.of( TimerData.of( StateNamespaces.global(), new Instant(500L), new Instant(500L), TimeDomain.EVENT_TIME)); Iterable<WindowedValue<Integer>> elements = ImmutableList.of( WindowedValue.valueInGlobalWindow(1), WindowedValue.valueInGlobalWindow(4), WindowedValue.valueInGlobalWindow(8)); KeyedWorkItemCoder<String, Integer> coder = KeyedWorkItemCoder.of(StringUtf8Coder.of(), VarIntCoder.of(), GlobalWindow.Coder.INSTANCE); CoderProperties.coderDecodeEncodeEqual(coder, KeyedWorkItems.workItem("foo", timers, elements)); CoderProperties.coderDecodeEncodeEqual(coder, KeyedWorkItems.elementsWorkItem("foo", elements)); CoderProperties.coderDecodeEncodeEqual(coder, KeyedWorkItems.timersWorkItem("foo", timers)); }
@Test public void testCoderIsSerializableWithWellKnownCoderType() throws Exception { CoderProperties.coderSerializable( KeyedWorkItemCoder.of( GlobalWindow.Coder.INSTANCE, GlobalWindow.Coder.INSTANCE, GlobalWindow.Coder.INSTANCE)); }
@Test public void testEncodeDecodeEqual() throws Exception { Iterable<TimerData> timers = ImmutableList.of( TimerData.of(StateNamespaces.global(), new Instant(500L), TimeDomain.EVENT_TIME)); Iterable<WindowedValue<Integer>> elements = ImmutableList.of( WindowedValue.valueInGlobalWindow(1), WindowedValue.valueInGlobalWindow(4), WindowedValue.valueInGlobalWindow(8)); KeyedWorkItemCoder<String, Integer> coder = KeyedWorkItemCoder.of(StringUtf8Coder.of(), VarIntCoder.of(), GlobalWindow.Coder.INSTANCE); CoderProperties.coderDecodeEncodeEqual(coder, KeyedWorkItems.workItem("foo", timers, elements)); CoderProperties.coderDecodeEncodeEqual(coder, KeyedWorkItems.elementsWorkItem("foo", elements)); CoderProperties.coderDecodeEncodeEqual(coder, KeyedWorkItems.timersWorkItem("foo", timers)); }