@Test public void when_getting_up_particles_with_an_empty_store__store_should_return_an_empty_stream() { InMemoryAtomStore inMemoryAtomStore = new InMemoryAtomStore(); Stream<Particle> upParticles = inMemoryAtomStore.getUpParticles(mock(RadixAddress.class), null); assertThat(upParticles).isEmpty(); }
@Override public Stream<Particle> getUpParticles(RadixAddress address, @Nullable String stagedUuid) { synchronized (lock) { List<Particle> upParticles = particleIndex.entrySet().stream() .filter(e -> e.getKey().getShardables().contains(address)) .filter(e -> { final Map<Spin, Set<Atom>> spinParticleIndex = e.getValue(); final boolean hasDown = spinParticleIndex.getOrDefault(Spin.DOWN, Collections.emptySet()) .stream().anyMatch(a -> atoms.get(a).isStore()); if (hasDown) { return false; } if (stagedUuid != null && stagedParticleIndex.getOrDefault(stagedUuid, Collections.emptyMap()).get(e.getKey()) == Spin.DOWN) { return false; } Set<Atom> uppingAtoms = spinParticleIndex.getOrDefault(Spin.UP, Collections.emptySet()); return uppingAtoms.stream().anyMatch(a -> atoms.get(a).isStore()); }) .map(Map.Entry::getKey) .collect(Collectors.toList()); if (stagedUuid != null) { stagedParticleIndex.getOrDefault(stagedUuid, Collections.emptyMap()).entrySet().stream() .filter(e -> e.getValue() == Spin.UP) .map(Entry::getKey) .filter(p -> p.getShardables().contains(address)) .forEach(upParticles::add); } return upParticles.stream(); } }
InMemoryAtomStore implements AtomStore { @Override public Stream<Particle> getUpParticles(RadixAddress address, @Nullable String stagedUuid) { synchronized (lock) { List<Particle> upParticles = particleIndex.entrySet().stream() .filter(e -> e.getKey().getShardables().contains(address)) .filter(e -> { final Map<Spin, Set<Atom>> spinParticleIndex = e.getValue(); final boolean hasDown = spinParticleIndex.getOrDefault(Spin.DOWN, Collections.emptySet()) .stream().anyMatch(a -> atoms.get(a).isStore()); if (hasDown) { return false; } if (stagedUuid != null && stagedParticleIndex.getOrDefault(stagedUuid, Collections.emptyMap()).get(e.getKey()) == Spin.DOWN) { return false; } Set<Atom> uppingAtoms = spinParticleIndex.getOrDefault(Spin.UP, Collections.emptySet()); return uppingAtoms.stream().anyMatch(a -> atoms.get(a).isStore()); }) .map(Map.Entry::getKey) .collect(Collectors.toList()); if (stagedUuid != null) { stagedParticleIndex.getOrDefault(stagedUuid, Collections.emptyMap()).entrySet().stream() .filter(e -> e.getValue() == Spin.UP) .map(Entry::getKey) .filter(p -> p.getShardables().contains(address)) .forEach(upParticles::add); } return upParticles.stream(); } } }
InMemoryAtomStore implements AtomStore { @Override public Stream<Particle> getUpParticles(RadixAddress address, @Nullable String stagedUuid) { synchronized (lock) { List<Particle> upParticles = particleIndex.entrySet().stream() .filter(e -> e.getKey().getShardables().contains(address)) .filter(e -> { final Map<Spin, Set<Atom>> spinParticleIndex = e.getValue(); final boolean hasDown = spinParticleIndex.getOrDefault(Spin.DOWN, Collections.emptySet()) .stream().anyMatch(a -> atoms.get(a).isStore()); if (hasDown) { return false; } if (stagedUuid != null && stagedParticleIndex.getOrDefault(stagedUuid, Collections.emptyMap()).get(e.getKey()) == Spin.DOWN) { return false; } Set<Atom> uppingAtoms = spinParticleIndex.getOrDefault(Spin.UP, Collections.emptySet()); return uppingAtoms.stream().anyMatch(a -> atoms.get(a).isStore()); }) .map(Map.Entry::getKey) .collect(Collectors.toList()); if (stagedUuid != null) { stagedParticleIndex.getOrDefault(stagedUuid, Collections.emptyMap()).entrySet().stream() .filter(e -> e.getValue() == Spin.UP) .map(Entry::getKey) .filter(p -> p.getShardables().contains(address)) .forEach(upParticles::add); } return upParticles.stream(); } }  }
InMemoryAtomStore implements AtomStore { @Override public Stream<Particle> getUpParticles(RadixAddress address, @Nullable String stagedUuid) { synchronized (lock) { List<Particle> upParticles = particleIndex.entrySet().stream() .filter(e -> e.getKey().getShardables().contains(address)) .filter(e -> { final Map<Spin, Set<Atom>> spinParticleIndex = e.getValue(); final boolean hasDown = spinParticleIndex.getOrDefault(Spin.DOWN, Collections.emptySet()) .stream().anyMatch(a -> atoms.get(a).isStore()); if (hasDown) { return false; } if (stagedUuid != null && stagedParticleIndex.getOrDefault(stagedUuid, Collections.emptyMap()).get(e.getKey()) == Spin.DOWN) { return false; } Set<Atom> uppingAtoms = spinParticleIndex.getOrDefault(Spin.UP, Collections.emptySet()); return uppingAtoms.stream().anyMatch(a -> atoms.get(a).isStore()); }) .map(Map.Entry::getKey) .collect(Collectors.toList()); if (stagedUuid != null) { stagedParticleIndex.getOrDefault(stagedUuid, Collections.emptyMap()).entrySet().stream() .filter(e -> e.getValue() == Spin.UP) .map(Entry::getKey) .filter(p -> p.getShardables().contains(address)) .forEach(upParticles::add); } return upParticles.stream(); } }  @Override void stageParticleGroup(String uuid, ParticleGroup particleGroup); @Override List<ParticleGroup> getStaged(String uuid); @Override List<ParticleGroup> getStagedAndClear(String uuid); void store(RadixAddress address, AtomObservation atomObservation); @Override Observable<Long> onSync(RadixAddress address); @Override Stream<Atom> getStoredAtoms(RadixAddress address); @Override Stream<Particle> getUpParticles(RadixAddress address, @Nullable String stagedUuid); @Override Observable<AtomObservation> getAtomObservations(RadixAddress address); }
InMemoryAtomStore implements AtomStore { @Override public Stream<Particle> getUpParticles(RadixAddress address, @Nullable String stagedUuid) { synchronized (lock) { List<Particle> upParticles = particleIndex.entrySet().stream() .filter(e -> e.getKey().getShardables().contains(address)) .filter(e -> { final Map<Spin, Set<Atom>> spinParticleIndex = e.getValue(); final boolean hasDown = spinParticleIndex.getOrDefault(Spin.DOWN, Collections.emptySet()) .stream().anyMatch(a -> atoms.get(a).isStore()); if (hasDown) { return false; } if (stagedUuid != null && stagedParticleIndex.getOrDefault(stagedUuid, Collections.emptyMap()).get(e.getKey()) == Spin.DOWN) { return false; } Set<Atom> uppingAtoms = spinParticleIndex.getOrDefault(Spin.UP, Collections.emptySet()); return uppingAtoms.stream().anyMatch(a -> atoms.get(a).isStore()); }) .map(Map.Entry::getKey) .collect(Collectors.toList()); if (stagedUuid != null) { stagedParticleIndex.getOrDefault(stagedUuid, Collections.emptyMap()).entrySet().stream() .filter(e -> e.getValue() == Spin.UP) .map(Entry::getKey) .filter(p -> p.getShardables().contains(address)) .forEach(upParticles::add); } return upParticles.stream(); } }  @Override void stageParticleGroup(String uuid, ParticleGroup particleGroup); @Override List<ParticleGroup> getStaged(String uuid); @Override List<ParticleGroup> getStagedAndClear(String uuid); void store(RadixAddress address, AtomObservation atomObservation); @Override Observable<Long> onSync(RadixAddress address); @Override Stream<Atom> getStoredAtoms(RadixAddress address); @Override Stream<Particle> getUpParticles(RadixAddress address, @Nullable String stagedUuid); @Override Observable<AtomObservation> getAtomObservations(RadixAddress address);  }