Add sofarpc test casees

apm-sniffer/apm-sdk-plugin/sofarpc-plugin/src/test/java/org/apache/skywalking/apm/plugin/sofarpc/SofaBoltCallbackInterceptorTest.java

@@ -99,7 +99,7 @@ public void before() throws NoSuchMethodException {
 
             @Override
             public void onResponse(Object o) {
-                ContextManager.createLocalSpan("TestBegin");
+                ContextManager.createLocalSpan("onResponse");
                 ContextManager.stopSpan();
             }
 
@@ -173,7 +173,9 @@ public void testResponseIntegrally() throws InterruptedException {
         TraceSegment traceSegment2 = segmentStorage.getTraceSegments().get(1);
         List<AbstractTracingSpan> spans2 = SegmentHelper.getSpans(traceSegment2);
         assertEquals(2, spans2.size());
-        assertEquals("TestBegin", spans2.get(0).getOperationName());
+        assertEquals("onResponse", spans2.get(0).getOperationName());
+
+        assertEquals(traceSegment1.getRelatedGlobalTrace().getId(),traceSegment2.getRef().getTraceId());
     }
 
 }

test/plugin/scenarios/sofarpc-scenario/config/expectedData.yaml

@@ -17,6 +17,23 @@ segmentItems:
 - serviceName: sofarpc-scenario
   segmentSize: gt 2
   segments:
+  - segmentId: not null
+    spans:
+      - operationName: HEAD:/sofarpc-scenario/case/healthCheck
+        parentSpanId: -1
+        spanId: 0
+        spanLayer: Http
+        startTime: nq 0
+        endTime: nq 0
+        componentId: 1
+        isError: false
+        spanType: Entry
+        peer: ''
+        skipAnalysis: false
+        tags:
+          - {key: url, value: 'http://localhost:8080/sofarpc-scenario/case/healthCheck'}
+          - {key: http.method, value: HEAD}
+          - {key: http.status_code, value: '200'}
   - segmentId: not null
     spans:
     - operationName: org.apache.skywalking.apm.testcase.sofarpc.interfaces.SofaRpcDemoService.hello(java.lang.String)
@@ -34,6 +51,37 @@ segmentItems:
         refType: CrossProcess, parentSpanId: 1, parentTraceSegmentId: not null, parentServiceInstance: not
           null, parentService: sofarpc-scenario, traceId: not null}
       skipAnalysis: 'false'
+  - segmentId: not null
+    spans:
+    - operationName: org.apache.skywalking.apm.testcase.sofarpc.interfaces.SofaRpcDemoService.onAppResponse(java.lang.String)
+      parentSpanId: 0
+      spanId: 1
+      spanLayer: RPCFramework
+      startTime: nq 0
+      endTime: nq 0
+      componentId: 43
+      isError: false
+      spanType: Exit
+      peer: 127.0.0.1:12200
+      skipAnalysis: false
+      tags:
+      - {key: url, value: 'bolt://127.0.0.1:12200/org.apache.skywalking.apm.testcase.sofarpc.interfaces.SofaRpcDemoService.onAppResponse(java.lang.String)'}
+    - operationName: Thread/com.alipay.sofa.rpc.message.bolt.BoltInvokerCallback/onResponse
+      parentSpanId: -1
+      spanId: 0
+      spanLayer: Unknown
+      startTime: nq 0
+      endTime: nq 0
+      componentId: 0
+      isError: false
+      spanType: Local
+      peer: ''
+      skipAnalysis: false
+      refs:
+      - {parentEndpoint: 'GET:/sofarpc-scenario/case/sofarpc', networkAddress: '',
+        refType: CrossThread, parentSpanId: 2, parentTraceSegmentId: not null,
+        parentServiceInstance: not null, parentService: sofarpc-scenario,
+        traceId: not null}
   - segmentId: not null
     spans:
     - operationName: org.apache.skywalking.apm.testcase.sofarpc.interfaces.SofaRpcDemoService.hello(java.lang.String)
@@ -49,6 +97,19 @@ segmentItems:
       tags:
       - {key: url, value: 'bolt://127.0.0.1:12200/org.apache.skywalking.apm.testcase.sofarpc.interfaces.SofaRpcDemoService.hello(java.lang.String)'}
       skipAnalysis: 'false'
+    - operationName: org.apache.skywalking.apm.testcase.sofarpc.interfaces.SofaRpcDemoService.callback(java.lang.String)
+      parentSpanId: 0
+      spanId: 2
+      spanLayer: RPCFramework
+      startTime: nq 0
+      endTime: nq 0
+      componentId: 43
+      isError: false
+      spanType: Exit
+      peer: 127.0.0.1:12200
+      tags:
+        - { key: url, value: 'bolt://127.0.0.1:12200/org.apache.skywalking.apm.testcase.sofarpc.interfaces.SofaRpcDemoService.callback(java.lang.String)' }
+      skipAnalysis: 'false'
     - operationName: GET:/sofarpc-scenario/case/sofarpc
       parentSpanId: -1
       spanId: 0
@@ -64,3 +125,20 @@ segmentItems:
       - {key: http.method, value: GET}
       - {key: http.status_code, value: '200'}
       skipAnalysis: 'false'
+  - segmentId: not null
+    spans:
+    - operationName: org.apache.skywalking.apm.testcase.sofarpc.interfaces.SofaRpcDemoService.callback(java.lang.String)
+      parentSpanId: -1
+      spanId: 0
+      spanLayer: RPCFramework
+      startTime: nq 0
+      endTime: nq 0
+      componentId: 43
+      isError: false
+      spanType: Entry
+      peer: ''
+      refs:
+        - { parentEndpoint: GET:/sofarpc-scenario/case/sofarpc, networkAddress: '127.0.0.1:12200',
+          refType: CrossProcess, parentSpanId: 2, parentTraceSegmentId: not null, parentServiceInstance: not
+              null, parentService: sofarpc-scenario, traceId: not null }
+      skipAnalysis: 'false'

test/plugin/scenarios/sofarpc-scenario/src/main/java/org/apache/skywalking/apm/testcase/sofarpc/SofaRpcApplication.java

@@ -18,6 +18,7 @@
 
 package org.apache.skywalking.apm.testcase.sofarpc;
 
+import com.alipay.sofa.rpc.common.RpcConstants;
 import com.alipay.sofa.rpc.config.ConsumerConfig;
 import com.alipay.sofa.rpc.config.ProviderConfig;
 import com.alipay.sofa.rpc.config.ServerConfig;
@@ -42,8 +43,9 @@ public static class SofaRpcConfiguration {
         public ProviderConfig provider() {
             ServerConfig config = new ServerConfig().setProtocol("bolt").setPort(12200).setDaemon(true);
 
-            ProviderConfig<SofaRpcDemoService> providerConfig = new ProviderConfig<SofaRpcDemoService>().setInterfaceId(SofaRpcDemoService.class
-                .getName()).setRef(new SofaRpcDemoServiceImpl()).setServer(config);
+            ProviderConfig<SofaRpcDemoService> providerConfig = new ProviderConfig<SofaRpcDemoService>().setInterfaceId(
+                SofaRpcDemoService.class
+                    .getName()).setRef(new SofaRpcDemoServiceImpl()).setServer(config);
 
             providerConfig.export();
             return providerConfig;
@@ -55,5 +57,13 @@ public ConsumerConfig consumer() {
                                                            .setProtocol("bolt")
                                                            .setDirectUrl("bolt://127.0.0.1:12200");
         }
+
+        @Bean
+        public ConsumerConfig callbackConsumer() {
+            return new ConsumerConfig<SofaRpcDemoService>().setInterfaceId(SofaRpcDemoService.class.getName())
+                                                           .setProtocol("bolt")
+                                                           .setInvokeType(RpcConstants.INVOKER_TYPE_CALLBACK)
+                                                           .setDirectUrl("bolt://127.0.0.1:12200");
+        }
     }
 }

test/plugin/scenarios/sofarpc-scenario/src/main/java/org/apache/skywalking/apm/testcase/sofarpc/callback/TestCallback.java

@@ -0,0 +1,48 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.skywalking.apm.testcase.sofarpc.callback;
+
+import com.alipay.sofa.rpc.core.exception.SofaRpcException;
+import com.alipay.sofa.rpc.core.invoke.SofaResponseCallback;
+import com.alipay.sofa.rpc.core.request.RequestBase;
+import org.apache.skywalking.apm.testcase.sofarpc.interfaces.SofaRpcDemoService;
+
+public class TestCallback implements SofaResponseCallback {
+
+    private SofaRpcDemoService service;
+
+    public TestCallback(final SofaRpcDemoService service) {
+        this.service = service;
+    }
+
+    @Override
+    public void onAppResponse(final Object o, final String s, final RequestBase requestBase) {
+        service.onAppResponse("hello");
+    }
+
+    @Override
+    public void onAppException(final Throwable throwable, final String s, final RequestBase requestBase) {
+
+    }
+
+    @Override
+    public void onSofaException(final SofaRpcException e, final String s, final RequestBase requestBase) {
+
+    }
+}

test/plugin/scenarios/sofarpc-scenario/src/main/java/org/apache/skywalking/apm/testcase/sofarpc/controller/CaseController.java

@@ -19,8 +19,11 @@
 package org.apache.skywalking.apm.testcase.sofarpc.controller;
 
 import com.alipay.sofa.rpc.config.ConsumerConfig;
+import com.alipay.sofa.rpc.context.RpcInvokeContext;
+import org.apache.skywalking.apm.testcase.sofarpc.callback.TestCallback;
 import org.apache.skywalking.apm.testcase.sofarpc.interfaces.SofaRpcDemoService;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.web.bind.annotation.RequestMapping;
 import org.springframework.web.bind.annotation.ResponseBody;
 import org.springframework.web.bind.annotation.RestController;
@@ -32,8 +35,13 @@ public class CaseController {
     private static final String SUCCESS = "Success";
 
     @Autowired
+    @Qualifier("consumer")
     private ConsumerConfig<SofaRpcDemoService> consumerConfig;
 
+    @Autowired
+    @Qualifier("callbackConsumer")
+    private ConsumerConfig<SofaRpcDemoService> callbackConsumer;
+
     @RequestMapping("/healthCheck")
     @ResponseBody
     public String healthCheck() {
@@ -45,6 +53,11 @@ public String healthCheck() {
     public String sofarpc() {
         SofaRpcDemoService service = consumerConfig.refer();
         service.hello("sofarpc");
+
+        SofaRpcDemoService callbackService = callbackConsumer.refer();
+        RpcInvokeContext invokeCtx = RpcInvokeContext.peekContext();
+        invokeCtx.setResponseCallback(new TestCallback(service));
+        callbackService.callback("sofarpc");
         return SUCCESS;
     }
 }

test/plugin/scenarios/sofarpc-scenario/src/main/java/org/apache/skywalking/apm/testcase/sofarpc/interfaces/SofaRpcDemoService.java

@@ -21,4 +21,8 @@
 public interface SofaRpcDemoService {
 
     String hello(String name);
+
+    String onAppResponse(String name);
+
+    String callback(String name);
 }

test/plugin/scenarios/sofarpc-scenario/src/main/java/org/apache/skywalking/apm/testcase/sofarpc/service/SofaRpcDemoServiceImpl.java

@@ -25,4 +25,14 @@ public class SofaRpcDemoServiceImpl implements SofaRpcDemoService {
     public String hello(String name) {
         return "hello, " + name;
     }
+
+    @Override
+    public String onAppResponse(String name) {
+        return "hello, " + name;
+    }
+
+    @Override
+    public String callback(String name) {
+        return "hello, " + name;
+    }
 }