ThrottledByJobKeyConstraint implements JobConstraint { @Override public Collection<String> test(AsyncJobStatusCurator jobCurator, AsyncJobStatus inbound) { if (jobCurator == null) { throw new IllegalArgumentException("jobCurator is null"); } if (inbound == null) { throw new IllegalArgumentException("inbound is null"); } List<String> matching = jobCurator.fetchJobIdsByArguments(inbound.getJobKey(), null); return (matching != null && matching.size() > this.limit) ? matching : Collections.emptyList(); } ThrottledByJobKeyConstraint(String jobKey, int limit); @Override Collection<String> test(AsyncJobStatusCurator jobCurator, AsyncJobStatus inbound);  }
@Test void shouldThrottleMessagesExceedingTheLimit() { List<AsyncJobStatus> allJobs = conflictingJobs(3); JobConstraint constraint = new ThrottledByJobKeyConstraint(TEST_KEY, LIMIT); AsyncJobStatus inboundJob = new AsyncJobStatus() .setJobKey(TEST_KEY); Collection<String> conflicting = constraint.test(this.asyncJobCurator, inboundJob); assertEquals(3, conflicting.size()); }
@Test void shouldNotThrottleMessagesWhenInLimit() { List<AsyncJobStatus> allJobs = conflictingJobs(1); JobConstraint constraint = new ThrottledByJobKeyConstraint(TEST_KEY, LIMIT); AsyncJobStatus inboundJob = new AsyncJobStatus() .setJobKey(TEST_KEY); Collection<String> conflicting = constraint.test(this.asyncJobCurator, inboundJob); assertTrue(conflicting.isEmpty()); }