/*
 * Copyright 2019 NAVER Corp.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.navercorp.pinpoint.profiler.sender.grpc;


import com.navercorp.pinpoint.common.util.Assert;
import com.navercorp.pinpoint.grpc.client.ChannelFactoryOption;

import com.google.protobuf.Empty;

import com.navercorp.pinpoint.grpc.trace.PSpan;
import com.navercorp.pinpoint.grpc.trace.PSpanChunk;
import com.navercorp.pinpoint.grpc.trace.PSpanMessage;
import com.navercorp.pinpoint.grpc.trace.SpanGrpc;
import com.navercorp.pinpoint.profiler.context.thrift.MessageConverter;

import com.google.protobuf.GeneratedMessageV3;
import io.grpc.stub.StreamObserver;

import java.util.concurrent.RejectedExecutionException;

import static com.navercorp.pinpoint.grpc.MessageFormatUtils.debugLog;

/**
 * @author jaehong.kim
 */
public class SpanGrpcDataSender extends GrpcDataSender {
    private final SpanGrpc.SpanStub spanStub;
    private final ReconnectExecutor reconnectExecutor;

    private volatile StreamObserver<PSpanMessage> spanStream;
    private final Reconnector spanStreamReconnector;

    public SpanGrpcDataSender(String host, int port,
                              int executorQueueSize,
                              MessageConverter<GeneratedMessageV3> messageConverter,
                              ReconnectExecutor reconnectExecutor,
                              ChannelFactoryOption channelFactoryOption) {
        super(host, port, executorQueueSize, messageConverter, channelFactoryOption);

        this.spanStub = SpanGrpc.newStub(managedChannel);
        this.reconnectExecutor = Assert.requireNonNull(reconnectExecutor, "reconnectExecutor must not be null");
        {
            final Runnable spanStreamReconnectJob = new Runnable() {
                @Override
                public void run() {
                    spanStream = newSpanStream();
                }
            };
            this.spanStreamReconnector = reconnectExecutor.newReconnector(spanStreamReconnectJob);
            this.spanStream = newSpanStream();
        }
    }

    private StreamObserver<PSpanMessage> newSpanStream() {
        StreamId spanId = StreamId.newStreamId("SpanStream");
        ResponseStreamObserver<PSpanMessage, Empty> responseStreamObserver = new ResponseStreamObserver<PSpanMessage, Empty>(spanId, spanStreamReconnector);
        return spanStub.sendSpan(responseStreamObserver);
    }

    @Override
    public boolean send(final Object data) {
        final Runnable command = new Runnable() {
            @Override
            public void run() {
                try {
                    send0(data);
                } catch (Exception ex) {
                    logger.debug("send fail:{}", data, ex);
                }
            }
        };
        try {
            executor.execute(command);
        } catch (RejectedExecutionException reject) {
            logger.debug("reject:{}", command);
            return false;
        }
        return true;
    }

    private boolean send0(Object data) {
        final GeneratedMessageV3 message = messageConverter.toMessage(data);
        if (logger.isDebugEnabled()) {
            logger.debug("Send message={}", debugLog(message));
        }
        if (message instanceof PSpanChunk) {
            final PSpanChunk spanChunk = (PSpanChunk) message;
            final PSpanMessage spanMessage = PSpanMessage.newBuilder().setSpanChunk(spanChunk).build();
            spanStream.onNext(spanMessage);
            return true;
        }
        if (message instanceof PSpan) {
            final PSpan pSpan = (PSpan) message;
            final PSpanMessage spanMessage = PSpanMessage.newBuilder().setSpan(pSpan).build();
            spanStream.onNext(spanMessage);
            return true;
        }
        throw new IllegalStateException("unsupported message " + data);
    }

    @Override
    public void stop() {
        if (shutdown) {
            return;
        }
        this.shutdown = true;

        logger.info("Stop {}, channel={}", name, managedChannel);
        final ReconnectExecutor reconnectExecutor = this.reconnectExecutor;
        if (reconnectExecutor != null) {
            reconnectExecutor.close();
        }
        logger.info("{} close()", this.spanStream);
        StreamUtils.close(this.spanStream);
        release();
    }

    @Override
    public String toString() {
        return "SpanGrpcDataSender{" +
                "name='" + name + '\'' +
                ", host='" + host + '\'' +
                ", port=" + port +
                "} " + super.toString();
    }
}
@@ -1141,29 +1141,286 @@
 rpc.
-stub.StreamObserver;%0A
+CallOptions;%0Aimport io.grpc.Channel;%0Aimport io.grpc.ClientCall;%0Aimport io.grpc.ClientInterceptor;%0Aimport io.grpc.ForwardingClientCall;%0Aimport io.grpc.MethodDescriptor;%0Aimport io.grpc.stub.CallStreamObserver;%0Aimport io.grpc.stub.StreamObserver;%0A%0Aimport javax.annotation.Nullable;
 %0Aimp
@@ -2321,32 +2321,1103 @@
 anagedChannel);%0A
+        this.spanStub.withInterceptors(new ClientInterceptor() %7B%0A            @Override%0A            public %3CReqT, RespT%3E ClientCall%3CReqT, RespT%3E interceptCall(MethodDescriptor%3CReqT, RespT%3E method, CallOptions callOptions, Channel next) %7B%0A                return new ForwardingClientCall.SimpleForwardingClientCall%3CReqT, RespT%3E(next.newCall(method, callOptions)) %7B%0A                    @Override%0A                    public void sendMessage(ReqT message) %7B%0A                        if (isReady()) %7B%0A                            super.sendMessage(message);%0A                        %7D else %7B%0A                            logger.info(%22Drop span message, stream not ready.%22);%0A                        %7D%0A                    %7D%0A%0A                    @Override%0A                    public void cancel(@Nullable String message, @Nullable Throwable cause) %7B%0A                        logger.info(%22Cancel message. message=%7B%7D, cause=%7B%7D%22, message, cause.getMessage(), cause);%0A                        super.cancel(message, cause);%0A                    %7D%0A                %7D;%0A            %7D%0A        %7D);%0A%0A
         this.rec

Add grpc client interceptor of SpanGrpcDataSender