[pinpoint-apm#9932] Reduce thrift dependency

Author: emeroad

profiler-test/src/main/java/com/navercorp/pinpoint/test/TestTcpDataSender.java

@@ -20,9 +20,7 @@
 import com.navercorp.pinpoint.profiler.metadata.SqlMetaData;
 import com.navercorp.pinpoint.profiler.metadata.StringMetaData;
 import com.navercorp.pinpoint.profiler.sender.EnhancedDataSender;
-import com.navercorp.pinpoint.rpc.FutureListener;
 import com.navercorp.pinpoint.rpc.ResponseMessage;
-import com.navercorp.pinpoint.rpc.client.PinpointClientReconnectEventListener;
 import com.navercorp.pinpoint.test.util.BiHashMap;
 import com.navercorp.pinpoint.test.util.Pair;
 
@@ -33,6 +31,8 @@
 import java.util.List;
 import java.util.Map.Entry;
 import java.util.NoSuchElementException;
+import java.util.concurrent.CompletableFuture;
+import java.util.function.Consumer;
 
 /**
  * @author Jongho Moon
@@ -140,20 +140,11 @@ public boolean request(MetaDataType data, int retry) {
     }
 
     @Override
-    public boolean request(MetaDataType data, FutureListener<ResponseMessage> listener) {
+    public boolean request(MetaDataType data, Consumer<CompletableFuture<ResponseMessage>> listener) {
         addData(data);
         return true;
     }
 
-    @Override
-    public boolean addReconnectEventListener(PinpointClientReconnectEventListener eventListener) {
-        return false;
-    }
-
-    @Override
-    public boolean removeReconnectEventListener(PinpointClientReconnectEventListener eventListener) {
-        return false;
-    }
 
     public String getApiDescription(int id) {
         return syncGet(apiIdMap, id);

profiler/src/main/java/com/navercorp/pinpoint/profiler/AgentInfoSender.java

@@ -16,22 +16,21 @@
 
 package com.navercorp.pinpoint.profiler;
 
-import java.util.Timer;
-import java.util.TimerTask;
-import java.util.concurrent.atomic.AtomicInteger;
-
-import java.util.Objects;
 import com.navercorp.pinpoint.profiler.context.thrift.MessageConverter;
 import com.navercorp.pinpoint.profiler.metadata.AgentInfo;
 import com.navercorp.pinpoint.profiler.metadata.MetaDataType;
+import com.navercorp.pinpoint.profiler.sender.EnhancedDataSender;
 import com.navercorp.pinpoint.profiler.sender.ResultResponse;
 import com.navercorp.pinpoint.profiler.util.AgentInfoFactory;
-import com.navercorp.pinpoint.rpc.DefaultFuture;
 import com.navercorp.pinpoint.rpc.ResponseMessage;
-import org.apache.logging.log4j.Logger;
 import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
 
-import com.navercorp.pinpoint.profiler.sender.EnhancedDataSender;
+import java.util.Objects;
+import java.util.Timer;
+import java.util.TimerTask;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
 
 /**
  * @author emeroad
@@ -175,31 +174,22 @@ public void run() {
         private boolean sendAgentInfo() {
             try {
                 AgentInfo agentInfo = agentInfoFactory.createAgentInfo();
-                final DefaultFuture<ResponseMessage> future = new DefaultFuture<>();
 
                 logger.info("Sending AgentInfo {}", agentInfo);
-                dataSender.request(agentInfo, new ResponseMessageFutureListener(future));
-                if (!future.await()) {
-                    logger.warn("request timed out while waiting for response.");
-                    return false;
-                }
-                if (!future.isSuccess()) {
-                    Throwable t = future.getCause();
-                    logger.warn("request failed.", t);
-                    return false;
-                }
-                ResponseMessage responseMessage = future.getResult();
+                ResponseFutureListener<ResponseMessage> listener = new ResponseFutureListener<>();
+                dataSender.request(agentInfo, listener);
+                ResponseMessage responseMessage = listener.getResponseFeture().get(3000, TimeUnit.MILLISECONDS);
                 if (responseMessage == null) {
-                    logger.warn("result not set.");
+                    logger.warn("result not set");
                     return false;
                 }
                 final ResultResponse result = messageConverter.toMessage(responseMessage);
                 if (!result.isSuccess()) {
                     logger.warn("request unsuccessful. Cause : {}", result.getMessage());
                 }
                 return result.isSuccess();
-            } catch (Exception e) {
-                logger.warn("failed to send agent info.", e);
+            } catch (Throwable th) {
+                logger.warn("failed to send agent info", th);
             }
             return false;
         }

profiler/src/main/java/com/navercorp/pinpoint/profiler/ResponseFutureListener.java

@@ -0,0 +1,49 @@
+/*
+ * Copyright 2014 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;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.function.Consumer;
+
+public class ResponseFutureListener<T> implements Consumer<CompletableFuture<T>> {
+
+    private final CompletableFuture<T> future;
+
+    public ResponseFutureListener() {
+        this.future = new CompletableFuture<>();
+    }
+
+    @Override
+    public void accept(CompletableFuture<T> cf) {
+        if (!cf.isDone()) {
+            this.future.completeExceptionally(new IllegalStateException("ResponseMessage future is not complete"));
+            return;
+        }
+
+        cf.whenComplete((responseMessage, throwable) -> {
+            if (throwable != null) {
+                future.completeExceptionally(throwable);
+            } else {
+                future.complete(responseMessage);
+            }
+        });
+    }
+
+    public CompletableFuture<T> getResponseFeture() {
+        return future;
+    }
+}

profiler/src/main/java/com/navercorp/pinpoint/profiler/ResponseMessageFutureListener.java

@@ -16,9 +16,10 @@
 
 package com.navercorp.pinpoint.profiler.sender;
 
-import com.navercorp.pinpoint.rpc.FutureListener;
 import com.navercorp.pinpoint.rpc.ResponseMessage;
-import com.navercorp.pinpoint.rpc.client.PinpointClientReconnectEventListener;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.function.Consumer;
 
 
 /**
@@ -55,17 +56,7 @@ public boolean request(T data, int retry) {
 
 
     @Override
-    public boolean request(T data, FutureListener<ResponseMessage> listener) {
-        return false;
-    }
-
-    @Override
-    public boolean addReconnectEventListener(PinpointClientReconnectEventListener eventListener) {
-        return false;
-    }
-
-    @Override
-    public boolean removeReconnectEventListener(PinpointClientReconnectEventListener eventListener) {
+    public boolean request(T data, Consumer<CompletableFuture<ResponseMessage>> listener) {
         return false;
     }
 

profiler/src/main/java/com/navercorp/pinpoint/profiler/sender/EmptyDataSender.java

@@ -16,9 +16,10 @@
 
 package com.navercorp.pinpoint.profiler.sender;
 
-import com.navercorp.pinpoint.rpc.FutureListener;
 import com.navercorp.pinpoint.rpc.ResponseMessage;
-import com.navercorp.pinpoint.rpc.client.PinpointClientReconnectEventListener;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.function.Consumer;
 
 
 /**
@@ -27,10 +28,10 @@
 public interface EnhancedDataSender<T> extends DataSender<T> {
 
     boolean request(T data);
+
     boolean request(T data, int retry);
-    boolean request(T data, FutureListener<ResponseMessage> listener);
 
-    boolean addReconnectEventListener(PinpointClientReconnectEventListener eventListener);
-    boolean removeReconnectEventListener(PinpointClientReconnectEventListener eventListener);
+    boolean request(T data, Consumer<CompletableFuture<ResponseMessage>> listener);
+
 
 }

profiler/src/main/java/com/navercorp/pinpoint/profiler/sender/EnhancedDataSender.java

@@ -16,18 +16,20 @@
 
 package com.navercorp.pinpoint.profiler.sender;
 
-import com.navercorp.pinpoint.rpc.FutureListener;
 import com.navercorp.pinpoint.rpc.ResponseMessage;
 
+import java.util.concurrent.CompletableFuture;
+import java.util.function.Consumer;
+
 /**
  * @author Woonduk Kang(emeroad)
  */
 class ListenerableRequestMessage<T> implements RequestMessage<T> {
     private final T message;
-    private final FutureListener<ResponseMessage> futureListener;
+    private final Consumer<CompletableFuture<ResponseMessage>> futureListener;
 
 
-    ListenerableRequestMessage(T message, FutureListener<ResponseMessage> futureListener) {
+    ListenerableRequestMessage(T message, Consumer<CompletableFuture<ResponseMessage>> futureListener) {
         this.message = message;
         this.futureListener = futureListener;
     }
@@ -44,7 +46,7 @@ public int getRetryCount() {
     }
 
     @Override
-    public FutureListener<ResponseMessage> getFutureListener() {
+    public Consumer<CompletableFuture<ResponseMessage>> getFutureListener() {
         return futureListener;
     }
 }

profiler/src/main/java/com/navercorp/pinpoint/profiler/sender/ListenerableRequestMessage.java

@@ -16,12 +16,12 @@
 
 package com.navercorp.pinpoint.profiler.sender;
 
-import com.navercorp.pinpoint.rpc.FutureListener;
 import com.navercorp.pinpoint.rpc.ResponseMessage;
-import com.navercorp.pinpoint.rpc.client.PinpointClientReconnectEventListener;
-
-import org.apache.logging.log4j.Logger;
 import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.function.Consumer;
 
 
 /**
@@ -58,21 +58,9 @@ public boolean request(T data, int retry) {
 
 
     @Override
-    public boolean request(T data, FutureListener<ResponseMessage> listener) {
+    public boolean request(T data, Consumer<CompletableFuture<ResponseMessage>> listener) {
         logger.info("request tBase:{} FutureListener:{}", data, listener);
         return false;
     }
 
-    @Override
-    public boolean addReconnectEventListener(PinpointClientReconnectEventListener eventListener) {
-        logger.info("addReconnectEventListener eventListener:{}", eventListener);
-        return false;
-    }
-
-    @Override
-    public boolean removeReconnectEventListener(PinpointClientReconnectEventListener eventListener) {
-        logger.info("removeReconnectEventListener eventListener:{}", eventListener);
-        return false;
-    }
-
 }

profiler/src/main/java/com/navercorp/pinpoint/profiler/sender/LoggingDataSender.java

@@ -0,0 +1,8 @@
+package com.navercorp.pinpoint.profiler.sender;
+
+import java.util.function.Consumer;
+
+public interface ReconnectEventListenerRegistry<T> {
+    boolean addEventListener(Consumer<T> eventListener);
+
+}

profiler/src/main/java/com/navercorp/pinpoint/profiler/sender/ReconnectEventListenerRegistry.java

@@ -16,9 +16,11 @@
 
 package com.navercorp.pinpoint.profiler.sender;
 
-import com.navercorp.pinpoint.rpc.FutureListener;
 import com.navercorp.pinpoint.rpc.ResponseMessage;
 
+import java.util.concurrent.CompletableFuture;
+import java.util.function.Consumer;
+
 /**
  * @author Woonduk Kang(emeroad)
  */
@@ -28,6 +30,6 @@ public interface RequestMessage<M> {
 
     int getRetryCount();
 
-    FutureListener<ResponseMessage> getFutureListener();
+    Consumer<CompletableFuture<ResponseMessage>> getFutureListener();
 
 }

profiler/src/main/java/com/navercorp/pinpoint/profiler/sender/RequestMessage.java

@@ -17,9 +17,11 @@
 package com.navercorp.pinpoint.profiler.sender;
 
 
-import com.navercorp.pinpoint.rpc.FutureListener;
 import com.navercorp.pinpoint.rpc.ResponseMessage;
 
+import java.util.concurrent.CompletableFuture;
+import java.util.function.Consumer;
+
 /**
  * @author Woonduk Kang(emeroad)
  */
@@ -32,7 +34,7 @@ public static <T> RequestMessage<T> request(T message, int retryCount) {
         return new RetryRequestMessage<>(message, retryCount);
     }
 
-    public static <T> RequestMessage<T> request(T message, FutureListener<ResponseMessage> futureListener) {
+    public static <T> RequestMessage<T> request(T message, Consumer<CompletableFuture<ResponseMessage>> futureListener) {
         return new ListenerableRequestMessage<>(message, futureListener);
     }