@Test public void sortByAckStartIndexTest() { List<Position> list = new ArrayList<>(); Position position = new Position(1, 1, 1, 1); Position position1 = new Position(2, 2, 2, 2); Position position2 = new Position(3, 3, 3, 3); Position position3 = new Position(4, 4, 4, 4); list.add(position3); list.add(position); list.add(position2); list.add(position1); System.out.println(list); List<Position> positions = sortByAckStartIndex(list); System.out.println(positions); }
private List<Position> sortByAckStartIndex(List<Position> list) { return Lists.newArrayList(list).stream().sorted((thisPosition, thatPosition) -> Long.compare(thisPosition.getAckStartIndex(), thatPosition.getAckStartIndex())).collect(Collectors.toList()); }
ConcurrentConsumption extends Service implements ConcurrentConsumer { private List<Position> sortByAckStartIndex(List<Position> list) { return Lists.newArrayList(list).stream().sorted((thisPosition, thatPosition) -> Long.compare(thisPosition.getAckStartIndex(), thatPosition.getAckStartIndex())).collect(Collectors.toList()); } }
ConcurrentConsumption extends Service implements ConcurrentConsumer { private List<Position> sortByAckStartIndex(List<Position> list) { return Lists.newArrayList(list).stream().sorted((thisPosition, thatPosition) -> Long.compare(thisPosition.getAckStartIndex(), thatPosition.getAckStartIndex())).collect(Collectors.toList()); } ConcurrentConsumption(ClusterManager clusterManager, StoreService storeService, PartitionManager partitionManager,
                          MessageRetry messageRetry, PositionManager positionManager,
                          FilterMessageSupport filterMessageSupport, ArchiveManager archiveManager, SessionManager sessionManager); }
ConcurrentConsumption extends Service implements ConcurrentConsumer { private List<Position> sortByAckStartIndex(List<Position> list) { return Lists.newArrayList(list).stream().sorted((thisPosition, thatPosition) -> Long.compare(thisPosition.getAckStartIndex(), thatPosition.getAckStartIndex())).collect(Collectors.toList()); } ConcurrentConsumption(ClusterManager clusterManager, StoreService storeService, PartitionManager partitionManager,
                          MessageRetry messageRetry, PositionManager positionManager,
                          FilterMessageSupport filterMessageSupport, ArchiveManager archiveManager, SessionManager sessionManager); @Override PullResult getMessage(Consumer consumer, int count, long ackTimeout, long accessTimes, int concurrent); @Override boolean acknowledge(MessageLocation[] locations, Consumer consumer, boolean isSuccessAck); }
ConcurrentConsumption extends Service implements ConcurrentConsumer { private List<Position> sortByAckStartIndex(List<Position> list) { return Lists.newArrayList(list).stream().sorted((thisPosition, thatPosition) -> Long.compare(thisPosition.getAckStartIndex(), thatPosition.getAckStartIndex())).collect(Collectors.toList()); } ConcurrentConsumption(ClusterManager clusterManager, StoreService storeService, PartitionManager partitionManager,
                          MessageRetry messageRetry, PositionManager positionManager,
                          FilterMessageSupport filterMessageSupport, ArchiveManager archiveManager, SessionManager sessionManager); @Override PullResult getMessage(Consumer consumer, int count, long ackTimeout, long accessTimes, int concurrent); @Override boolean acknowledge(MessageLocation[] locations, Consumer consumer, boolean isSuccessAck);  }