Less magic numbers and more readability in IPFSCompat.

apache · Aug 13, 2020 · 39bab37 · 39bab37
1 parent d7ee62c
commit 39bab37
Showing 1 changed file with 57 additions and 20 deletions.
diff --git a/contrib/storage-ipfs/src/main/java/org/apache/drill/exec/store/ipfs/IPFSCompat.java b/contrib/storage-ipfs/src/main/java/org/apache/drill/exec/store/ipfs/IPFSCompat.java
@@ -101,6 +101,28 @@ public Map resolve(String scheme, String path, boolean recursive) {
     return ret.get();
   }
 
+  /**
+   * As defined in https://github.com/libp2p/go-libp2p-core/blob/b77fd280f2bfcce22f10a000e8e1d9ec53c47049/routing/query.go#L16
+   */
+  public enum DHTQueryEventType {
+    // Sending a query to a peer.
+    SendingQuery,
+    // Got a response from a peer.
+    PeerResponse,
+    // Found a "closest" peer (not currently used).
+    FinalPeer,
+    // Got an error when querying.
+    QueryError,
+    // Found a provider.
+    Provider,
+    // Found a value.
+    Value,
+    // Adding a peer to the query.
+    AddingPeer,
+    // Dialing a peer.
+    DialingPeer;
+  }
+
   public class DHT {
     /**
      * Find internet addresses of a given peer.
@@ -117,17 +139,25 @@ public List<String> findpeerListTimeout(Multihash id, int timeout, ExecutorServi
           timeout,
           res -> {
             Map peer = (Map) res;
-            if (peer != null) {
-              if ( (int) peer.get("Type") == 2 ) {
-                ret.set((List<String>)
-                    ((Map)
-                        ((List) peer.get("Responses")
-                        ).get(0)
-                    ).get("Addrs"));
-                return true;
-              }
+            if (peer == null) {
+              return false;
+            }
+            if ( (int) peer.get("Type") != DHTQueryEventType.FinalPeer.ordinal() ) {
+              return false;
+            }
+            List<Map> responses = (List<Map>) peer.get("Responses");
+            if (responses == null || responses.size() == 0) {
+              return false;
+            }
+            // FinalPeer responses have exactly one response
+            Map<String, List<String>> response = responses.get(0);
+            if (response == null) {
+              return false;
             }
-            return false;
+            List<String> addrs = response.get("Addrs");
+
+            ret.set(addrs);
+            return true;
           },
           err -> {
             if (!(err instanceof TimeoutException)) {
@@ -158,17 +188,24 @@ public List<String> findprovsListTimeout(Multihash id, int maxPeers, int timeout
           timeout,
           res -> {
             Map peer = (Map) res;
-            if ( peer != null ) {
-              if ( (int) peer.get("Type") == 4 ) {
-                ret.get().add(
-                    (String)
-                        ((Map)
-                            ((List) peer.get("Responses")
-                            ).get(0)
-                        ).get("ID")
-                );
-              }
+            if ( peer == null ) {
+              return false;
+            }
+            if ( (int) peer.get("Type") != DHTQueryEventType.Provider.ordinal() ) {
+              return false;
             }
+            List<Map> responses = (List<Map>) peer.get("Responses");
+            if (responses == null || responses.size() == 0) {
+              return false;
+            }
+            // One Provider message contains only one provider
+            Map<String, String> response = responses.get(0);
+            if (response == null) {
+              return false;
+            }
+            String providerID = response.get("ID");
+
+            ret.get().add(providerID);
             return ret.get().size() >= maxPeers;
           },
           err -> {