DoFnSignatures { public static <FnT extends DoFn<?, ?>> DoFnSignature signatureForDoFn(FnT fn) { return getSignature(fn.getClass()); } private  DoFnSignatures(); static DoFnSignature signatureForDoFn(FnT fn); static synchronized DoFnSignature getSignature(Class<FnT> fn); static StateSpec<?> getStateSpecOrThrow(
      StateDeclaration stateDeclaration, DoFn<?, ?> target); static TimerSpec getTimerSpecOrThrow(
      TimerDeclaration timerDeclaration, DoFn<?, ?> target); static TimerSpec getTimerFamilySpecOrThrow(
      TimerFamilyDeclaration timerFamilyDeclaration, DoFn<?, ?> target); static boolean isSplittable(DoFn<?, ?> doFn); static boolean isStateful(DoFn<?, ?> doFn); static boolean usesMapState(DoFn<?, ?> doFn); static boolean usesSetState(DoFn<?, ?> doFn); static boolean usesOrderedListState(DoFn<?, ?> doFn); static boolean usesValueState(DoFn<?, ?> doFn); static boolean usesBagState(DoFn<?, ?> doFn); static boolean usesWatermarkHold(DoFn<?, ?> doFn); static boolean usesTimers(DoFn<?, ?> doFn); static boolean usesState(DoFn<?, ?> doFn); static boolean requiresTimeSortedInput(DoFn<?, ?> doFn);  }
@Test public void testSimpleTimerIdNamedDoFn() throws Exception { class DoFnForTestSimpleTimerIdNamedDoFn extends DoFn<KV<String, Integer>, Long> { @TimerId("foo") private final TimerSpec bizzle = TimerSpecs.timer(TimeDomain.EVENT_TIME); @ProcessElement public void foo(ProcessContext context) {} @OnTimer("foo") public void onFoo() {} } DoFnSignature sig = DoFnSignatures.signatureForDoFn(new DoFnForTestSimpleTimerIdNamedDoFn()); final String timerDeclarationId = TimerDeclaration.PREFIX + "foo"; assertThat(sig.timerDeclarations().size(), equalTo(1)); DoFnSignature.TimerDeclaration decl = sig.timerDeclarations().get(timerDeclarationId); assertThat(decl.id(), equalTo(timerDeclarationId)); assertThat( decl.field(), equalTo(DoFnForTestSimpleTimerIdNamedDoFn.class.getDeclaredField("bizzle"))); }
@Test public void testSimpleStateIdNamedDoFn() throws Exception { class DoFnForTestSimpleStateIdNamedDoFn extends DoFn<KV<String, Integer>, Long> { @StateId("foo") private final StateSpec<ValueState<Integer>> bizzle = StateSpecs.value(VarIntCoder.of()); @ProcessElement public void foo(ProcessContext context) {} } DoFnSignature sig = DoFnSignatures.signatureForDoFn(new DoFnForTestSimpleStateIdNamedDoFn()); assertThat(sig.stateDeclarations().size(), equalTo(1)); DoFnSignature.StateDeclaration decl = sig.stateDeclarations().get("foo"); assertThat(decl.id(), equalTo("foo")); assertThat( decl.field(), equalTo(DoFnForTestSimpleStateIdNamedDoFn.class.getDeclaredField("bizzle"))); assertThat( decl.stateType(), Matchers.<TypeDescriptor<?>>equalTo(new TypeDescriptor<ValueState<Integer>>() {})); }
@Test public void testGenericStatefulDoFn() throws Exception { class DoFnForTestGenericStatefulDoFn<T> extends DoFn<KV<String, T>, Long> { @StateId("foo") private final StateSpec<ValueState<T>> bizzle = null; @ProcessElement public void foo(ProcessContext context) {} } DoFn<KV<String, Integer>, Long> myDoFn = new DoFnForTestGenericStatefulDoFn<Integer>() {}; DoFnSignature sig = DoFnSignatures.signatureForDoFn(myDoFn); assertThat(sig.stateDeclarations().size(), equalTo(1)); DoFnSignature.StateDeclaration decl = sig.stateDeclarations().get("foo"); assertThat(decl.id(), equalTo("foo")); assertThat( decl.field(), equalTo(DoFnForTestGenericStatefulDoFn.class.getDeclaredField("bizzle"))); assertThat( decl.stateType(), Matchers.<TypeDescriptor<?>>equalTo(new TypeDescriptor<ValueState<Integer>>() {})); }