@Test public void mergeBulletins() throws Exception { final BulletinEntity bulletinEntity1 = createBulletin("This is bulletin 1"); final BulletinEntity bulletinEntity2 = createBulletin("This is bulletin 2"); final BulletinEntity unauthorizedBulletin = new BulletinEntity(); unauthorizedBulletin.setId(bulletinId++); unauthorizedBulletin.setTimestamp(new Date()); unauthorizedBulletin.setCanRead(false); final BulletinEntity copyOfBulletin1 = createBulletin("This is bulletin 1"); final NodeIdentifier node1 = new NodeIdentifier("node-1", "host-1", 8080, "host-1", 19998, null, null, null, false); final NodeIdentifier node2 = new NodeIdentifier("node-2", "host-2", 8081, "host-2", 19999, null, null, null, false); final Map<NodeIdentifier, List<BulletinEntity>> nodeMap = new HashMap<>(); nodeMap.put(node1, new ArrayList<>()); nodeMap.put(node2, new ArrayList<>()); nodeMap.get(node1).add(bulletinEntity1); nodeMap.get(node1).add(bulletinEntity2); nodeMap.get(node1).add(unauthorizedBulletin); nodeMap.get(node2).add(copyOfBulletin1); final List<BulletinEntity> bulletinEntities = BulletinMerger.mergeBulletins(nodeMap, nodeMap.size()); assertEquals(bulletinEntities.size(), 3); assertTrue(bulletinEntities.contains(copyOfBulletin1)); assertEquals(copyOfBulletin1.getNodeAddress(), ALL_NODES_MESSAGE); assertTrue(bulletinEntities.contains(bulletinEntity2)); assertTrue(bulletinEntities.contains(unauthorizedBulletin)); }
public static List<BulletinEntity> mergeBulletins(final Map<NodeIdentifier, List<BulletinEntity>> bulletins, final int totalNodes) { final List<BulletinEntity> bulletinEntities = new ArrayList<>(); for (final Map.Entry<NodeIdentifier, List<BulletinEntity>> entry : bulletins.entrySet()) { final NodeIdentifier nodeId = entry.getKey(); final List<BulletinEntity> nodeBulletins = entry.getValue(); final String nodeAddress = nodeId.getApiAddress() + ":" + nodeId.getApiPort(); for (final BulletinEntity bulletinEntity : nodeBulletins) { if (bulletinEntity.getNodeAddress() == null) { bulletinEntity.setNodeAddress(nodeAddress); } if (bulletinEntity.getCanRead() && bulletinEntity.getBulletin() != null && bulletinEntity.getBulletin().getNodeAddress() == null) { bulletinEntity.getBulletin().setNodeAddress(nodeAddress); } bulletinEntities.add(bulletinEntity); } } final List<BulletinEntity> entities = Lists.newArrayList(); final Map<String,List<BulletinEntity>> groupingEntities = bulletinEntities.stream() .filter(bulletinEntity -> bulletinEntity.getCanRead()) .collect(Collectors.groupingBy(b -> b.getBulletin().getMessage())); groupingEntities.forEach((message, groupedBulletinEntities) -> { if (groupedBulletinEntities.size() == totalNodes) { final BulletinEntity selectedBulletinEntity = groupedBulletinEntities.stream() .max(Comparator.comparingLong(bulletinEntity -> { if (bulletinEntity.getTimestamp() == null) { return 0; } else { return bulletinEntity.getTimestamp().getTime(); } })).orElse(null); if (selectedBulletinEntity != null) { selectedBulletinEntity.setNodeAddress(ALL_NODES_MESSAGE); selectedBulletinEntity.getBulletin().setNodeAddress(ALL_NODES_MESSAGE); entities.add(selectedBulletinEntity); } } else { entities.addAll(groupedBulletinEntities); } }); bulletinEntities.stream() .filter(bulletinEntity -> !bulletinEntity.getCanRead()) .forEach(entities::add); Collections.sort(entities, (BulletinEntity o1, BulletinEntity o2) -> { final int timeComparison = o1.getTimestamp().compareTo(o2.getTimestamp()); if (timeComparison != 0) { return timeComparison; } return o1.getNodeAddress().compareTo(o2.getNodeAddress()); }); return entities; }
BulletinMerger { public static List<BulletinEntity> mergeBulletins(final Map<NodeIdentifier, List<BulletinEntity>> bulletins, final int totalNodes) { final List<BulletinEntity> bulletinEntities = new ArrayList<>(); for (final Map.Entry<NodeIdentifier, List<BulletinEntity>> entry : bulletins.entrySet()) { final NodeIdentifier nodeId = entry.getKey(); final List<BulletinEntity> nodeBulletins = entry.getValue(); final String nodeAddress = nodeId.getApiAddress() + ":" + nodeId.getApiPort(); for (final BulletinEntity bulletinEntity : nodeBulletins) { if (bulletinEntity.getNodeAddress() == null) { bulletinEntity.setNodeAddress(nodeAddress); } if (bulletinEntity.getCanRead() && bulletinEntity.getBulletin() != null && bulletinEntity.getBulletin().getNodeAddress() == null) { bulletinEntity.getBulletin().setNodeAddress(nodeAddress); } bulletinEntities.add(bulletinEntity); } } final List<BulletinEntity> entities = Lists.newArrayList(); final Map<String,List<BulletinEntity>> groupingEntities = bulletinEntities.stream() .filter(bulletinEntity -> bulletinEntity.getCanRead()) .collect(Collectors.groupingBy(b -> b.getBulletin().getMessage())); groupingEntities.forEach((message, groupedBulletinEntities) -> { if (groupedBulletinEntities.size() == totalNodes) { final BulletinEntity selectedBulletinEntity = groupedBulletinEntities.stream() .max(Comparator.comparingLong(bulletinEntity -> { if (bulletinEntity.getTimestamp() == null) { return 0; } else { return bulletinEntity.getTimestamp().getTime(); } })).orElse(null); if (selectedBulletinEntity != null) { selectedBulletinEntity.setNodeAddress(ALL_NODES_MESSAGE); selectedBulletinEntity.getBulletin().setNodeAddress(ALL_NODES_MESSAGE); entities.add(selectedBulletinEntity); } } else { entities.addAll(groupedBulletinEntities); } }); bulletinEntities.stream() .filter(bulletinEntity -> !bulletinEntity.getCanRead()) .forEach(entities::add); Collections.sort(entities, (BulletinEntity o1, BulletinEntity o2) -> { final int timeComparison = o1.getTimestamp().compareTo(o2.getTimestamp()); if (timeComparison != 0) { return timeComparison; } return o1.getNodeAddress().compareTo(o2.getNodeAddress()); }); return entities; } }
BulletinMerger { public static List<BulletinEntity> mergeBulletins(final Map<NodeIdentifier, List<BulletinEntity>> bulletins, final int totalNodes) { final List<BulletinEntity> bulletinEntities = new ArrayList<>(); for (final Map.Entry<NodeIdentifier, List<BulletinEntity>> entry : bulletins.entrySet()) { final NodeIdentifier nodeId = entry.getKey(); final List<BulletinEntity> nodeBulletins = entry.getValue(); final String nodeAddress = nodeId.getApiAddress() + ":" + nodeId.getApiPort(); for (final BulletinEntity bulletinEntity : nodeBulletins) { if (bulletinEntity.getNodeAddress() == null) { bulletinEntity.setNodeAddress(nodeAddress); } if (bulletinEntity.getCanRead() && bulletinEntity.getBulletin() != null && bulletinEntity.getBulletin().getNodeAddress() == null) { bulletinEntity.getBulletin().setNodeAddress(nodeAddress); } bulletinEntities.add(bulletinEntity); } } final List<BulletinEntity> entities = Lists.newArrayList(); final Map<String,List<BulletinEntity>> groupingEntities = bulletinEntities.stream() .filter(bulletinEntity -> bulletinEntity.getCanRead()) .collect(Collectors.groupingBy(b -> b.getBulletin().getMessage())); groupingEntities.forEach((message, groupedBulletinEntities) -> { if (groupedBulletinEntities.size() == totalNodes) { final BulletinEntity selectedBulletinEntity = groupedBulletinEntities.stream() .max(Comparator.comparingLong(bulletinEntity -> { if (bulletinEntity.getTimestamp() == null) { return 0; } else { return bulletinEntity.getTimestamp().getTime(); } })).orElse(null); if (selectedBulletinEntity != null) { selectedBulletinEntity.setNodeAddress(ALL_NODES_MESSAGE); selectedBulletinEntity.getBulletin().setNodeAddress(ALL_NODES_MESSAGE); entities.add(selectedBulletinEntity); } } else { entities.addAll(groupedBulletinEntities); } }); bulletinEntities.stream() .filter(bulletinEntity -> !bulletinEntity.getCanRead()) .forEach(entities::add); Collections.sort(entities, (BulletinEntity o1, BulletinEntity o2) -> { final int timeComparison = o1.getTimestamp().compareTo(o2.getTimestamp()); if (timeComparison != 0) { return timeComparison; } return o1.getNodeAddress().compareTo(o2.getNodeAddress()); }); return entities; } private  BulletinMerger(); }
BulletinMerger { public static List<BulletinEntity> mergeBulletins(final Map<NodeIdentifier, List<BulletinEntity>> bulletins, final int totalNodes) { final List<BulletinEntity> bulletinEntities = new ArrayList<>(); for (final Map.Entry<NodeIdentifier, List<BulletinEntity>> entry : bulletins.entrySet()) { final NodeIdentifier nodeId = entry.getKey(); final List<BulletinEntity> nodeBulletins = entry.getValue(); final String nodeAddress = nodeId.getApiAddress() + ":" + nodeId.getApiPort(); for (final BulletinEntity bulletinEntity : nodeBulletins) { if (bulletinEntity.getNodeAddress() == null) { bulletinEntity.setNodeAddress(nodeAddress); } if (bulletinEntity.getCanRead() && bulletinEntity.getBulletin() != null && bulletinEntity.getBulletin().getNodeAddress() == null) { bulletinEntity.getBulletin().setNodeAddress(nodeAddress); } bulletinEntities.add(bulletinEntity); } } final List<BulletinEntity> entities = Lists.newArrayList(); final Map<String,List<BulletinEntity>> groupingEntities = bulletinEntities.stream() .filter(bulletinEntity -> bulletinEntity.getCanRead()) .collect(Collectors.groupingBy(b -> b.getBulletin().getMessage())); groupingEntities.forEach((message, groupedBulletinEntities) -> { if (groupedBulletinEntities.size() == totalNodes) { final BulletinEntity selectedBulletinEntity = groupedBulletinEntities.stream() .max(Comparator.comparingLong(bulletinEntity -> { if (bulletinEntity.getTimestamp() == null) { return 0; } else { return bulletinEntity.getTimestamp().getTime(); } })).orElse(null); if (selectedBulletinEntity != null) { selectedBulletinEntity.setNodeAddress(ALL_NODES_MESSAGE); selectedBulletinEntity.getBulletin().setNodeAddress(ALL_NODES_MESSAGE); entities.add(selectedBulletinEntity); } } else { entities.addAll(groupedBulletinEntities); } }); bulletinEntities.stream() .filter(bulletinEntity -> !bulletinEntity.getCanRead()) .forEach(entities::add); Collections.sort(entities, (BulletinEntity o1, BulletinEntity o2) -> { final int timeComparison = o1.getTimestamp().compareTo(o2.getTimestamp()); if (timeComparison != 0) { return timeComparison; } return o1.getNodeAddress().compareTo(o2.getNodeAddress()); }); return entities; } private  BulletinMerger(); static List<BulletinEntity> mergeBulletins(final Map<NodeIdentifier, List<BulletinEntity>> bulletins, final int totalNodes); }
BulletinMerger { public static List<BulletinEntity> mergeBulletins(final Map<NodeIdentifier, List<BulletinEntity>> bulletins, final int totalNodes) { final List<BulletinEntity> bulletinEntities = new ArrayList<>(); for (final Map.Entry<NodeIdentifier, List<BulletinEntity>> entry : bulletins.entrySet()) { final NodeIdentifier nodeId = entry.getKey(); final List<BulletinEntity> nodeBulletins = entry.getValue(); final String nodeAddress = nodeId.getApiAddress() + ":" + nodeId.getApiPort(); for (final BulletinEntity bulletinEntity : nodeBulletins) { if (bulletinEntity.getNodeAddress() == null) { bulletinEntity.setNodeAddress(nodeAddress); } if (bulletinEntity.getCanRead() && bulletinEntity.getBulletin() != null && bulletinEntity.getBulletin().getNodeAddress() == null) { bulletinEntity.getBulletin().setNodeAddress(nodeAddress); } bulletinEntities.add(bulletinEntity); } } final List<BulletinEntity> entities = Lists.newArrayList(); final Map<String,List<BulletinEntity>> groupingEntities = bulletinEntities.stream() .filter(bulletinEntity -> bulletinEntity.getCanRead()) .collect(Collectors.groupingBy(b -> b.getBulletin().getMessage())); groupingEntities.forEach((message, groupedBulletinEntities) -> { if (groupedBulletinEntities.size() == totalNodes) { final BulletinEntity selectedBulletinEntity = groupedBulletinEntities.stream() .max(Comparator.comparingLong(bulletinEntity -> { if (bulletinEntity.getTimestamp() == null) { return 0; } else { return bulletinEntity.getTimestamp().getTime(); } })).orElse(null); if (selectedBulletinEntity != null) { selectedBulletinEntity.setNodeAddress(ALL_NODES_MESSAGE); selectedBulletinEntity.getBulletin().setNodeAddress(ALL_NODES_MESSAGE); entities.add(selectedBulletinEntity); } } else { entities.addAll(groupedBulletinEntities); } }); bulletinEntities.stream() .filter(bulletinEntity -> !bulletinEntity.getCanRead()) .forEach(entities::add); Collections.sort(entities, (BulletinEntity o1, BulletinEntity o2) -> { final int timeComparison = o1.getTimestamp().compareTo(o2.getTimestamp()); if (timeComparison != 0) { return timeComparison; } return o1.getNodeAddress().compareTo(o2.getNodeAddress()); }); return entities; } private  BulletinMerger(); static List<BulletinEntity> mergeBulletins(final Map<NodeIdentifier, List<BulletinEntity>> bulletins, final int totalNodes); static Comparator<BulletinEntity> BULLETIN_COMPARATOR; }