@Test public void when_stored_action_received__a_soft_store_observation_should_be_stored() { InMemoryAtomStore atomStore = mock(InMemoryAtomStore.class); InMemoryAtomStoreReducer reducer = new InMemoryAtomStoreReducer(atomStore); RadixAddress address = mock(RadixAddress.class); final Atom atom = mock(Atom.class); when(atom.addresses()).thenAnswer(inv -> Stream.of(address)); RadixNode node = mock(RadixNode.class); RadixNodeAction action = SubmitAtomStatusAction.fromStatusNotification( "different-id", atom, node, new AtomStatusEvent(AtomStatus.STORED, null) ); reducer.reduce(action); InOrder inOrder = Mockito.inOrder(atomStore, atomStore); inOrder.verify(atomStore).store(eq(address), argThat(o -> o.isStore() && o.getAtom().equals(atom) && o.getUpdateType().isSoft() )); inOrder.verify(atomStore).store(eq(address), argThat(AtomObservation::isHead)); }
public void reduce(RadixNodeAction action) { if (action instanceof FetchAtomsObservationAction) { FetchAtomsObservationAction fetchAtomsObservationAction = (FetchAtomsObservationAction) action; AtomObservation observation = fetchAtomsObservationAction.getObservation(); this.atomStore.store(fetchAtomsObservationAction.getAddress(), observation); } else if (action instanceof SubmitAtomStatusAction) { final SubmitAtomStatusAction submitAtomStatusAction = (SubmitAtomStatusAction) action; final Atom atom = submitAtomStatusAction.getAtom(); if (submitAtomStatusAction.getStatusNotification().getAtomStatus() == AtomStatus.STORED) { final long timestamp = retrieveTimestamp(submitAtomStatusAction.getStatusNotification().getData()); atom.addresses().forEach(address -> { this.atomStore.store(address, AtomObservation.softStored(atom, timestamp)); this.atomStore.store(address, AtomObservation.head()); }); } } }
InMemoryAtomStoreReducer { public void reduce(RadixNodeAction action) { if (action instanceof FetchAtomsObservationAction) { FetchAtomsObservationAction fetchAtomsObservationAction = (FetchAtomsObservationAction) action; AtomObservation observation = fetchAtomsObservationAction.getObservation(); this.atomStore.store(fetchAtomsObservationAction.getAddress(), observation); } else if (action instanceof SubmitAtomStatusAction) { final SubmitAtomStatusAction submitAtomStatusAction = (SubmitAtomStatusAction) action; final Atom atom = submitAtomStatusAction.getAtom(); if (submitAtomStatusAction.getStatusNotification().getAtomStatus() == AtomStatus.STORED) { final long timestamp = retrieveTimestamp(submitAtomStatusAction.getStatusNotification().getData()); atom.addresses().forEach(address -> { this.atomStore.store(address, AtomObservation.softStored(atom, timestamp)); this.atomStore.store(address, AtomObservation.head()); }); } } } }
InMemoryAtomStoreReducer { public void reduce(RadixNodeAction action) { if (action instanceof FetchAtomsObservationAction) { FetchAtomsObservationAction fetchAtomsObservationAction = (FetchAtomsObservationAction) action; AtomObservation observation = fetchAtomsObservationAction.getObservation(); this.atomStore.store(fetchAtomsObservationAction.getAddress(), observation); } else if (action instanceof SubmitAtomStatusAction) { final SubmitAtomStatusAction submitAtomStatusAction = (SubmitAtomStatusAction) action; final Atom atom = submitAtomStatusAction.getAtom(); if (submitAtomStatusAction.getStatusNotification().getAtomStatus() == AtomStatus.STORED) { final long timestamp = retrieveTimestamp(submitAtomStatusAction.getStatusNotification().getData()); atom.addresses().forEach(address -> { this.atomStore.store(address, AtomObservation.softStored(atom, timestamp)); this.atomStore.store(address, AtomObservation.head()); }); } } } InMemoryAtomStoreReducer(InMemoryAtomStore atomStore); }
InMemoryAtomStoreReducer { public void reduce(RadixNodeAction action) { if (action instanceof FetchAtomsObservationAction) { FetchAtomsObservationAction fetchAtomsObservationAction = (FetchAtomsObservationAction) action; AtomObservation observation = fetchAtomsObservationAction.getObservation(); this.atomStore.store(fetchAtomsObservationAction.getAddress(), observation); } else if (action instanceof SubmitAtomStatusAction) { final SubmitAtomStatusAction submitAtomStatusAction = (SubmitAtomStatusAction) action; final Atom atom = submitAtomStatusAction.getAtom(); if (submitAtomStatusAction.getStatusNotification().getAtomStatus() == AtomStatus.STORED) { final long timestamp = retrieveTimestamp(submitAtomStatusAction.getStatusNotification().getData()); atom.addresses().forEach(address -> { this.atomStore.store(address, AtomObservation.softStored(atom, timestamp)); this.atomStore.store(address, AtomObservation.head()); }); } } } InMemoryAtomStoreReducer(InMemoryAtomStore atomStore); void reduce(RadixNodeAction action); }
InMemoryAtomStoreReducer { public void reduce(RadixNodeAction action) { if (action instanceof FetchAtomsObservationAction) { FetchAtomsObservationAction fetchAtomsObservationAction = (FetchAtomsObservationAction) action; AtomObservation observation = fetchAtomsObservationAction.getObservation(); this.atomStore.store(fetchAtomsObservationAction.getAddress(), observation); } else if (action instanceof SubmitAtomStatusAction) { final SubmitAtomStatusAction submitAtomStatusAction = (SubmitAtomStatusAction) action; final Atom atom = submitAtomStatusAction.getAtom(); if (submitAtomStatusAction.getStatusNotification().getAtomStatus() == AtomStatus.STORED) { final long timestamp = retrieveTimestamp(submitAtomStatusAction.getStatusNotification().getData()); atom.addresses().forEach(address -> { this.atomStore.store(address, AtomObservation.softStored(atom, timestamp)); this.atomStore.store(address, AtomObservation.head()); }); } } } InMemoryAtomStoreReducer(InMemoryAtomStore atomStore); void reduce(RadixNodeAction action);  }