@Test public void testPrintOnCreatesFullyFledgedJson() { String expectedJson = "{\"name\":\"processor name\"," + "\"mode\":\"tracking\"," + "\"fullName\":\"processor name@TokenStoreIdentifier\"," + "\"warnings\":[]," + "\"tokenStoreIdentifier\":\"TokenStoreIdentifier\"," + "\"freeThreadInstances\":[\"clientIdOne\"]," + "\"activeThreads\":2," + "\"canPause\":true," + "\"canPlay\":false," + "\"canSplit\":true," + "\"canMerge\":true," + "\"trackers\":[" + "{\"clientId\":\"clientIdOne\",\"segmentId\":0,\"caughtUp\":true,\"replaying\":false,\"tokenPosition\":0,\"errorState\":\"\",\"onePartOf\":2}," + "{\"clientId\":\"clientIdTwo\",\"segmentId\":1,\"caughtUp\":true,\"replaying\":false,\"tokenPosition\":0,\"errorState\":\"\",\"onePartOf\":2}" + "]}"; EventProcessorInfo.SegmentStatus trackerInfo0 = EventProcessorInfo.SegmentStatus.newBuilder() .setCaughtUp(true) .setReplaying(false) .setOnePartOf(2) .setSegmentId(0) .build(); EventProcessorInfo processorInfo0 = EventProcessorInfo.newBuilder() .setMode("Tracking") .setTokenStoreIdentifier("TokenStoreIdentifier") .setActiveThreads(1) .setAvailableThreads(3) .setRunning(true) .addSegmentStatus(trackerInfo0) .build(); EventProcessorInfo.SegmentStatus trackerInfo1 = EventProcessorInfo.SegmentStatus.newBuilder() .setCaughtUp(true) .setReplaying(false) .setOnePartOf(2) .setSegmentId(1) .build(); EventProcessorInfo processorInfo1 = EventProcessorInfo.newBuilder() .setMode("Tracking") .setTokenStoreIdentifier("TokenStoreIdentifier") .setActiveThreads(1) .setAvailableThreads(0) .setRunning(true) .addSegmentStatus(trackerInfo1) .build(); List<ClientProcessor> testClientProcessors = asList( new FakeClientProcessor("clientIdOne", true, processorInfo0), new FakeClientProcessor("clientIdTwo", true, processorInfo1) ); TrackingProcessor testSubject = new TrackingProcessor("processor name", "tracking", testClientProcessors); GsonMedia media = new GsonMedia(); testSubject.printOn(media); assertEquals(expectedJson, media.toString()); }
@Override public void printOn(Media media) { super.printOn(media); media.with("tokenStoreIdentifier", tokenStoreIdentifier); Set<String> freeThreadInstances = processors().stream() .filter(processor -> processor.eventProcessorInfo().getAvailableThreads() > ZERO_THREADS) .map(ClientProcessor::clientId) .collect(Collectors.toSet()); Integer activeThreads = processorInstances().stream() .map(EventProcessorInfo::getActiveThreads) .reduce(Integer::sum) .orElse(0); boolean isRunning = processors().stream().anyMatch(ClientProcessor::running); int largestSegmentFactor = processorInstances().stream() .map(EventProcessorInfo::getSegmentStatusList) .flatMap(List::stream) .mapToInt(EventProcessorInfo.SegmentStatus::getOnePartOf) .min() .orElse(1); boolean canMerge = isRunning && largestSegmentFactor != 1; media.withStrings(FREE_THREAD_INSTANCES_COUNT, freeThreadInstances) .with(ACTIVE_THREADS_COUNT, activeThreads) .with(CAN_PAUSE_KEY, isRunning) .with(CAN_PLAY_KEY, processors().stream().anyMatch(p -> !p.running())) .with(CAN_SPLIT_KEY, isRunning) .with(CAN_MERGE_KEY, canMerge) .with(TRACKERS_LIST_KEY, trackers()); }
TrackingProcessor extends GenericProcessor implements EventProcessor { @Override public void printOn(Media media) { super.printOn(media); media.with("tokenStoreIdentifier", tokenStoreIdentifier); Set<String> freeThreadInstances = processors().stream() .filter(processor -> processor.eventProcessorInfo().getAvailableThreads() > ZERO_THREADS) .map(ClientProcessor::clientId) .collect(Collectors.toSet()); Integer activeThreads = processorInstances().stream() .map(EventProcessorInfo::getActiveThreads) .reduce(Integer::sum) .orElse(0); boolean isRunning = processors().stream().anyMatch(ClientProcessor::running); int largestSegmentFactor = processorInstances().stream() .map(EventProcessorInfo::getSegmentStatusList) .flatMap(List::stream) .mapToInt(EventProcessorInfo.SegmentStatus::getOnePartOf) .min() .orElse(1); boolean canMerge = isRunning && largestSegmentFactor != 1; media.withStrings(FREE_THREAD_INSTANCES_COUNT, freeThreadInstances) .with(ACTIVE_THREADS_COUNT, activeThreads) .with(CAN_PAUSE_KEY, isRunning) .with(CAN_PLAY_KEY, processors().stream().anyMatch(p -> !p.running())) .with(CAN_SPLIT_KEY, isRunning) .with(CAN_MERGE_KEY, canMerge) .with(TRACKERS_LIST_KEY, trackers()); } }
TrackingProcessor extends GenericProcessor implements EventProcessor { @Override public void printOn(Media media) { super.printOn(media); media.with("tokenStoreIdentifier", tokenStoreIdentifier); Set<String> freeThreadInstances = processors().stream() .filter(processor -> processor.eventProcessorInfo().getAvailableThreads() > ZERO_THREADS) .map(ClientProcessor::clientId) .collect(Collectors.toSet()); Integer activeThreads = processorInstances().stream() .map(EventProcessorInfo::getActiveThreads) .reduce(Integer::sum) .orElse(0); boolean isRunning = processors().stream().anyMatch(ClientProcessor::running); int largestSegmentFactor = processorInstances().stream() .map(EventProcessorInfo::getSegmentStatusList) .flatMap(List::stream) .mapToInt(EventProcessorInfo.SegmentStatus::getOnePartOf) .min() .orElse(1); boolean canMerge = isRunning && largestSegmentFactor != 1; media.withStrings(FREE_THREAD_INSTANCES_COUNT, freeThreadInstances) .with(ACTIVE_THREADS_COUNT, activeThreads) .with(CAN_PAUSE_KEY, isRunning) .with(CAN_PLAY_KEY, processors().stream().anyMatch(p -> !p.running())) .with(CAN_SPLIT_KEY, isRunning) .with(CAN_MERGE_KEY, canMerge) .with(TRACKERS_LIST_KEY, trackers()); } TrackingProcessor(String name, String mode, Collection<ClientProcessor> processors); }
TrackingProcessor extends GenericProcessor implements EventProcessor { @Override public void printOn(Media media) { super.printOn(media); media.with("tokenStoreIdentifier", tokenStoreIdentifier); Set<String> freeThreadInstances = processors().stream() .filter(processor -> processor.eventProcessorInfo().getAvailableThreads() > ZERO_THREADS) .map(ClientProcessor::clientId) .collect(Collectors.toSet()); Integer activeThreads = processorInstances().stream() .map(EventProcessorInfo::getActiveThreads) .reduce(Integer::sum) .orElse(0); boolean isRunning = processors().stream().anyMatch(ClientProcessor::running); int largestSegmentFactor = processorInstances().stream() .map(EventProcessorInfo::getSegmentStatusList) .flatMap(List::stream) .mapToInt(EventProcessorInfo.SegmentStatus::getOnePartOf) .min() .orElse(1); boolean canMerge = isRunning && largestSegmentFactor != 1; media.withStrings(FREE_THREAD_INSTANCES_COUNT, freeThreadInstances) .with(ACTIVE_THREADS_COUNT, activeThreads) .with(CAN_PAUSE_KEY, isRunning) .with(CAN_PLAY_KEY, processors().stream().anyMatch(p -> !p.running())) .with(CAN_SPLIT_KEY, isRunning) .with(CAN_MERGE_KEY, canMerge) .with(TRACKERS_LIST_KEY, trackers()); } TrackingProcessor(String name, String mode, Collection<ClientProcessor> processors); @Override String fullName(); @Override Iterable<Warning> warnings(); @Override void printOn(Media media); }
TrackingProcessor extends GenericProcessor implements EventProcessor { @Override public void printOn(Media media) { super.printOn(media); media.with("tokenStoreIdentifier", tokenStoreIdentifier); Set<String> freeThreadInstances = processors().stream() .filter(processor -> processor.eventProcessorInfo().getAvailableThreads() > ZERO_THREADS) .map(ClientProcessor::clientId) .collect(Collectors.toSet()); Integer activeThreads = processorInstances().stream() .map(EventProcessorInfo::getActiveThreads) .reduce(Integer::sum) .orElse(0); boolean isRunning = processors().stream().anyMatch(ClientProcessor::running); int largestSegmentFactor = processorInstances().stream() .map(EventProcessorInfo::getSegmentStatusList) .flatMap(List::stream) .mapToInt(EventProcessorInfo.SegmentStatus::getOnePartOf) .min() .orElse(1); boolean canMerge = isRunning && largestSegmentFactor != 1; media.withStrings(FREE_THREAD_INSTANCES_COUNT, freeThreadInstances) .with(ACTIVE_THREADS_COUNT, activeThreads) .with(CAN_PAUSE_KEY, isRunning) .with(CAN_PLAY_KEY, processors().stream().anyMatch(p -> !p.running())) .with(CAN_SPLIT_KEY, isRunning) .with(CAN_MERGE_KEY, canMerge) .with(TRACKERS_LIST_KEY, trackers()); } TrackingProcessor(String name, String mode, Collection<ClientProcessor> processors); @Override String fullName(); @Override Iterable<Warning> warnings(); @Override void printOn(Media media);  }