@Override
public void encode(Channel channel, ChannelBuffer buffer, Object message) throws IOException {
    if (message instanceof Request) {
        Object data = ((Request) message).getData();
        if (data instanceof RpcInvocation) {
            RpcInvocation invocation = (RpcInvocation) data;
            encodeRequestWithTracer(channel, buffer, message, invocation);
            return;
        }
    } else if (message instanceof Response) {
        Object response = ((Response) message).getResult();
        if (response instanceof AppResponse) {
            encodeResultWithTracer(channel, buffer, message);
            return;
        }
    }
    codec.encode(channel, buffer, message);
}
 

/**
 * 调用方法
 *
 * @param c
 * @param method
 * @param fullUrl
 * @param args
 * @return
 * @throws java.lang.Exception
 */
public static Object invoke(Class c, Method method, String fullUrl, Object... args) throws Exception {
    URL url = URL.valueOf(fullUrl);
    AsyncToSyncInvoker<?> invoker = (AsyncToSyncInvoker<?>) PROTOCOL.refer(c, url);
    if (invoker.isAvailable()) {
        Invocation inv = new RpcInvocation(method, url.getParameter("interface"), args);
        Result ret = invoker.invoke(inv);
        PROTOCOL.destroy();
        return JSON.json(ret.getValue());
    }
    return null;
}
 

/**
 * @param channel       a long connection
 * @param buffer        buffer
 * @param message       the original Request object
 * @param invocation    Invocation in Request
 * @throws IOException  serialization exception
 */
protected void encodeRequestWithTracer(Channel channel, ChannelBuffer buffer, Object message,
                                       RpcInvocation invocation) throws IOException {
    long startTime = System.currentTimeMillis();
    int index = buffer.writerIndex();
    // serialization
    codec.encode(channel, buffer, message);
    int reqSize = buffer.writerIndex() - index;
    long elapsed = System.currentTimeMillis() - startTime;
    invocation.setAttachment(AttachmentKeyConstants.CLIENT_SERIALIZE_SIZE,
        String.valueOf(reqSize));
    invocation.setAttachment(AttachmentKeyConstants.CLIENT_SERIALIZE_TIME,
        String.valueOf(elapsed));
}
 

/**
 * deserialization operation
 * @param channel
 * @param input
 * @return
 * @throws IOException
 */
@Override
public Object decode(Channel channel, ChannelBuffer input) throws IOException {
    long startTime = System.currentTimeMillis();
    int index = input.readerIndex();
    Object ret = codec.decode(channel, input);
    int size = input.readerIndex() - index;
    long elapsed = System.currentTimeMillis() - startTime;
    if (ret instanceof Request) {
        // server-side deserialize the Request
        Object data = ((Request) ret).getData();
        if (data instanceof RpcInvocation) {
            RpcInvocation invocation = (RpcInvocation) data;
            invocation.setAttachment(AttachmentKeyConstants.SERVER_DESERIALIZE_SIZE,
                String.valueOf(size));
            invocation.setAttachment(AttachmentKeyConstants.SERVER_DESERIALIZE_TIME,
                String.valueOf(elapsed));
        }
    } else if (ret instanceof Response) {
        // client-side deserialize the Response
        Object result = ((Response) ret).getResult();
        if (result instanceof AppResponse) {
            AppResponse rpcResult = (AppResponse) result;
            rpcResult.setAttachment(AttachmentKeyConstants.CLIENT_DESERIALIZE_SIZE,
                String.valueOf(size));
            rpcResult.setAttachment(AttachmentKeyConstants.CLIENT_DESERIALIZE_TIME,
                String.valueOf(elapsed));
        }
    }
    return ret;
}
 

@Override
public void after(Object target, Object[] args, Object result, Throwable throwable) {
    if (isDebug) {
        logger.afterInterceptor(target, args);
    }

    // Ignore monitor service
    if (isMonitorService(target)) {
        return;
    }

    final Trace trace = traceContext.currentTraceObject();
    if (trace == null) {
        return;
    }

    try {
        final RpcInvocation invocation = (RpcInvocation) args[0];
        final SpanEventRecorder recorder = trace.currentSpanEventRecorder();
        recorder.recordApi(descriptor);
        if (throwable == null) {
            String endPoint = RpcContext.getContext().getRemoteAddressString();
            // RPC client have to record end point (server address)
            recorder.recordEndPoint(endPoint);

            // Optionally, record the destination id (logical name of server. e.g. DB name)
            recorder.recordDestinationId(endPoint);
            recorder.recordAttribute(ApacheDubboConstants.DUBBO_ARGS_ANNOTATION_KEY, invocation.getArguments());
            recorder.recordAttribute(ApacheDubboConstants.DUBBO_RESULT_ANNOTATION_KEY, result);
        } else {
            recorder.recordException(throwable);
        }
    } finally {
        trace.traceBlockEnd();
    }
}
 

private Trace readRequestTrace(Object target, Object[] args) {
    final Invoker invoker = (Invoker) target;
    // Ignore monitor service.
    if (ApacheDubboConstants.MONITOR_SERVICE_FQCN.equals(invoker.getInterface().getName())) {
        return traceContext.disableSampling();
    }

    final RpcInvocation invocation = (RpcInvocation) args[0];
    // If this transaction is not traceable, mark as disabled.
    if (invocation.getAttachment(ApacheDubboConstants.META_DO_NOT_TRACE) != null) {
        return traceContext.disableSampling();
    }
    final String transactionId = invocation.getAttachment(ApacheDubboConstants.META_TRANSACTION_ID);
    // If there's no trasanction id, a new trasaction begins here.
    // FIXME There seems to be cases where the invoke method is called after a span is already created.
    // We'll have to check if a trace object already exists and create a span event instead of a span in that case.
    if (transactionId == null) {
        return traceContext.newTraceObject();
    }

    // otherwise, continue tracing with given data.
    final long parentSpanID = NumberUtils.parseLong(invocation.getAttachment(ApacheDubboConstants.META_PARENT_SPAN_ID), SpanId.NULL);
    final long spanID = NumberUtils.parseLong(invocation.getAttachment(ApacheDubboConstants.META_SPAN_ID), SpanId.NULL);
    final short flags = NumberUtils.parseShort(invocation.getAttachment(ApacheDubboConstants.META_FLAGS), (short) 0);
    final TraceId traceId = traceContext.createTraceId(transactionId, parentSpanID, spanID, flags);

    return traceContext.continueTraceObject(traceId);
}
 

private void recordRequest(SpanRecorder recorder, Object target, Object[] args) {
    final RpcInvocation invocation = (RpcInvocation) args[0];
    final RpcContext rpcContext = RpcContext.getContext();

    // Record rpc name, client address, server address.
    recorder.recordRpcName(invocation.getInvoker().getInterface().getSimpleName() + ":" + invocation.getMethodName());
    recorder.recordEndPoint(rpcContext.getLocalAddressString());
    if (rpcContext.getRemoteHost() != null) {
        recorder.recordRemoteAddress(rpcContext.getRemoteAddressString());
    } else {
        recorder.recordRemoteAddress("Unknown");
    }

    // If this transaction did not begin here, record parent(client who sent this request) information
    if (!recorder.isRoot()) {
        final String parentApplicationName = invocation.getAttachment(ApacheDubboConstants.META_PARENT_APPLICATION_NAME);
        if (parentApplicationName != null) {
            final short parentApplicationType = NumberUtils.parseShort(invocation.getAttachment(ApacheDubboConstants.META_PARENT_APPLICATION_TYPE), ServiceType.UNDEFINED.getCode());
            recorder.recordParentApplication(parentApplicationName, parentApplicationType);

            final String host = invocation.getAttachment(ApacheDubboConstants.META_HOST);
            if (host != null) {
                recorder.recordAcceptorHost(host);
            } else {
                // old version fallback
                final String estimatedLocalHost = getLocalHost(rpcContext);
                if (estimatedLocalHost != null) {
                    recorder.recordAcceptorHost(estimatedLocalHost);
                }
            }
        }
    }
    //clear attachments
    this.clearAttachments(rpcContext);
}
 

@Override
protected void doInBeforeTrace(SpanEventRecorder recorder, Object target, Object[] args) {
    final RpcInvocation invocation = (RpcInvocation) args[0];
    recorder.recordServiceType(ApacheDubboConstants.DUBBO_PROVIDER_SERVICE_NO_STATISTICS_TYPE);
    recorder.recordApi(methodDescriptor);
    recorder.recordAttribute(ApacheDubboConstants.DUBBO_RPC_ANNOTATION_KEY,
            invocation.getInvoker().getInterface().getSimpleName() + ":" + invocation.getMethodName());
}
 

@Override
protected void doInAfterTrace(SpanEventRecorder recorder, Object target, Object[] args, Object result, Throwable throwable) {
    final RpcInvocation invocation = (RpcInvocation) args[0];
    recorder.recordServiceType(ApacheDubboConstants.DUBBO_PROVIDER_SERVICE_NO_STATISTICS_TYPE);
    recorder.recordApi(methodDescriptor);
    recorder.recordAttribute(ApacheDubboConstants.DUBBO_ARGS_ANNOTATION_KEY, invocation.getArguments());

    if (throwable == null) {
        recorder.recordAttribute(ApacheDubboConstants.DUBBO_RESULT_ANNOTATION_KEY, result);
    } else {
        recorder.recordException(throwable);
    }
}
 

@Test
public void before() {
    doReturn(trace).when(traceContext).currentRawTraceObject();
    doReturn(true).when(trace).canSampled();
    doReturn(traceId).when(trace).getTraceId();
    doReturn(nextId).when(traceId).getNextTraceId();
    doReturn(spanRecorder).when(trace).traceBlockBegin();

    ApacheDubboConsumerInterceptor interceptor = new ApacheDubboConsumerInterceptor(traceContext, descriptor);
    RpcInvocation rpcInvocation = new RpcInvocation();
    rpcInvocation.setInvoker(new DubboInvoker(Object.class, new URL("http", "127.0.0.1", 8080), null));
    rpcInvocation.setMethodName("test");
    Object[] args = new Object[]{rpcInvocation};
    interceptor.before(obj, args);
}
 

@Test
public void after() {
    doReturn(trace).when(traceContext).currentTraceObject();
    doReturn(spanRecorder).when(trace).currentSpanEventRecorder();

    RpcInvocation rpcInvocation = new RpcInvocation();
    Object[] args = new Object[]{rpcInvocation};
    ApacheDubboConsumerInterceptor interceptor = new ApacheDubboConsumerInterceptor(traceContext, descriptor);
    interceptor.after(obj, args, null, null);
}
 

@Test
public void createTrace() {
    doReturn(true).when(trace).canSampled();
    doReturn(spanRecorder).when(trace).getSpanRecorder();
    doReturn(trace).when(traceContext).newTraceObject();

    Invoker invoker = new DubboInvoker(Object.class, new URL("http", "127.0.0.1", 8080), null);
    ApacheDubboProviderInterceptor interceptor = new ApacheDubboProviderInterceptor(traceContext, descriptor);
    RpcInvocation rpcInvocation = new RpcInvocation();
    rpcInvocation.setInvoker(invoker);
    rpcInvocation.setMethodName("test");
    rpcInvocation.setAttachment(ApacheDubboConstants.META_PARENT_APPLICATION_NAME, UUID.randomUUID().toString());
    Object[] args = new Object[]{rpcInvocation};
    interceptor.createTrace(invoker, args);
}
 

@Test
public void doInBeforeTrace() {
    ApacheDubboProviderInterceptor interceptor = new ApacheDubboProviderInterceptor(traceContext, descriptor);
    RpcInvocation rpcInvocation = new RpcInvocation();
    rpcInvocation.setInvoker(new DubboInvoker(Object.class, new URL("http", "127.0.0.1", 8080), null));
    rpcInvocation.setMethodName("test");
    Object[] args = new Object[]{rpcInvocation};
    interceptor.doInBeforeTrace(recorder, obj, args);
}
 

@Override
public void before(Object target, Object[] args) {
    if (isDebug) {
        logger.beforeInterceptor(target, args);
    }

    // Ignore monitor service
    if (isMonitorService(target)) {
        return;
    }

    final Trace trace = traceContext.currentRawTraceObject();
    if (trace == null) {
        return;
    }

    final RpcInvocation invocation = (RpcInvocation) args[0];

    if (trace.canSampled()) {
        final SpanEventRecorder recorder = trace.traceBlockBegin();

        // RPC call trace have to be recorded with a service code in RPC client code range.
        recorder.recordServiceType(ApacheDubboConstants.DUBBO_CONSUMER_SERVICE_TYPE);

        // You have to issue a TraceId the receiver of this request will use.
        final TraceId nextId = trace.getTraceId().getNextTraceId();

        // Then record it as next span id.
        recorder.recordNextSpanId(nextId.getSpanId());

        // Finally, pass some tracing data to the server.
        // How to put them in a message is protocol specific.
        // This example assumes that the target protocol message can include any metadata (like HTTP headers).
        setAttachment(invocation, ApacheDubboConstants.META_TRANSACTION_ID, nextId.getTransactionId());
        setAttachment(invocation, ApacheDubboConstants.META_SPAN_ID, Long.toString(nextId.getSpanId()));
        setAttachment(invocation, ApacheDubboConstants.META_PARENT_SPAN_ID, Long.toString(nextId.getParentSpanId()));
        setAttachment(invocation, ApacheDubboConstants.META_PARENT_APPLICATION_TYPE, Short.toString(traceContext.getServerTypeCode()));
        setAttachment(invocation, ApacheDubboConstants.META_PARENT_APPLICATION_NAME, traceContext.getApplicationName());
        setAttachment(invocation, ApacheDubboConstants.META_FLAGS, Short.toString(nextId.getFlags()));

        setAttachment(invocation, ApacheDubboConstants.META_HOST, getHostAddress(invocation));
    } else {
        // If sampling this transaction is disabled, pass only that infomation to the server.
        setAttachment(invocation, ApacheDubboConstants.META_DO_NOT_TRACE, "1");
    }
}
 

private String getHostAddress(RpcInvocation invocation) {
    final URL url = invocation.getInvoker().getUrl();
    return HostAndPort.toHostAndPortString(url.getHost(), url.getPort());
}
 

private void setAttachment(RpcInvocation invocation, String name, String value) {
    invocation.setAttachment(name, value);
    if (isDebug) {
        logger.debug("Set attachment {}={}", name, value);
    }
}