OutgoingBatches implements Serializable { public void sortChannels(List<NodeChannel> channels) { final HashMap<String, Date> errorChannels = new HashMap<String, Date>(); for (OutgoingBatch batch : batches) { if (batch.isErrorFlag()) { Date date = errorChannels.get(batch.getChannelId()); if (date == null || batch.getLastUpdatedTime().compareTo(date) > 0) { errorChannels.put(batch.getChannelId(), batch.getLastUpdatedTime()); } } } Collections.sort(channels, new Comparator<NodeChannel>() { public int compare(NodeChannel b1, NodeChannel b2) { boolean isError1 = errorChannels.containsKey(b1.getChannelId()); boolean isError2 = errorChannels.containsKey(b2.getChannelId()); if (!isError1 && !isError2) { return b1.getProcessingOrder() < b2.getProcessingOrder() ? -1 : 1; } else if (isError1 && isError2) { return errorChannels.get(b1.getChannelId()).compareTo( errorChannels.get(b2.getChannelId())); } else if (!isError1 && isError2) { return -1; } else { return 1; } } }); for (NodeChannel nodeChannel : channels) { long extractPeriodMillis = nodeChannel.getExtractPeriodMillis(); Date lastExtractedTime = nodeChannel.getLastExtractTime(); if ((extractPeriodMillis < 1) || (lastExtractedTime == null) || (Calendar.getInstance().getTimeInMillis() - lastExtractedTime.getTime() >= extractPeriodMillis)) { addActiveChannel(nodeChannel); } } filterBatchesForInactiveChannels(); } OutgoingBatches(List<OutgoingBatch> batches);  OutgoingBatches(); boolean containsBatches(); Set<NodeChannel> getActiveChannels(); void addActiveChannel(NodeChannel nodeChannel); void setActiveChannels(Set<NodeChannel> activeChannels); List<OutgoingBatch> getBatches(); void setBatches(List<OutgoingBatch> batches); List<OutgoingBatch> filterBatchesForChannel(Channel channel); int countBatches(boolean includeOnlyErrors); List<OutgoingBatch> filterBatchesForChannel(String channelId); List<OutgoingBatch> filterBatchesForChannels(Set<String> channels); void removeNonLoadBatches(); void removeThreadChannelBatches(String channelId); boolean containsLoadBatches(); boolean containsBatchesInError(); List<OutgoingBatch> getBatchesForChannel(Channel channel); List<OutgoingBatch> getBatchesForChannel(String channelId); List<OutgoingBatch> getBatchesForChannels(Set<String> channelIds); List<OutgoingBatch> filterBatchesForInactiveChannels(); void sortChannels(List<NodeChannel> channels);  }
@Test public void testChannelSortingNoErrors() { List<NodeChannel> channels = new ArrayList<NodeChannel>(); NodeChannel channelA = new NodeChannel("a"); channelA.setProcessingOrder(1); NodeChannel channelB = new NodeChannel("b"); channelB.setProcessingOrder(2); NodeChannel channelC = new NodeChannel("c"); channelC.setProcessingOrder(3); channels.add(channelC); channels.add(channelB); channels.add(channelA); List<OutgoingBatch> batches = new ArrayList<OutgoingBatch>(); OutgoingBatch batch1 = new OutgoingBatch("1", channelA.getChannelId(), Status.NE); batches.add(batch1); OutgoingBatch batch2 = new OutgoingBatch("1", channelB.getChannelId(), Status.NE); batches.add(batch2); OutgoingBatch batch3 = new OutgoingBatch("1", channelC.getChannelId(), Status.NE); batches.add(batch3); OutgoingBatches outgoingBatches = new OutgoingBatches(batches); outgoingBatches.sortChannels(channels); assertEquals(channelA, channels.get(0)); assertEquals(channelB, channels.get(1)); assertEquals(channelC, channels.get(2)); }
@Test public void testChannelSortingOneErrors() { List<NodeChannel> channels = new ArrayList<NodeChannel>(); NodeChannel channelA = new NodeChannel("a"); channelA.setProcessingOrder(1); NodeChannel channelB = new NodeChannel("b"); channelB.setProcessingOrder(2); NodeChannel channelC = new NodeChannel("c"); channelC.setProcessingOrder(3); channels.add(channelC); channels.add(channelB); channels.add(channelA); List<OutgoingBatch> batches = new ArrayList<OutgoingBatch>(); OutgoingBatch batch1 = new OutgoingBatch("1", channelA.getChannelId(), Status.NE); batch1.setStatus(OutgoingBatch.Status.ER); batch1.setErrorFlag(true); batches.add(batch1); OutgoingBatch batch2 = new OutgoingBatch("1", channelB.getChannelId(), Status.NE); batches.add(batch2); OutgoingBatch batch3 = new OutgoingBatch("1", channelC.getChannelId(), Status.NE); batches.add(batch3); OutgoingBatches outgoingBatches = new OutgoingBatches(batches); outgoingBatches.sortChannels(channels); assertEquals(channelB, channels.get(0)); assertEquals(channelC, channels.get(1)); assertEquals(channelA, channels.get(2)); }
@Test public void testChannelSortingTwoErrors() { List<NodeChannel> channels = new ArrayList<NodeChannel>(); NodeChannel channelA = new NodeChannel("a"); channelA.setProcessingOrder(1); NodeChannel channelB = new NodeChannel("b"); channelB.setProcessingOrder(2); NodeChannel channelC = new NodeChannel("c"); channelC.setProcessingOrder(3); channels.add(channelC); channels.add(channelB); channels.add(channelA); Date startTime = new Date(); Calendar cal = Calendar.getInstance(); cal.setTime(startTime); List<OutgoingBatch> batches = new ArrayList<OutgoingBatch>(); OutgoingBatch batch1 = new OutgoingBatch("1", channelA.getChannelId(), Status.NE); batch1.setStatus(OutgoingBatch.Status.ER); batch1.setErrorFlag(true); batch1.setLastUpdatedTime(cal.getTime()); batches.add(batch1); cal.setTime(startTime); cal.add(Calendar.SECOND, 1); OutgoingBatch batch2 = new OutgoingBatch("1", channelB.getChannelId(), Status.NE); batch2.setStatus(OutgoingBatch.Status.ER); batch2.setErrorFlag(true); batch2.setLastUpdatedTime(cal.getTime()); batches.add(batch2); OutgoingBatch batch3 = new OutgoingBatch("1", channelC.getChannelId(), Status.NE); batches.add(batch3); OutgoingBatches outgoingBatches = new OutgoingBatches(batches); outgoingBatches.sortChannels(channels); assertEquals(channelC, channels.get(0)); assertEquals(channelA, channels.get(1)); assertEquals(channelB, channels.get(2)); cal.setTime(startTime); cal.add(Calendar.SECOND, 2); batch1.setLastUpdatedTime(cal.getTime()); outgoingBatches.sortChannels(channels); assertEquals(channelC, channels.get(0)); assertEquals(channelB, channels.get(1)); assertEquals(channelA, channels.get(2)); cal.setTime(startTime); cal.add(Calendar.SECOND, -1); OutgoingBatch batch4 = new OutgoingBatch("1", channelA.getChannelId(), Status.NE); batch4.setStatus(OutgoingBatch.Status.ER); batch4.setErrorFlag(true); batch4.setLastUpdatedTime(cal.getTime()); batches.add(batch4); cal.setTime(startTime); cal.add(Calendar.SECOND, -2); OutgoingBatch batch5 = new OutgoingBatch("1", channelB.getChannelId(), Status.NE); batch5.setStatus(OutgoingBatch.Status.ER); batch5.setErrorFlag(true); batch5.setLastUpdatedTime(cal.getTime()); batches.add(batch5); cal.setTime(startTime); cal.add(Calendar.SECOND, 3); batch2.setLastUpdatedTime(cal.getTime()); outgoingBatches.sortChannels(channels); assertEquals(channelC, channels.get(0)); assertEquals(channelA, channels.get(1)); assertEquals(channelB, channels.get(2)); cal.setTime(startTime); cal.add(Calendar.SECOND, 4); batch1.setLastUpdatedTime(cal.getTime()); outgoingBatches.sortChannels(channels); assertEquals(channelC, channels.get(0)); assertEquals(channelB, channels.get(1)); assertEquals(channelA, channels.get(2)); }