InMemoryManager { Map<String, SystemStreamMetadata> getSystemStreamMetadata(String systemName, Set<String> streamNames) { Map<String, Map<SystemStreamPartition, List<IncomingMessageEnvelope>>> result = bufferedMessages.entrySet() .stream() .filter(entry -> systemName.equals(entry.getKey().getSystem()) && streamNames.contains(entry.getKey().getStream())) .collect(Collectors.groupingBy(entry -> entry.getKey().getStream(), Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue))); return result.entrySet() .stream() .collect(Collectors.toMap(Map.Entry::getKey, entry -> constructSystemStreamMetadata(entry.getKey(), entry.getValue()))); } InMemoryManager();   }
@Test public void testGetSystemStreamMetadata() { this.inMemoryManager.initializeStream(new StreamSpec(STREAM0, STREAM0, SYSTEM, 1)); this.inMemoryManager.initializeStream(new StreamSpec(STREAM1, STREAM1, SYSTEM, 1)); this.inMemoryManager.initializeStream(new StreamSpec("otherStream", "otherStream", SYSTEM, 1)); SystemStreamMetadata systemStreamMetadata0 = new SystemStreamMetadata(STREAM0, ImmutableMap.of(new Partition(0), new SystemStreamMetadata.SystemStreamPartitionMetadata(null, null, "0"))); assertEquals(ImmutableMap.of(STREAM0, systemStreamMetadata0), this.inMemoryManager.getSystemStreamMetadata(SYSTEM, ImmutableSet.of(STREAM0))); SystemStreamPartition ssp0 = new SystemStreamPartition(SYSTEM, STREAM0, new Partition(0)); this.inMemoryManager.put(ssp0, "key00", "message00"); systemStreamMetadata0 = new SystemStreamMetadata(STREAM0, ImmutableMap.of(new Partition(0), new SystemStreamMetadata.SystemStreamPartitionMetadata("0", "0", "1"))); assertEquals(ImmutableMap.of(STREAM0, systemStreamMetadata0), this.inMemoryManager.getSystemStreamMetadata(SYSTEM, ImmutableSet.of(STREAM0))); this.inMemoryManager.put(ssp0, "key01", "message01"); SystemStreamPartition ssp1 = new SystemStreamPartition(SYSTEM, STREAM1, new Partition(0)); this.inMemoryManager.put(ssp1, "key10", "message10"); systemStreamMetadata0 = new SystemStreamMetadata(STREAM0, ImmutableMap.of(new Partition(0), new SystemStreamMetadata.SystemStreamPartitionMetadata("0", "1", "2"))); SystemStreamMetadata systemStreamMetadata1 = new SystemStreamMetadata(STREAM1, ImmutableMap.of(new Partition(0), new SystemStreamMetadata.SystemStreamPartitionMetadata("0", "0", "1"))); assertEquals(ImmutableMap.of(STREAM0, systemStreamMetadata0, STREAM1, systemStreamMetadata1), this.inMemoryManager.getSystemStreamMetadata(SYSTEM, ImmutableSet.of(STREAM0, STREAM1))); this.inMemoryManager.put(ssp0, "key02", new EndOfStreamMessage()); systemStreamMetadata0 = new SystemStreamMetadata(STREAM0, ImmutableMap.of(new Partition(0), new SystemStreamMetadata.SystemStreamPartitionMetadata("0", "1", "2"))); assertEquals(ImmutableMap.of(STREAM0, systemStreamMetadata0), this.inMemoryManager.getSystemStreamMetadata(SYSTEM, ImmutableSet.of(STREAM0))); }