DoFnLifecycleManager { public <InputT, OutputT> DoFn<InputT, OutputT> get() throws Exception { Thread currentThread = Thread.currentThread(); return (DoFn<InputT, OutputT>) outstanding.get(currentThread); } private  DoFnLifecycleManager(DoFn<?, ?> original); static DoFnLifecycleManager of(DoFn<?, ?> original); DoFn<InputT, OutputT> get(); void remove(); Collection<Exception> removeAll();  }
@Test public void setupOnGet() throws Exception { TestFn obtained = (TestFn) mgr.get(); assertThat(obtained, not(theInstance(fn))); assertThat(obtained.setupCalled, is(true)); assertThat(obtained.teardownCalled, is(false)); }
@Test public void getMultipleCallsSingleSetupCall() throws Exception { TestFn obtained = (TestFn) mgr.get(); TestFn secondObtained = (TestFn) mgr.get(); assertThat(obtained, theInstance(secondObtained)); assertThat(obtained.setupCalled, is(true)); assertThat(obtained.teardownCalled, is(false)); }
@Test public void getMultipleThreadsDifferentInstances() throws Exception { CountDownLatch startSignal = new CountDownLatch(1); ExecutorService executor = Executors.newCachedThreadPool(); List<Future<TestFn>> futures = new ArrayList<>(); for (int i = 0; i < 10; i++) { futures.add(executor.submit(new GetFnCallable(mgr, startSignal))); } startSignal.countDown(); List<TestFn> fns = new ArrayList<>(); for (Future<TestFn> future : futures) { fns.add(future.get(1L, TimeUnit.SECONDS)); } for (TestFn fn : fns) { assertThat(fn.setupCalled, is(true)); int sameInstances = 0; for (TestFn otherFn : fns) { if (otherFn == fn) { sameInstances++; } } assertThat(sameInstances, equalTo(1)); } }