InMemoryJobService extends JobServiceGrpc.JobServiceImplBase implements FnService { public static InMemoryJobService create( GrpcFnServer<ArtifactStagingService> stagingService, Function<String, String> stagingServiceTokenProvider, ThrowingConsumer<Exception, String> cleanupJobFn, JobInvoker invoker) { return new InMemoryJobService( stagingService, stagingServiceTokenProvider, cleanupJobFn, invoker, DEFAULT_MAX_INVOCATION_HISTORY); } private  InMemoryJobService(
      GrpcFnServer<ArtifactStagingService> stagingService,
      Function<String, String> stagingServiceTokenProvider,
      ThrowingConsumer<Exception, String> cleanupJobFn,
      JobInvoker invoker,
      int maxInvocationHistory); static InMemoryJobService create(
      GrpcFnServer<ArtifactStagingService> stagingService,
      Function<String, String> stagingServiceTokenProvider,
      ThrowingConsumer<Exception, String> cleanupJobFn,
      JobInvoker invoker); static InMemoryJobService create(
      GrpcFnServer<ArtifactStagingService> stagingService,
      Function<String, String> stagingServiceTokenProvider,
      ThrowingConsumer<Exception, String> cleanupJobFn,
      JobInvoker invoker,
      int maxInvocationHistory); @Override void prepare(
      PrepareJobRequest request, StreamObserver<PrepareJobResponse> responseObserver); @Override void run(RunJobRequest request, StreamObserver<RunJobResponse> responseObserver); @Override void getJobs(GetJobsRequest request, StreamObserver<GetJobsResponse> responseObserver); @Override void getState(GetJobStateRequest request, StreamObserver<JobStateEvent> responseObserver); @Override void getPipeline(
      GetJobPipelineRequest request, StreamObserver<GetJobPipelineResponse> responseObserver); @Override void cancel(CancelJobRequest request, StreamObserver<CancelJobResponse> responseObserver); @Override void getStateStream(
      GetJobStateRequest request, StreamObserver<JobStateEvent> responseObserver); @Override void getMessageStream(
      JobMessagesRequest request, StreamObserver<JobMessagesResponse> responseObserver); @Override void getJobMetrics(
      JobApi.GetJobMetricsRequest request,
      StreamObserver<JobApi.GetJobMetricsResponse> responseObserver); @Override void describePipelineOptions(
      DescribePipelineOptionsRequest request,
      StreamObserver<DescribePipelineOptionsResponse> responseObserver); @Override void close(); static final int DEFAULT_MAX_INVOCATION_HISTORY; }
@Test public void testInvocationCleanup() { final int maxInvocationHistory = 3; service = InMemoryJobService.create( stagingServer, session -> "token", null, invoker, maxInvocationHistory); assertThat(getNumberOfInvocations(), is(0)); Job job1 = runJob(); assertThat(getNumberOfInvocations(), is(1)); Job job2 = runJob(); assertThat(getNumberOfInvocations(), is(2)); Job job3 = runJob(); assertThat(getNumberOfInvocations(), is(maxInvocationHistory)); Job job4 = runJob(); assertThat(getNumberOfInvocations(), is(maxInvocationHistory + 1)); job1.finish(); assertThat(getNumberOfInvocations(), is(maxInvocationHistory + 1)); job2.finish(); assertThat(getNumberOfInvocations(), is(maxInvocationHistory + 1)); job3.finish(); assertThat(getNumberOfInvocations(), is(maxInvocationHistory + 1)); job4.finish(); assertThat(getNumberOfInvocations(), is(maxInvocationHistory)); Job job5 = runJob(); assertThat(getNumberOfInvocations(), is(maxInvocationHistory + 1)); job5.finish(); assertThat(getNumberOfInvocations(), is(maxInvocationHistory)); }