@Test public void when_many_subscribers_pull_atoms_on_the_same_address__fetch_atoms_action_should_be_emitted_once() throws Exception { RadixNetworkController controller = mock(RadixNetworkController.class); RadixAddress address = mock(RadixAddress.class); RadixAtomPuller radixAtomPuller = new RadixAtomPuller(controller); List<TestObserver<?>> observers = Stream.iterate(TestObserver.create(), t -> TestObserver.create()).limit(10) .collect(Collectors.toList()); List<Disposable> disposables = observers.stream() .map(observer -> radixAtomPuller.pull(address).subscribe()) .collect(Collectors.toList()); verify(controller, times(1)).dispatch(argThat(t -> { if (!(t instanceof FetchAtomsRequestAction)) { return false; } FetchAtomsRequestAction fetchAtomsAction = (FetchAtomsRequestAction) t; return fetchAtomsAction.getAddress().equals(address); })); disposables.forEach(Disposable::dispose); }
@Override public Observable<Object> pull(RadixAddress address) { return cache.computeIfAbsent( address, addr -> Observable.create(emitter -> { final FetchAtomsAction initialAction = FetchAtomsRequestAction.newRequest(addr); emitter.setCancellable(() -> { controller.dispatch(FetchAtomsCancelAction.of(initialAction.getUuid(), initialAction.getAddress())); }); controller.dispatch(initialAction); }) .publish() .refCount() ); }
RadixAtomPuller implements AtomPuller { @Override public Observable<Object> pull(RadixAddress address) { return cache.computeIfAbsent( address, addr -> Observable.create(emitter -> { final FetchAtomsAction initialAction = FetchAtomsRequestAction.newRequest(addr); emitter.setCancellable(() -> { controller.dispatch(FetchAtomsCancelAction.of(initialAction.getUuid(), initialAction.getAddress())); }); controller.dispatch(initialAction); }) .publish() .refCount() ); } }
RadixAtomPuller implements AtomPuller { @Override public Observable<Object> pull(RadixAddress address) { return cache.computeIfAbsent( address, addr -> Observable.create(emitter -> { final FetchAtomsAction initialAction = FetchAtomsRequestAction.newRequest(addr); emitter.setCancellable(() -> { controller.dispatch(FetchAtomsCancelAction.of(initialAction.getUuid(), initialAction.getAddress())); }); controller.dispatch(initialAction); }) .publish() .refCount() ); } RadixAtomPuller(RadixNetworkController controller); }
RadixAtomPuller implements AtomPuller { @Override public Observable<Object> pull(RadixAddress address) { return cache.computeIfAbsent( address, addr -> Observable.create(emitter -> { final FetchAtomsAction initialAction = FetchAtomsRequestAction.newRequest(addr); emitter.setCancellable(() -> { controller.dispatch(FetchAtomsCancelAction.of(initialAction.getUuid(), initialAction.getAddress())); }); controller.dispatch(initialAction); }) .publish() .refCount() ); } RadixAtomPuller(RadixNetworkController controller); @Override Observable<Object> pull(RadixAddress address); }
RadixAtomPuller implements AtomPuller { @Override public Observable<Object> pull(RadixAddress address) { return cache.computeIfAbsent( address, addr -> Observable.create(emitter -> { final FetchAtomsAction initialAction = FetchAtomsRequestAction.newRequest(addr); emitter.setCancellable(() -> { controller.dispatch(FetchAtomsCancelAction.of(initialAction.getUuid(), initialAction.getAddress())); }); controller.dispatch(initialAction); }) .publish() .refCount() ); } RadixAtomPuller(RadixNetworkController controller); @Override Observable<Object> pull(RadixAddress address);  }