@Test public void testSplitSegmentSelectsTheLargestSegmentToSplit() { testSubject.splitSegment(CONTEXT, CLIENTS, PROCESSOR_NAME_TO_SPLIT); ArgumentCaptor<SplitSegmentRequest> argumentCaptor = ArgumentCaptor.forClass(SplitSegmentRequest.class); verify(applicationEventPublisher).publishEvent(argumentCaptor.capture()); SplitSegmentRequest result = argumentCaptor.getValue(); assertFalse(result.isProxied()); assertEquals(CLIENT_NAME_OF_SPLIT, result.getClientId()); assertEquals(PROCESSOR_NAME_TO_SPLIT, result.getProcessorName()); assertEquals(SEGMENT_ID_TO_SPLIT, result.getSegmentId()); }
public void splitSegment(String context, List<String> clientIds, String processorName) { Map<ClientSegmentPair, SegmentStatus> clientToTracker = buildClientToTrackerMap(clientIds, processorName, REGULAR_ORDER); Integer biggestSegment = clientToTracker.values().stream() .min(Comparator.comparingInt(SegmentStatus::getOnePartOf)) .map(SegmentStatus::getSegmentId) .orElseThrow(() -> new IllegalArgumentException( "No segments found for processor name [" + processorName + "]" )); applicationEventPublisher.publishEvent(new SplitSegmentRequest( NOT_PROXIED, context, getClientIdForSegment(clientToTracker, biggestSegment) .orElseThrow(() -> new IllegalArgumentException("No client found which has a claim on segment [" + biggestSegment + "]")), processorName, biggestSegment )); }
ProcessorEventPublisher { public void splitSegment(String context, List<String> clientIds, String processorName) { Map<ClientSegmentPair, SegmentStatus> clientToTracker = buildClientToTrackerMap(clientIds, processorName, REGULAR_ORDER); Integer biggestSegment = clientToTracker.values().stream() .min(Comparator.comparingInt(SegmentStatus::getOnePartOf)) .map(SegmentStatus::getSegmentId) .orElseThrow(() -> new IllegalArgumentException( "No segments found for processor name [" + processorName + "]" )); applicationEventPublisher.publishEvent(new SplitSegmentRequest( NOT_PROXIED, context, getClientIdForSegment(clientToTracker, biggestSegment) .orElseThrow(() -> new IllegalArgumentException("No client found which has a claim on segment [" + biggestSegment + "]")), processorName, biggestSegment )); } }
ProcessorEventPublisher { public void splitSegment(String context, List<String> clientIds, String processorName) { Map<ClientSegmentPair, SegmentStatus> clientToTracker = buildClientToTrackerMap(clientIds, processorName, REGULAR_ORDER); Integer biggestSegment = clientToTracker.values().stream() .min(Comparator.comparingInt(SegmentStatus::getOnePartOf)) .map(SegmentStatus::getSegmentId) .orElseThrow(() -> new IllegalArgumentException( "No segments found for processor name [" + processorName + "]" )); applicationEventPublisher.publishEvent(new SplitSegmentRequest( NOT_PROXIED, context, getClientIdForSegment(clientToTracker, biggestSegment) .orElseThrow(() -> new IllegalArgumentException("No client found which has a claim on segment [" + biggestSegment + "]")), processorName, biggestSegment )); } ProcessorEventPublisher(PlatformService platformService,
                                   ApplicationEventPublisher applicationEventPublisher,
                                   ClientProcessors clientProcessors); }
ProcessorEventPublisher { public void splitSegment(String context, List<String> clientIds, String processorName) { Map<ClientSegmentPair, SegmentStatus> clientToTracker = buildClientToTrackerMap(clientIds, processorName, REGULAR_ORDER); Integer biggestSegment = clientToTracker.values().stream() .min(Comparator.comparingInt(SegmentStatus::getOnePartOf)) .map(SegmentStatus::getSegmentId) .orElseThrow(() -> new IllegalArgumentException( "No segments found for processor name [" + processorName + "]" )); applicationEventPublisher.publishEvent(new SplitSegmentRequest( NOT_PROXIED, context, getClientIdForSegment(clientToTracker, biggestSegment) .orElseThrow(() -> new IllegalArgumentException("No client found which has a claim on segment [" + biggestSegment + "]")), processorName, biggestSegment )); } ProcessorEventPublisher(PlatformService platformService,
                                   ApplicationEventPublisher applicationEventPublisher,
                                   ClientProcessors clientProcessors); @PostConstruct void init(); void pauseProcessorRequest(String context, String clientId, String processorName); void startProcessorRequest(String context, String clientId, String processorName); void releaseSegment(String context, String clientId, String processorName, int segmentId); void splitSegment(String context, List<String> clientIds, String processorName); void mergeSegment(String context, List<String> clientIds, String processorName); }
ProcessorEventPublisher { public void splitSegment(String context, List<String> clientIds, String processorName) { Map<ClientSegmentPair, SegmentStatus> clientToTracker = buildClientToTrackerMap(clientIds, processorName, REGULAR_ORDER); Integer biggestSegment = clientToTracker.values().stream() .min(Comparator.comparingInt(SegmentStatus::getOnePartOf)) .map(SegmentStatus::getSegmentId) .orElseThrow(() -> new IllegalArgumentException( "No segments found for processor name [" + processorName + "]" )); applicationEventPublisher.publishEvent(new SplitSegmentRequest( NOT_PROXIED, context, getClientIdForSegment(clientToTracker, biggestSegment) .orElseThrow(() -> new IllegalArgumentException("No client found which has a claim on segment [" + biggestSegment + "]")), processorName, biggestSegment )); } ProcessorEventPublisher(PlatformService platformService,
                                   ApplicationEventPublisher applicationEventPublisher,
                                   ClientProcessors clientProcessors); @PostConstruct void init(); void pauseProcessorRequest(String context, String clientId, String processorName); void startProcessorRequest(String context, String clientId, String processorName); void releaseSegment(String context, String clientId, String processorName, int segmentId); void splitSegment(String context, List<String> clientIds, String processorName); void mergeSegment(String context, List<String> clientIds, String processorName);  }