[pinpoint-apm#8941] Update reactor plugin for subscribeOrReturn method.

bootstraps/bootstrap-core/src/main/java/com/navercorp/pinpoint/bootstrap/plugin/reactor/FluxAndMonoConstructorInterceptor.java

@@ -22,7 +22,7 @@
 import com.navercorp.pinpoint.bootstrap.logging.PLogger;
 import com.navercorp.pinpoint.bootstrap.logging.PLoggerFactory;
 
-public abstract class FluxAndMonoConstructorInterceptor implements AroundInterceptor {
+public class FluxAndMonoConstructorInterceptor implements AroundInterceptor {
     private final PLogger logger = PLoggerFactory.getLogger(getClass());
     private final boolean isDebug = logger.isDebugEnabled();
 

bootstraps/bootstrap-core/src/main/java/com/navercorp/pinpoint/bootstrap/plugin/reactor/FluxAndMonoSubscribeOrReturnInterceptor.java

@@ -0,0 +1,93 @@
+/*
+ * Copyright 2022 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.bootstrap.plugin.reactor;
+
+import com.navercorp.pinpoint.bootstrap.async.AsyncContextAccessorUtils;
+import com.navercorp.pinpoint.bootstrap.context.AsyncContext;
+import com.navercorp.pinpoint.bootstrap.context.MethodDescriptor;
+import com.navercorp.pinpoint.bootstrap.context.TraceContext;
+import com.navercorp.pinpoint.bootstrap.interceptor.AroundInterceptor;
+import com.navercorp.pinpoint.bootstrap.logging.PLogger;
+import com.navercorp.pinpoint.bootstrap.logging.PLoggerFactory;
+
+public class FluxAndMonoSubscribeOrReturnInterceptor implements AroundInterceptor {
+    private final PLogger logger = PLoggerFactory.getLogger(getClass());
+    private final boolean isDebug = logger.isDebugEnabled();
+
+    public FluxAndMonoSubscribeOrReturnInterceptor() {
+    }
+
+    @Override
+    public void before(Object target, Object[] args) {
+    }
+
+    @Override
+    public void after(Object target, Object[] args, Object result, Throwable throwable) {
+        if (isDebug) {
+            logger.afterInterceptor(target, args, result, throwable);
+        }
+
+        if (throwable != null) {
+            // Ignore if an error occurs.
+            return;
+        }
+        if (result == null) {
+            return;
+        }
+
+        if (checkTargetReactorContextAccessor(target, args, result)) {
+            return;
+        }
+        if (checkTargetAsyncContextAccessor(target, args, result)) {
+            return;
+        }
+        if (checkSubscriberReactorContextAccessor(target, args, result)) {
+            return;
+        }
+    }
+
+    boolean checkTargetReactorContextAccessor(final Object target, final Object[] args, final Object result) {
+        final AsyncContext asyncContext = ReactorContextAccessorUtils.getAsyncContext(target);
+        if (asyncContext != null) {
+            setReactorContextToResult(asyncContext, result);
+            return true;
+        }
+        return false;
+    }
+
+    boolean checkTargetAsyncContextAccessor(final Object target, final Object[] args, final Object result) {
+        final AsyncContext asyncContext = AsyncContextAccessorUtils.getAsyncContext(target);
+        if (asyncContext != null) {
+            setReactorContextToResult(asyncContext, result);
+            return true;
+        }
+        return false;
+    }
+
+    boolean checkSubscriberReactorContextAccessor(final Object target, final Object[] args, final Object result) {
+        final AsyncContext asyncContext = ReactorContextAccessorUtils.getAsyncContext(args, 0);
+        if (asyncContext != null) {
+            setReactorContextToResult(asyncContext, result);
+            return true;
+        }
+        return false;
+    }
+
+    protected void setReactorContextToResult(AsyncContext asyncContext, Object result) {
+        ReactorContextAccessorUtils.setAsyncContext(asyncContext, result);
+    }
+}

bootstraps/bootstrap-core/src/test/java/com/navercorp/pinpoint/bootstrap/plugin/reactor/CoreSubscriberConstructorInterceptorTest.java

@@ -0,0 +1,83 @@
+/*
+ * Copyright 2022 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.bootstrap.plugin.reactor;
+
+import com.navercorp.pinpoint.bootstrap.context.AsyncContext;
+import org.junit.Test;
+
+import static org.junit.Assert.*;
+import static org.mockito.Mockito.mock;
+
+public class CoreSubscriberConstructorInterceptorTest {
+
+    @Test
+    public void arg0ContainReactorContext() {
+        AsyncContext mockAsyncContext = mock(AsyncContext.class);
+        MockAsyncContextAndReactorContextImpl target = new MockAsyncContextAndReactorContextImpl();
+        MockAsyncContextAndReactorContextImpl arg0 = new MockAsyncContextAndReactorContextImpl();
+        CoreSubscriberConstructorInterceptor interceptor = new CoreSubscriberConstructorInterceptor();
+
+        // Set asyncContext to target
+        arg0._$PINPOINT$_setReactorContext(mockAsyncContext);
+        interceptor.after(target, new Object[]{arg0}, new Object(), null);
+
+        assertNotNull(target._$PINPOINT$_getReactorContext());
+        assertEquals(target._$PINPOINT$_getReactorContext(), arg0._$PINPOINT$_getReactorContext());
+    }
+
+    @Test
+    public void argNotContainReactorContext() {
+        MockAsyncContextAndReactorContextImpl target = new MockAsyncContextAndReactorContextImpl();
+        MockAsyncContextAndReactorContextImpl arg0 = new MockAsyncContextAndReactorContextImpl();
+        CoreSubscriberConstructorInterceptor interceptor = new CoreSubscriberConstructorInterceptor();
+
+        // Not set asyncContext to target
+        interceptor.after(target, new Object[]{arg0}, new Object(), null);
+
+        assertNull(target._$PINPOINT$_getReactorContext());
+    }
+
+    @Test
+    public void arg1ContainReactorContext() {
+        AsyncContext mockAsyncContext = mock(AsyncContext.class);
+        MockAsyncContextAndReactorContextImpl target = new MockAsyncContextAndReactorContextImpl();
+        MockAsyncContextAndReactorContextImpl arg0 = new MockAsyncContextAndReactorContextImpl();
+        MockAsyncContextAndReactorContextImpl arg1 = new MockAsyncContextAndReactorContextImpl();
+        CoreSubscriberConstructorInterceptor interceptor = new CoreSubscriberConstructorInterceptor();
+
+        // Set asyncContext to target
+        arg1._$PINPOINT$_setReactorContext(mockAsyncContext);
+        interceptor.after(target, new Object[]{arg0, arg1}, new Object(), null);
+
+        assertNotNull(target._$PINPOINT$_getReactorContext());
+        assertEquals(target._$PINPOINT$_getReactorContext(), arg1._$PINPOINT$_getReactorContext());
+    }
+
+    @Test
+    public void throwableIsNotNull() {
+        AsyncContext mockAsyncContext = mock(AsyncContext.class);
+        MockAsyncContextAndReactorContextImpl target = new MockAsyncContextAndReactorContextImpl();
+        MockAsyncContextAndReactorContextImpl arg0 = new MockAsyncContextAndReactorContextImpl();
+        Throwable throwable = new Throwable("ERROR");
+        CoreSubscriberConstructorInterceptor interceptor = new CoreSubscriberConstructorInterceptor();
+
+        arg0._$PINPOINT$_setReactorContext(mockAsyncContext);
+        interceptor.after(target, new Object[]{arg0}, new Object(), throwable);
+
+        assertNull(target._$PINPOINT$_getReactorContext());
+    }
+}

bootstraps/bootstrap-core/src/test/java/com/navercorp/pinpoint/bootstrap/plugin/reactor/FluxAndMonoConstructorInterceptorTest.java

@@ -0,0 +1,88 @@
+/*
+ * Copyright 2022 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.bootstrap.plugin.reactor;
+
+import com.navercorp.pinpoint.bootstrap.context.AsyncContext;
+import org.junit.Test;
+
+import static org.junit.Assert.*;
+import static org.mockito.Mockito.mock;
+
+public class FluxAndMonoConstructorInterceptorTest {
+
+    @Test
+    public void arg0ContainAsyncContext() {
+        AsyncContext mockAsyncContext = mock(AsyncContext.class);
+        MockAsyncContextAndReactorContextImpl target = new MockAsyncContextAndReactorContextImpl();
+        MockAsyncContextAndReactorContextImpl arg0 = new MockAsyncContextAndReactorContextImpl();
+        FluxAndMonoConstructorInterceptor interceptor = new FluxAndMonoConstructorInterceptor();
+
+        // Set asyncContext to target
+        arg0._$PINPOINT$_setAsyncContext(mockAsyncContext);
+        interceptor.after(target, new Object[]{arg0}, new Object(), null);
+
+        assertNotNull(target._$PINPOINT$_getAsyncContext());
+        assertEquals(target._$PINPOINT$_getAsyncContext(), arg0._$PINPOINT$_getAsyncContext());
+        assertNotNull(target._$PINPOINT$_getReactorContext());
+        assertEquals(target._$PINPOINT$_getReactorContext(), arg0._$PINPOINT$_getAsyncContext());
+    }
+
+    @Test
+    public void arg0NotContainAsyncContext() {
+        MockAsyncContextAndReactorContextImpl target = new MockAsyncContextAndReactorContextImpl();
+        MockAsyncContextAndReactorContextImpl arg0 = new MockAsyncContextAndReactorContextImpl();
+        FluxAndMonoConstructorInterceptor interceptor = new FluxAndMonoConstructorInterceptor();
+
+        // Not set asyncContext to target
+        interceptor.after(target, new Object[]{arg0}, new Object(), null);
+
+        assertNull(target._$PINPOINT$_getAsyncContext());
+        assertNull(target._$PINPOINT$_getReactorContext());
+    }
+
+    @Test
+    public void arg1ContainAsyncContext() {
+        AsyncContext mockAsyncContext = mock(AsyncContext.class);
+        MockAsyncContextAndReactorContextImpl target = new MockAsyncContextAndReactorContextImpl();
+        MockAsyncContextAndReactorContextImpl arg0 = new MockAsyncContextAndReactorContextImpl();
+        MockAsyncContextAndReactorContextImpl arg1 = new MockAsyncContextAndReactorContextImpl();
+        FluxAndMonoConstructorInterceptor interceptor = new FluxAndMonoConstructorInterceptor();
+
+        // Set asyncContext to target
+        arg1._$PINPOINT$_setAsyncContext(mockAsyncContext);
+        interceptor.after(target, new Object[]{arg0, arg1}, new Object(), null);
+
+        assertNotNull(target._$PINPOINT$_getAsyncContext());
+        assertEquals(target._$PINPOINT$_getAsyncContext(), arg1._$PINPOINT$_getAsyncContext());
+        assertNotNull(target._$PINPOINT$_getReactorContext());
+        assertEquals(target._$PINPOINT$_getReactorContext(), arg1._$PINPOINT$_getAsyncContext());
+    }
+
+    @Test
+    public void throwableIsNotNull() {
+        AsyncContext mockAsyncContext = mock(AsyncContext.class);
+        MockAsyncContextAndReactorContextImpl target = new MockAsyncContextAndReactorContextImpl();
+        MockAsyncContextAndReactorContextImpl arg0 = new MockAsyncContextAndReactorContextImpl();
+        Throwable throwable = new Throwable("ERROR");
+        FluxAndMonoConstructorInterceptor interceptor = new FluxAndMonoConstructorInterceptor();
+
+        interceptor.after(target, new Object[]{arg0}, new Object(), throwable);
+
+        assertNull(target._$PINPOINT$_getAsyncContext());
+        assertNull(target._$PINPOINT$_getReactorContext());
+    }
+}

bootstraps/bootstrap-core/src/test/java/com/navercorp/pinpoint/bootstrap/plugin/reactor/FluxAndMonoOperatorConstructorInterceptorTest.java

@@ -0,0 +1,103 @@
+/*
+ * Copyright 2022 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.bootstrap.plugin.reactor;
+
+import com.navercorp.pinpoint.bootstrap.context.AsyncContext;
+import org.junit.Test;
+
+import static org.junit.Assert.*;
+import static org.mockito.Mockito.mock;
+
+public class FluxAndMonoOperatorConstructorInterceptorTest {
+
+    @Test
+    public void arg0ContainAsyncContext() {
+        AsyncContext mockAsyncContext = mock(AsyncContext.class);
+        MockAsyncContextAndReactorContextImpl target = new MockAsyncContextAndReactorContextImpl();
+        MockAsyncContextAndReactorContextImpl arg0 = new MockAsyncContextAndReactorContextImpl();
+        FluxAndMonoOperatorConstructorInterceptor interceptor = new FluxAndMonoOperatorConstructorInterceptor();
+
+        // Set asyncContext to target
+        arg0._$PINPOINT$_setAsyncContext(mockAsyncContext);
+        interceptor.after(target, new Object[]{arg0}, new Object(), null);
+
+        assertNull(target._$PINPOINT$_getAsyncContext());
+        assertNotNull(target._$PINPOINT$_getReactorContext());
+        assertEquals(target._$PINPOINT$_getReactorContext(), mockAsyncContext);
+    }
+
+    @Test
+    public void arg0ContainReactorContext() {
+        AsyncContext mockAsyncContext = mock(AsyncContext.class);
+        MockAsyncContextAndReactorContextImpl target = new MockAsyncContextAndReactorContextImpl();
+        MockAsyncContextAndReactorContextImpl arg0 = new MockAsyncContextAndReactorContextImpl();
+        FluxAndMonoOperatorConstructorInterceptor interceptor = new FluxAndMonoOperatorConstructorInterceptor();
+
+        // Set asyncContext to target
+        arg0._$PINPOINT$_setReactorContext(mockAsyncContext);
+        interceptor.after(target, new Object[]{arg0}, new Object(), null);
+
+        assertNull(target._$PINPOINT$_getAsyncContext());
+        assertNotNull(target._$PINPOINT$_getReactorContext());
+        assertEquals(target._$PINPOINT$_getReactorContext(), mockAsyncContext);
+    }
+
+    @Test
+    public void arg0NotContainAsyncContext() {
+        MockAsyncContextAndReactorContextImpl target = new MockAsyncContextAndReactorContextImpl();
+        MockAsyncContextAndReactorContextImpl arg0 = new MockAsyncContextAndReactorContextImpl();
+        FluxAndMonoOperatorConstructorInterceptor interceptor = new FluxAndMonoOperatorConstructorInterceptor();
+
+        // Not set asyncContext to target
+        interceptor.after(target, new Object[]{arg0}, new Object(), null);
+
+        assertNull(target._$PINPOINT$_getAsyncContext());
+        assertNull(target._$PINPOINT$_getReactorContext());
+    }
+
+    @Test
+    public void arg1ContainAsyncContext() {
+        AsyncContext mockAsyncContext = mock(AsyncContext.class);
+        MockAsyncContextAndReactorContextImpl target = new MockAsyncContextAndReactorContextImpl();
+        MockAsyncContextAndReactorContextImpl arg0 = new MockAsyncContextAndReactorContextImpl();
+        MockAsyncContextAndReactorContextImpl arg1 = new MockAsyncContextAndReactorContextImpl();
+        FluxAndMonoOperatorConstructorInterceptor interceptor = new FluxAndMonoOperatorConstructorInterceptor();
+
+        // Set asyncContext to target
+        arg1._$PINPOINT$_setAsyncContext(mockAsyncContext);
+        interceptor.after(target, new Object[]{arg0, arg1}, new Object(), null);
+
+        assertNotNull(target._$PINPOINT$_getAsyncContext());
+        assertEquals(target._$PINPOINT$_getAsyncContext(), mockAsyncContext);
+        assertNotNull(target._$PINPOINT$_getReactorContext());
+        assertEquals(target._$PINPOINT$_getReactorContext(), mockAsyncContext);
+    }
+
+    @Test
+    public void throwableIsNotNull() {
+        AsyncContext mockAsyncContext = mock(AsyncContext.class);
+        MockAsyncContextAndReactorContextImpl target = new MockAsyncContextAndReactorContextImpl();
+        MockAsyncContextAndReactorContextImpl arg0 = new MockAsyncContextAndReactorContextImpl();
+        Throwable throwable = new Throwable("ERROR");
+        FluxAndMonoOperatorConstructorInterceptor interceptor = new FluxAndMonoOperatorConstructorInterceptor();
+
+        interceptor.after(target, new Object[]{arg0}, new Object(), throwable);
+
+        assertNull(target._$PINPOINT$_getAsyncContext());
+        assertNull(target._$PINPOINT$_getReactorContext());
+    }
+}