@Service
// GrpcServiceImplBase -> 프로토콜버퍼파일에 명시된 service의 추상클래스입니다
public class GrpcService extends GrpcServiceGrpc.GrpcServiceImplBase { 

    Logger log = LoggerFactory.getLogger(this.getClass());

    // Unary : 1요청, 1응답 방식
    @Override
    public void getOne(GrpcModel.Request request, StreamObserver<GrpcModel.Response> responseObserver) {

        try {
            log.info("getOne: request: {}", request);
            // 기본타입객체의 경우 validation을 위한 has메서드를 제공합니다
            if (request.hasCustomObject()) {
                // TODO: exception
            }

            // 서비스의 비즈니스로직을 수행합니다
            // ...

            // grpc가 구성해준 모델클래스입니다
            GrpcModel.Response response = GrpcModel.Response
                    .newBuilder()
                    .addAllValue(Lists.newArrayList(1.0,2.0,3.0,4.0,5.0))
                    .build();

            // 하나의 요청에 대한 하나의 응답을 처리합니다
            responseObserver.onNext(response);

            // 응답에 대한 완료이벤트를 처리합니다
            responseObserver.onCompleted();

        } catch (Exception e) {

            // 로직수행중 에러가 발생할 경우 클라이언트에 Error 반환합니다
            responseObserver.onError(
                    Status.INTERNAL
                            .withDescription(e.getMessage())
                            .withCause(e) // cause는 클라이언트로는 전송되지 않습니다. 인터셉터를 구성할경우 참조용도입니다.
                            .asRuntimeException()
            );
        }
    }

    // 클라이언트 스트리밍 - 클라이언트에서 완료이벤트가 전송되지 않으면, 서버는 계속 응답을 대기합니다
    // n요청: 1응답
    @Override
    public StreamObserver<GrpcModel.Request> clientStream(StreamObserver<GrpcModel.Response> responseObserver) {
        return new StreamObserver<GrpcModel.Request>() {

            @Override public void onNext(GrpcModel.Request value) {
                // 클라이언트로부터 완료이벤트가 전송될때까지 계속 요청이
                log.info("clientStream onNext request: {}", value);

                // 비즈니스 로직 처리
                // ...
            }

            @Override
            public void onError(Throwable t) {
                //스트리밍중 에러 처리
                log.info("clientStream onError: {}", t.getMessage());
            }

            @Override public void onCompleted() {

                log.info("clientStream onCompleted");
								
                // 비즈니스 로직 처리
                // ...

                // 클라이언트로부터 스트림완료이벤트가 오면 클라이언트로 1번의 응답을 보낼수 있습니다
                GrpcModel.Response response = GrpcModel.Response
                        .newBuilder()
                        .addAllValue(Lists.newArrayList(1.0,2.0,3.0,4.0,5.0))
                        .build();
                responseObserver.onNext(response);
                responseObserver.onCompleted();
            }
        };
    }

    // 서버 스트리밍 방식 - 클라이언트의 스트리밍과 반대개념 입니다 ( 개념적으로 모바일의 알림의 그것과 같습니다 )
	// 클라이언트의 요청이 시작되면 서버가 완료이벤트를 전송하기 전까지 클라이언트는 서버로부터 데이터를 받을수 있습니다
    // 1요청: n응답
    @Override
    public void serverStream(GrpcModel.Request request, StreamObserver<GrpcModel.Response> responseObserver) {

        log.info("serverStream: request: {}", request);

        GrpcModel.Response response = GrpcModel.Response
                .newBuilder()
                .addAllValue(Lists.newArrayList(1.0,2.0,3.0,4.0,5.0))
                .build();

        // 클라이언트에의 요청은 1번이지만 서버는 아래처럼 여러번의 데이터를 스트리밍 할수 있습니다
        responseObserver.onNext(response);
        responseObserver.onNext(response);
        responseObserver.onNext(response);
        responseObserver.onNext(response);
        responseObserver.onNext(response);

        responseObserver.onCompleted();
    }

    // 양방향 스트리밍방식 (Bi-directional) - 클라이언트에서 완료이벤트가 전송하기 전까지, 서로간의 스트리밍을 진행합니다
    // n요청: n응답
    @Override
    public StreamObserver<GrpcModel.Request> biStream(StreamObserver<GrpcModel.Response> responseObserver) {
        return new StreamObserver<GrpcModel.Request>() {
            @Override public void onNext(GrpcModel.Request value) {

                log.info("biStream onNext request: {}", value);

                // 클라이언트로부터 데이터가 올 때마다 onNext가 호출된다
                // 1개의 요청이 올 때마다 n번의 응답을 스트리밍 전송한다
                GrpcModel.Response response = GrpcModel.Response
                        .newBuilder()
                        .addAllValue(Lists.newArrayList(1.0,2.0,3.0,4.0,5.0))
                        .build();

                responseObserver.onNext(response);
                responseObserver.onNext(response);
                responseObserver.onNext(response);
            }
            @Override public void onError(Throwable t) {
                log.info("biStream onError request: {}", t.getMessage());
            }
            @Override public void onCompleted() {
                log.info("biStream onComplete");
                // 클라이언트의 완료이벤트가 오면 서버도 완료이벤트를 진행한다
                responseObserver.onCompleted();
            }
        };
    }
}

@Configuration
@RequiredArgsConstructor
public class GrpcServerConfig {

    @Value("${grpc.port:8888}")
    Integer grpcPort;

    // service 구현체 bean을 주입받습니다
    private final GrpcService rpcService;

    // Grpc Java는 Netty를 사용합니다
    @Bean
    public Server grpcServer() {
        return ServerBuilder
                .forPort(grpcPort)
                .addService(rpcService)
                .build();
    }
}

@Component
@RequiredArgsConstructor
public class ServerStartRunner implements ApplicationRunner, DisposableBean {
		
    // config의 grpc bean을 주입받습니다
    private final Server grpcServer;

    // grpc는 스프링과는 별도로 구동시켜야 하므로 runner 사용합니다
    @Override
    public void run(ApplicationArguments args) throws Exception {
        grpcServer.start();
        grpcServer.awaitTermination();
    }

    @Override
    public void destroy() throws Exception {
        if (!ObjectUtils.isEmpty(grpcServer)) {
            grpcServer.shutdown();
        }
    }
}