content: package akka.grpc.interop;
  
  import java.util.concurrent.CompletionStage;
  import java.util.concurrent.CompletableFuture;
  
  import akka.NotUsed;
  import akka.grpc.GrpcServiceException;
  import akka.stream.Materializer;
  import akka.stream.javadsl.Flow;
  import akka.stream.javadsl.Source;
  import com.google.protobuf.ByteString;
  import io.grpc.Status;
  
  // Generated by our plugin
  import io.grpc.testing.integration.EmptyProtos;
  import io.grpc.testing.integration.Messages;
  import io.grpc.testing.integration.Messages.*;
  import io.grpc.testing.integration.TestService;
  
  /**
   * Implementation of the generated service.
   *
   * Essentially porting the client code from [[io.grpc.testing.integration.TestServiceImpl]] against our API's
   *
   * The same implementation is also be found as part of the 'non-scripted' tests at
   * /interop-tests/src/test/java/akka/grpc/interop/JavaTestServiceImpl.scala
   */
  public class JavaTestServiceImpl implements TestService {
    private final Materializer mat;
  
    private static final Flow<ResponseParameters, StreamingOutputCallResponse, NotUsed> parametersToResponseFlow =
      Flow.<ResponseParameters>create()
        .map(parameters ->
        StreamingOutputCallResponse.newBuilder()
          .setPayload(
            Payload.newBuilder()
              .setBody(ByteString.copyFrom(new byte[parameters.getSize()]))
              .build()
          )
          .build()
      );
  
    public JavaTestServiceImpl(Materializer mat) {
      this.mat = mat;
    }
  
    @Override
    public CompletionStage<EmptyProtos.Empty> emptyCall(EmptyProtos.Empty in) {
      return CompletableFuture.completedFuture(EmptyProtos.Empty.newBuilder().build());
    }
  
      @Override
      public CompletionStage<SimpleResponse> unaryCall(SimpleRequest in) {
          if (in.hasResponseStatus()) {
              EchoStatus requestStatus = in.getResponseStatus();
              Status status = Status.fromCodeValue(requestStatus.getCode()).withDescription(requestStatus.getMessage());
              CompletableFuture<SimpleResponse> cf = new CompletableFuture<>();
              cf.completeExceptionally(new GrpcServiceException(status));
              return cf;
          } else {
              return CompletableFuture.completedFuture(
                      SimpleResponse.newBuilder()
                              .setPayload(Payload.newBuilder()
                                      .setBody(ByteString.copyFrom(new byte[in.getResponseSize()]))
                                      .build())
                              .build()
              );
          }
      }
  
    @Override
    public CompletionStage<SimpleResponse> cacheableUnaryCall(SimpleRequest in) {
      return unaryCall(in);
    }
  
    @Override
    public Source<StreamingOutputCallResponse, NotUsed> streamingOutputCall(StreamingOutputCallRequest in) {
      return Source.from(in.getResponseParametersList())
        .via(parametersToResponseFlow)
        .mapMaterializedValue(x -> x);
    }
  
    @Override
    public CompletionStage<StreamingInputCallResponse> streamingInputCall(Source<StreamingInputCallRequest, NotUsed> in) {
      return in
        .map(i -> i.getPayload().getBody().size())
        .runFold(0, (Integer x, Integer y)->x+y, mat)
        .thenApply(sum -> StreamingInputCallResponse.newBuilder().setAggregatedPayloadSize(sum).build());
    }
  
  
    @Override
    public Source<StreamingOutputCallResponse, NotUsed> fullDuplexCall(Source<StreamingOutputCallRequest, NotUsed> in) {
      return in
              .map(req -> {
                  if(req.hasResponseStatus()) {
                      throw new GrpcServiceException(
                              Status
                                      .fromCodeValue(req.getResponseStatus().getCode())
                                      .withDescription(req.getResponseStatus().getMessage())
                      );
                  }else {
                      return req;
                  }
              })
        .mapConcat(r -> r.getResponseParametersList())
        .via(parametersToResponseFlow);
    }
  
  
    @Override
    public Source<StreamingOutputCallResponse, NotUsed> halfDuplexCall(Source<StreamingOutputCallRequest, NotUsed> in) {
      return null;
    }
  
    @Override
    public CompletionStage<EmptyProtos.Empty> unimplementedCall(EmptyProtos.Empty in) {
       throw new UnsupportedOperationException();
    }
  
  }
  