<commit_before>package cc.osint.graphd.server;

import java.lang.*;
import java.lang.ref.*;
import java.util.*;
import java.util.concurrent.*;
import java.util.logging.Level;
import java.util.logging.Logger;
import org.json.*;
import org.jboss.netty.channel.Channel;
import cc.osint.graphd.graph.*;

public class GraphCommandExecutor implements Runnable {
    private static final Logger log = Logger.getLogger(
        GraphCommandExecutor.class.getName());

    private String graphName;
    private WeakReference<Graph> graphRef;
    LinkedBlockingQueue<GraphCommand> graphCommandQueue;
    
    public GraphCommandExecutor(String graphName,
                                WeakReference<Graph> graphRef) {
        this.graphName = graphName;
        this.graphRef = graphRef;
        graphCommandQueue = new LinkedBlockingQueue<GraphCommand>();
        log.info("start: GraphCommandExecutor(" + this.graphName + ")");
    }
    
    public void queue(GraphCommand cmd) throws Exception {
        graphCommandQueue.put(cmd);
    }
    
    public void run() {
        while (true) {
            GraphCommand graphCommand = null;
            String response;
            try {
                graphCommand = graphCommandQueue.take();
                if (graphCommand.poisonPill) {
                    log.info(graphName + ": detected poison pill: terminator processor");
                    return;
                }
                long requestTimeStart = System.currentTimeMillis();
                response = 
                    execute(graphCommand.responseChannel,
                            graphCommand.clientId,
                            graphCommand.clientState,
                            graphCommand.request,
                            graphCommand.cmd,
                            graphCommand.args);
                long requestTimeElapsed = System.currentTimeMillis() - requestTimeStart;
                log.info("[" + requestTimeElapsed + "ms]");
            } catch (Exception ex) {
                ex.printStackTrace();
                response = GraphServerProtocol.R_ERR + GraphServerProtocol.SPACE + ex.getMessage();
            }
            (graphCommand.responseChannel).write(response.trim() + GraphServerProtocol.NL);
        }
    }
    
    protected String execute(Channel responseChannel,
                             String clientId, 
                             ConcurrentHashMap<String, String> clientState, 
                             String request, 
                             String cmd, 
                             String[] args) throws Exception {
        
        StringBuffer rsb = new StringBuffer();
        
        // all operations require a db to be selected (via GraphServerProtocol.CMD_USE)
        
        if (null == clientState.get(GraphServerHandler.ST_DB)) {
            rsb.append(GraphServerProtocol.R_ERR);
            rsb.append(" REQUIRE_USE_DB");
            return rsb.toString();
        }
        
        Graph gr = graphRef.get();
        if (null == gr) {
            GraphCommand killCmd = new GraphCommand();
            killCmd.poisonPill = true;
            graphCommandQueue.clear();
            graphCommandQueue.put(killCmd);
            return GraphServerProtocol.R_ERR + GraphServerProtocol.SPACE + " GRAPH_NO_LONGER_EXISTS :queue cleared & poison pill sent";
        }
        
        /*
         * handle dynamic <<query>> expansion, replacement & execution, e.g.
         * 
         *   [...] <<_type:v _key:m*>> [...]
         *
         * TODO: take the responseChannel direct-send hack out of this and think about
         *  how to either queue the commands (& add an "echo-batch-ok" or equiv to the tail to
         *  signify end of the batched command execution) or move this up to the thread loop?
         *
        */
        if (request.indexOf("<<") != -1 &&
            request.indexOf(">>") != -1) {
            String query = request.substring(request.indexOf("<<")+2,
                                             request.indexOf(">>"));
            String prefix = request.substring(0, request.indexOf("<<")).trim();
            String suffix = request.substring(request.indexOf(">>")+2).trim();
            log.info("executing selector: [" + prefix + "] " + query + " [" + suffix + "]:");
            List<JSONObject> selectorResults = gr.queryGraphIndex(query);
            for(JSONObject selectorResult: selectorResults) {
                String batchRequestKey = selectorResult.getString(Graph.KEY_FIELD);
                String batchRequest = 
                    prefix + " " +
                    batchRequestKey + " " +
                    suffix;
                String batchCmd;
                String[] batchArgs;
                
                if (batchRequest.indexOf(GraphServerProtocol.SPACE) != -1) {
                    batchCmd = batchRequest.substring(0, 
                        batchRequest.indexOf(GraphServerProtocol.SPACE)).trim().toLowerCase();
                    batchArgs = batchRequest.substring(
                        batchRequest.indexOf(GraphServerProtocol.SPACE)).trim().split(GraphServerProtocol.SPACE);
                } else {
                    batchCmd = batchRequest.trim().toLowerCase();
                    batchArgs = new String[0];
                }
                    
                String batchCmdResponse =
                    execute(responseChannel,
                            clientId, 
                            clientState, 
                            batchRequest,
                            batchCmd,
                            batchArgs);
                responseChannel.write(batchCmdResponse.trim() + GraphServerProtocol.NL);
            }
            return GraphServerProtocol.R_BATCH_OK;
        }
            
        // CREATE VERTEX: cvert <key> <json>
        if (cmd.equals(GraphServerProtocol.CMD_CVERT)) {
            String key = args[0];
            String json = request.substring(request.indexOf(GraphServerProtocol.SPACE + key) +
                (key.length()+1)).trim(); // remainder of line
            JSONObject jo = null;
            try {
                jo = new JSONObject(json);
                gr.addVertex(key, jo);
                rsb.append(GraphServerProtocol.R_OK);
            } catch (org.json.JSONException jsonEx) {
                jsonEx.printStackTrace();
                rsb.append(GraphServerProtocol.R_ERR);
                rsb.append(" BAD_JSON ");
                rsb.append(jsonEx.getMessage());
            } catch (Exception ex) {
                ex.printStackTrace();
                rsb.append(GraphServerProtocol.R_ERR);
                rsb.append(GraphServerProtocol.SPACE);
                rsb.append(ex.getMessage());
            }
                                
        // CREATE EDGE: cedge <key> <vFromKey> <vToKey> <rel> <json>
        // OR:          cedge <key> <vFromKey> <vToKey> <rel> <weight> <json><commit_msg>changed response format of dynamic <<q...>> commands to only include positive responses followed by a single -batch-ok -- this means you won't see exceptions, errors, ok, etc. only -batch-ok!  this will have to be changed, but for now is practical and still useful if you are sure what you are doing<commit_after>package cc.osint.graphd.server;

import java.lang.*;
import java.lang.ref.*;
import java.util.*;
import java.util.concurrent.*;
import java.util.logging.Level;
import java.util.logging.Logger;
import org.json.*;
import org.jboss.netty.channel.Channel;
import cc.osint.graphd.graph.*;

public class GraphCommandExecutor implements Runnable {
    private static final Logger log = Logger.getLogger(
        GraphCommandExecutor.class.getName());

    private String graphName;
    private WeakReference<Graph> graphRef;
    LinkedBlockingQueue<GraphCommand> graphCommandQueue;
    
    public GraphCommandExecutor(String graphName,
                                WeakReference<Graph> graphRef) {
        this.graphName = graphName;
        this.graphRef = graphRef;
        graphCommandQueue = new LinkedBlockingQueue<GraphCommand>();
        log.info("start: GraphCommandExecutor(" + this.graphName + ")");
    }
    
    public void queue(GraphCommand cmd) throws Exception {
        graphCommandQueue.put(cmd);
    }
    
    public void run() {
        while (true) {
            GraphCommand graphCommand = null;
            String response;
            try {
                graphCommand = graphCommandQueue.take();
                if (graphCommand.poisonPill) {
                    log.info(graphName + ": detected poison pill: terminator processor");
                    return;
                }
                long requestTimeStart = System.currentTimeMillis();
                response = 
                    execute(graphCommand.responseChannel,
                            graphCommand.clientId,
                            graphCommand.clientState,
                            graphCommand.request,
                            graphCommand.cmd,
                            graphCommand.args);
                long requestTimeElapsed = System.currentTimeMillis() - requestTimeStart;
                log.info("[" + requestTimeElapsed + "ms]");
            } catch (Exception ex) {
                ex.printStackTrace();
                response = GraphServerProtocol.R_ERR + GraphServerProtocol.SPACE + ex.getMessage();
            }
            (graphCommand.responseChannel).write(response.trim() + GraphServerProtocol.NL);
        }
    }
    
    protected String execute(Channel responseChannel,
                             String clientId, 
                             ConcurrentHashMap<String, String> clientState, 
                             String request, 
                             String cmd, 
                             String[] args) throws Exception {
        
        StringBuffer rsb = new StringBuffer();
        
        // all operations require a db to be selected (via GraphServerProtocol.CMD_USE)
        
        if (null == clientState.get(GraphServerHandler.ST_DB)) {
            rsb.append(GraphServerProtocol.R_ERR);
            rsb.append(" REQUIRE_USE_DB");
            return rsb.toString();
        }
        
        Graph gr = graphRef.get();
        if (null == gr) {
            GraphCommand killCmd = new GraphCommand();
            killCmd.poisonPill = true;
            graphCommandQueue.clear();
            graphCommandQueue.put(killCmd);
            return GraphServerProtocol.R_ERR + GraphServerProtocol.SPACE + " GRAPH_NO_LONGER_EXISTS :queue cleared & poison pill sent";
        }
        
        /*
         * handle dynamic <<query>> expansion, replacement & execution, e.g.
         * 
         *   [...] <<_type:v _key:m*>> [...]
         *
         * TODO: take the responseChannel direct-send hack out of this and think about
         *  how to either queue the commands (& add an "echo-batch-ok" or equiv to the tail to
         *  signify end of the batched command execution) or move this up to the thread loop?
         *
        */
        if (request.indexOf("<<") != -1 &&
            request.indexOf(">>") != -1) {
            String query = request.substring(request.indexOf("<<")+2,
                                             request.indexOf(">>"));
            String prefix = request.substring(0, request.indexOf("<<")).trim();
            String suffix = request.substring(request.indexOf(">>")+2).trim();
            log.info("executing selector: [" + prefix + "] " + query + " [" + suffix + "]:");
            List<JSONObject> selectorResults = gr.queryGraphIndex(query);
            for(JSONObject selectorResult: selectorResults) {
                String batchRequestKey = selectorResult.getString(Graph.KEY_FIELD);
                String batchRequest = 
                    prefix + " " +
                    batchRequestKey + " " +
                    suffix;
                String batchCmd;
                String[] batchArgs;
                
                if (batchRequest.indexOf(GraphServerProtocol.SPACE) != -1) {
                    batchCmd = batchRequest.substring(0, 
                        batchRequest.indexOf(GraphServerProtocol.SPACE)).trim().toLowerCase();
                    batchArgs = batchRequest.substring(
                        batchRequest.indexOf(GraphServerProtocol.SPACE)).trim().split(GraphServerProtocol.SPACE);
                } else {
                    batchCmd = batchRequest.trim().toLowerCase();
                    batchArgs = new String[0];
                }
                    
                String batchCmdResponse =
                    execute(responseChannel,
                            clientId, 
                            clientState, 
                            batchRequest,
                            batchCmd,
                            batchArgs);
                String responseParts[] = batchCmdResponse.split(GraphServerProtocol.NL);
                for(String responsePart: responseParts) {
                    if (responsePart.charAt(0) != '-') {
                        responseChannel.write(responsePart.trim() + GraphServerProtocol.NL);
                    }
                }
            }
            return GraphServerProtocol.R_BATCH_OK;
        }
            
        // CREATE VERTEX: cvert <key> <json>
        if (cmd.equals(GraphServerProtocol.CMD_CVERT)) {
            String key = args[0];
            String json = request.substring(request.indexOf(GraphServerProtocol.SPACE + key) +
                (key.length()+1)).trim(); // remainder of line
            JSONObject jo = null;
            try {
                jo = new JSONObject(json);
                gr.addVertex(key, jo);
                rsb.append(GraphServerProtocol.R_OK);
            } catch (org.json.JSONException jsonEx) {
                jsonEx.printStackTrace();
                rsb.append(GraphServerProtocol.R_ERR);
                rsb.append(" BAD_JSON ");
                rsb.append(jsonEx.getMessage());
            } catch (Exception ex) {
                ex.printStackTrace();
                rsb.append(GraphServerProtocol.R_ERR);
                rsb.append(GraphServerProtocol.SPACE);
                rsb.append(ex.getMessage());
            }
                                
        // CREATE EDGE: cedge <key> <vFromKey> <vToKey> <rel> <json>
        // OR:          cedge <key> <vFromKey> <vToKey> <rel> <weight> <json>
{'fraction_non_alphanumeric': 0.09824148435336472, 'fraction_numerical': 0.0014538908889504293, 'mean_word_length': 1.5744074140081983, 'pattern_counts': {'":': 2, '<': 63, '<?xml version=': 0, '>': 63, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3627259', 'n_tokens_mistral': 3515, 'n_tokens_neox': 3420, 'n_words': 754}