package org.languagetool.server;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
import java.util.stream.Collectors;

import org.languagetool.AnalyzedSentence;
import org.languagetool.CheckResults;
import org.languagetool.Language;
import org.languagetool.Languages;
import org.languagetool.UserConfig;
import org.languagetool.rules.GRPCUtils;
import org.languagetool.rules.RuleMatch;
import org.languagetool.rules.ml.MLServerProto.AnalyzeResponse;
import org.languagetool.rules.ml.MLServerProto.ProcessResponse;
import org.languagetool.rules.ml.MLServerProto.ProcessingOptions;
import org.languagetool.rules.ml.ProcessingServerGrpc.ProcessingServerImplBase;

import io.grpc.Server;
import io.grpc.ServerBuilder;
import lombok.extern.slf4j.Slf4j;

@Slf4j
public class GRPCServer extends ProcessingServerImplBase
{
  private PipelinePool pool;
  private UserConfig userConfig;

  public GRPCServer(HTTPServerConfig serverConfig) {
    pool = new PipelinePool(serverConfig, null, false);
    userConfig = new UserConfig();
  }

  private PipelineSettings buildSettings(ProcessingOptions options) {
    Language lang = Languages.getLanguageForShortCode(options.getLanguage());
    return new PipelineSettings(lang, userConfig);
  }

  /**
   */
  public void analyze(org.languagetool.rules.ml.MLServerProto.AnalyzeRequest request,
      io.grpc.stub.StreamObserver<org.languagetool.rules.ml.MLServerProto.AnalyzeResponse> responseObserver) {
    try {
      PipelineSettings settings = buildSettings(request.getOptions());
      Pipeline pipeline = pool.getPipeline(settings);
      List<AnalyzedSentence> sentences = pipeline.analyzeText(request.getText());
      AnalyzeResponse response = AnalyzeResponse.newBuilder()
        .addAllSentences(sentences.stream().map(GRPCUtils::toGRPC).collect(Collectors.toList()))
        .build();
      responseObserver.onNext(response);
    } catch (Exception e) {
      log.warn("Analyze request failed", e);
      responseObserver.onError(e);
    }
  }

  /**
   */
  public void process(org.languagetool.rules.ml.MLServerProto.ProcessRequest request,
      io.grpc.stub.StreamObserver<org.languagetool.rules.ml.MLServerProto.ProcessResponse> responseObserver) {
    try {
      PipelineSettings settings = buildSettings(request.getOptions());
      Pipeline pipeline = pool.getPipeline(settings);
      List<AnalyzedSentence> sentences = request.getSentencesList().stream()
        .map(GRPCUtils::fromGRPC).collect(Collectors.toList());

      // TODO: rawMatches should have all results, regardless of mode/level/tempOff/...
      List<RuleMatch> rawMatches = new ArrayList<>();
      CheckResults results = pipeline.checkAnalyzedSentences(sentences, m -> rawMatches.add(m));

      ProcessResponse response = ProcessResponse.newBuilder()
        .addAllRawMatches(rawMatches.stream().map(GRPCUtils::toGRPC).collect(Collectors.toList()))
        .addAllMatches(results.getRuleMatches().stream().map(GRPCUtils::toGRPC).collect(Collectors.toList()))
        .build();
      responseObserver.onNext(response);
    } catch (Exception e) {
      log.warn("Process request failed", e);
      responseObserver.onError(e);
    }
  }

  public static void main(String[] args) throws Exception {
    HTTPServerConfig config = new HTTPServerConfig(args);
    GRPCServer instance = new GRPCServer(config);
    Executor executor = Executors.newCachedThreadPool();
    Server server = ServerBuilder.forPort(config.getPort())
      .addService(instance)
      .executor(executor)
      .build();
    server.start();
    server.awaitTermination();
  }
}

@@ -1940,32 +1940,70 @@
 Next(response);%0A
+      responseObserver.onCompleted();%0A
     %7D catch (Exc
@@ -3157,24 +3157,62 @@
 (response);%0A
+      responseObserver.onCompleted();%0A
     %7D catch 

fix grpc server