feat: use an enhanced CypherAPI to refactor it (apache#2143)

Co-authored-by: lynn.bond <liyan75@baidu.com>
Co-authored-by: imbajin <jin@apache.org>

.licenserc.yaml

@@ -94,6 +94,8 @@ header: # `header` section is configurations for source codes license header.
     - '**/type/Nameable.java'
     - '**/define/Cardinality.java'
     - '**/util/StringEncoding.java'
+    - 'hugegraph-api/src/main/java/org/apache/hugegraph/opencypher/CypherOpProcessor.java'
+    - 'hugegraph-api/src/main/java/org/apache/hugegraph/opencypher/CypherPlugin.java'
   comment: on-failure # on what condition license-eye will comment on the pull request, `on-failure`, `always`, `never`.
 
   # license-location-threshold specifies the index threshold where the license header can be located,

LICENSE

@@ -214,3 +214,5 @@ hugegraph-core/src/main/java/org/apache/hugegraph/traversal/optimize/HugeScriptT
 hugegraph-core/src/main/java/org/apache/hugegraph/type/Nameable.java from https://github.com/JanusGraph/janusgraph
 hugegraph-core/src/main/java/org/apache/hugegraph/type/define/Cardinality.java from https://github.com/JanusGraph/janusgraph
 hugegraph-core/src/main/java/org/apache/hugegraph/util/StringEncoding.java from https://github.com/JanusGraph/janusgraph
+hugegraph-api/src/main/java/org/apache/hugegraph/opencypher/CypherOpProcessor.java from https://github.com/opencypher/cypher-for-gremlin
+hugegraph-api/src/main/java/org/apache/hugegraph/opencypher/CypherPlugin.java from https://github.com/opencypher/cypher-for-gremlin

hugegraph-api/src/main/java/org/apache/hugegraph/api/cypher/CypherAPI.java

@@ -0,0 +1,154 @@
+/*
+ * 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.hugegraph.api.cypher;
+
+import java.nio.charset.Charset;
+import java.nio.charset.StandardCharsets;
+import java.util.Base64;
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.commons.lang3.tuple.ImmutablePair;
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.hugegraph.api.API;
+import org.apache.hugegraph.api.filter.CompressInterceptor;
+import org.apache.hugegraph.util.E;
+import org.apache.hugegraph.util.Log;
+import org.slf4j.Logger;
+
+import com.codahale.metrics.annotation.Timed;
+
+import jakarta.inject.Singleton;
+import jakarta.ws.rs.Consumes;
+import jakarta.ws.rs.GET;
+import jakarta.ws.rs.NotAuthorizedException;
+import jakarta.ws.rs.POST;
+import jakarta.ws.rs.Path;
+import jakarta.ws.rs.PathParam;
+import jakarta.ws.rs.Produces;
+import jakarta.ws.rs.QueryParam;
+import jakarta.ws.rs.core.Context;
+import jakarta.ws.rs.core.HttpHeaders;
+
+@Path("graphs/{graph}/cypher")
+@Singleton
+public class CypherAPI extends API {
+
+    private static final Logger LOG = Log.logger(CypherAPI.class);
+    private static final Charset UTF8 = StandardCharsets.UTF_8;
+    private static final String CLIENT_CONF = "conf/remote-objects.yaml";
+    private final Base64.Decoder decoder = Base64.getUrlDecoder();
+    private final String basic = "Basic ";
+    private final String bearer = "Bearer ";
+
+    private CypherManager cypherManager;
+
+    private CypherManager cypherManager() {
+        if (this.cypherManager == null) {
+            this.cypherManager = CypherManager.configOf(CLIENT_CONF);
+        }
+        return this.cypherManager;
+    }
+
+    @GET
+    @Timed
+    @CompressInterceptor.Compress(buffer = (1024 * 40))
+    @Produces(APPLICATION_JSON_WITH_CHARSET)
+    public CypherModel query(@PathParam("graph") String graph, @Context HttpHeaders headers,
+                             @QueryParam("cypher") String cypher) {
+        LOG.debug("Graph [{}] query by cypher: {}", graph, cypher);
+        return this.queryByCypher(graph, headers, cypher);
+    }
+
+    @POST
+    @Timed
+    @CompressInterceptor.Compress
+    @Consumes(APPLICATION_JSON)
+    @Produces(APPLICATION_JSON_WITH_CHARSET)
+    public CypherModel post(@PathParam("graph") String graph,
+                            @Context HttpHeaders headers, String cypher) {
+        LOG.debug("Graph [{}] query by cypher: {}", graph, cypher);
+        return this.queryByCypher(graph, headers, cypher);
+    }
+
+    private CypherModel queryByCypher(String graph, HttpHeaders headers, String cypher) {
+        E.checkArgument(graph != null && !graph.isEmpty(),
+                        "The graph parameter can't be null or empty");
+        E.checkArgument(cypher != null && !cypher.isEmpty(),
+                        "The cypher parameter can't be null or empty");
+
+        Map<String, String> aliases = new HashMap<>(1, 1);
+        aliases.put("g", "__g_" + graph);
+
+        return this.client(headers).submitQuery(cypher, aliases);
+    }
+
+    private CypherClient client(HttpHeaders headers) {
+        String auth = headers.getHeaderString(HttpHeaders.AUTHORIZATION);
+
+        if (auth != null && !auth.isEmpty()) {
+            auth = auth.split(",")[0];
+        }
+
+        if (auth != null) {
+            if (auth.startsWith(basic)) {
+                return this.clientViaBasic(auth);
+            } else if (auth.startsWith(bearer)) {
+                return this.clientViaToken(auth);
+            }
+        }
+
+        throw new NotAuthorizedException("The Cypher-API called without any authorization.");
+    }
+
+    private CypherClient clientViaBasic(String auth) {
+        Pair<String, String> userPass = this.toUserPass(auth);
+        E.checkNotNull(userPass, "user-password-pair");
+
+        return this.cypherManager().getClient(userPass.getLeft(), userPass.getRight());
+    }
+
+    private CypherClient clientViaToken(String auth) {
+        return this.cypherManager().getClient(auth.substring(bearer.length()));
+    }
+
+    private Pair<String, String> toUserPass(String auth) {
+        if (auth == null || auth.isEmpty()) {
+            return null;
+        }
+        if (!auth.startsWith(basic)) {
+            return null;
+        }
+
+        String[] split;
+        try {
+            String encoded = auth.substring(basic.length());
+            byte[] userPass = this.decoder.decode(encoded);
+            String authorization = new String(userPass, UTF8);
+            split = authorization.split(":");
+        } catch (Exception e) {
+            LOG.error("Failed convert auth to credential.", e);
+            return null;
+        }
+
+        if (split.length != 2) {
+            return null;
+        }
+        return ImmutablePair.of(split[0], split[1]);
+    }
+}

hugegraph-api/src/main/java/org/apache/hugegraph/api/cypher/CypherClient.java

@@ -0,0 +1,153 @@
+/*
+ * 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.hugegraph.api.cypher;
+
+import org.apache.hugegraph.util.E;
+import org.apache.hugegraph.util.Log;
+import org.apache.commons.configuration2.Configuration;
+import org.apache.tinkerpop.gremlin.driver.*;
+import org.apache.tinkerpop.gremlin.driver.message.RequestMessage;
+import org.slf4j.Logger;
+
+import javax.annotation.Nullable;
+import javax.annotation.concurrent.ThreadSafe;
+
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.ExecutionException;
+import java.util.function.Supplier;
+
+@ThreadSafe
+public final class CypherClient {
+
+    private static final Logger LOG = Log.logger(CypherClient.class);
+    private final Supplier<Configuration> configurationSupplier;
+    private String userName;
+    private String password;
+    private String token;
+
+    CypherClient(String userName, String password,
+                 Supplier<Configuration> configurationSupplier) {
+        this.userName = userName;
+        this.password = password;
+        this.configurationSupplier = configurationSupplier;
+    }
+
+    CypherClient(String token, Supplier<Configuration> configurationSupplier) {
+        this.token = token;
+        this.configurationSupplier = configurationSupplier;
+    }
+
+    public CypherModel submitQuery(String cypherQuery, @Nullable Map<String, String> aliases) {
+        E.checkArgument(cypherQuery != null && !cypherQuery.isEmpty(),
+                        "The cypher-query parameter can't be null or empty");
+
+        Cluster cluster = Cluster.open(getConfig());
+        Client client = cluster.connect();
+
+        if (aliases != null && !aliases.isEmpty()) {
+            client = client.alias(aliases);
+        }
+
+        RequestMessage request = createRequest(cypherQuery);
+        CypherModel res;
+
+        try {
+            List<Object> list = this.doQueryList(client, request);
+            res = CypherModel.dataOf(request.getRequestId().toString(), list);
+        } catch (Exception e) {
+            LOG.error(String.format("Failed to submit cypher-query: [ %s ], caused by:",
+                                    cypherQuery), e);
+            res = CypherModel.failOf(request.getRequestId().toString(), e.getMessage());
+        } finally {
+            client.close();
+            cluster.close();
+        }
+
+        return res;
+    }
+
+    private RequestMessage createRequest(String cypherQuery) {
+        return RequestMessage.build(Tokens.OPS_EVAL)
+                             .processor("cypher")
+                             .add(Tokens.ARGS_GREMLIN, cypherQuery)
+                             .create();
+    }
+
+    private List<Object> doQueryList(Client client, RequestMessage request)
+        throws ExecutionException, InterruptedException {
+        ResultSet results = client.submitAsync(request).get();
+
+        Iterator<Result> iter = results.iterator();
+        List<Object> list = new LinkedList<>();
+
+        while (iter.hasNext()) {
+            Result data = iter.next();
+            list.add(data.getObject());
+        }
+
+        return list;
+    }
+
+    /**
+     * As Sasl does not support a token, which is a coded string to indicate a legal user,
+     * we had to use a trick to fix it. When the token is set, the password will be set to
+     * an empty string, which is an uncommon value under normal conditions.
+     * The token will then be transferred through the userName-property.
+     * To see org.apache.hugegraph.auth.StandardAuthenticator.TokenSaslAuthenticator
+     */
+    private Configuration getConfig() {
+        Configuration conf = this.configurationSupplier.get();
+        conf.addProperty("username", this.token == null ? this.userName : this.token);
+        conf.addProperty("password", this.token == null ? this.password : "");
+
+        return conf;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+        CypherClient that = (CypherClient) o;
+
+        return Objects.equals(userName, that.userName) &&
+               Objects.equals(password, that.password) &&
+               Objects.equals(token, that.token);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(userName, password, token);
+    }
+
+    @Override
+    public String toString() {
+        final StringBuilder builder = new StringBuilder("CypherClient{");
+        builder.append("userName='").append(userName).append('\'')
+               .append(", token='").append(token).append('\'').append('}');
+
+        return builder.toString();
+    }
+}