InMemoryJobService extends JobServiceGrpc.JobServiceImplBase implements FnService { @Override public void getPipeline( GetJobPipelineRequest request, StreamObserver<GetJobPipelineResponse> responseObserver) { LOG.trace("{} {}", GetJobPipelineRequest.class.getSimpleName(), request); String invocationId = request.getJobId(); try { JobInvocation invocation = getInvocation(invocationId); RunnerApi.Pipeline pipeline = invocation.getPipeline(); GetJobPipelineResponse response = GetJobPipelineResponse.newBuilder().setPipeline(pipeline).build(); responseObserver.onNext(response); responseObserver.onCompleted(); } catch (StatusRuntimeException | StatusException e) { responseObserver.onError(e); } catch (Exception e) { String errMessage = String.format("Encountered Unexpected Exception for Invocation %s", invocationId); LOG.error(errMessage, 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 testGetPipelineFailure() { prepareJob(); JobApi.GetJobPipelineRequest request = JobApi.GetJobPipelineRequest.newBuilder().setJobId(TEST_JOB_ID).build(); RecordingObserver<JobApi.GetJobPipelineResponse> recorder = new RecordingObserver<>(); service.getPipeline(request, recorder); assertThat(recorder.isSuccessful(), is(false)); assertThat(recorder.error, isA(StatusException.class)); }
@Test public void testGetPipelineIsSuccessful() throws Exception { prepareAndRunJob(); JobApi.GetJobPipelineRequest request = JobApi.GetJobPipelineRequest.newBuilder().setJobId(TEST_JOB_ID).build(); RecordingObserver<JobApi.GetJobPipelineResponse> recorder = new RecordingObserver<>(); service.getPipeline(request, recorder); assertThat(recorder.isSuccessful(), is(true)); assertThat(recorder.values, hasSize(1)); JobApi.GetJobPipelineResponse response = recorder.values.get(0); assertThat(response.getPipeline(), is(TEST_PIPELINE)); }