InMemoryJobService extends JobServiceGrpc.JobServiceImplBase implements FnService { @Override public void getJobs(GetJobsRequest request, StreamObserver<GetJobsResponse> responseObserver) { LOG.trace("{} {}", GetJobsRequest.class.getSimpleName(), request); try { List<JobInfo> result = new ArrayList<>(); for (JobInvocation invocation : invocations.values()) { result.add(invocation.toProto()); } GetJobsResponse response = GetJobsResponse.newBuilder().addAllJobInfo(result).build(); responseObserver.onNext(response); responseObserver.onCompleted(); } catch (Exception e) { LOG.error("Encountered Unexpected Exception", e); responseObserver.onError(Status.INTERNAL.withCause(e).asException()); } } 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 testGetJobsIsSuccessful() throws Exception { prepareAndRunJob(); JobApi.GetJobsRequest request = JobApi.GetJobsRequest.newBuilder().build(); RecordingObserver<JobApi.GetJobsResponse> recorder = new RecordingObserver<>(); service.getJobs(request, recorder); assertThat(recorder.isSuccessful(), is(true)); assertThat(recorder.values, hasSize(1)); JobApi.GetJobsResponse response = recorder.values.get(0); List<JobApi.JobInfo> jobs = response.getJobInfoList(); assertThat(jobs, hasSize(1)); JobApi.JobInfo job = jobs.get(0); assertThat(job.getJobId(), is(TEST_JOB_ID)); assertThat(job.getJobName(), is(TEST_JOB_NAME)); }