src/eric7/Plugins/CheckerPlugins/SyntaxChecker/jsCheckSyntax.py

# -*- coding: utf-8 -*-

# Copyright (c) 2014 - 2022 Detlev Offenbach <detlev@die-offenbachs.de>
#

"""
Module implementing the syntax check for JavaScript.
"""

import multiprocessing
import os
import queue
import sys


def initService():
    """
    Initialize the service and return the entry point.

    @return the entry point for the background client (function)
    """
    path = __file__
    for _ in range(4):
        path = os.path.dirname(path)
    sys.path.insert(2, os.path.join(path, "ThirdParty", "Jasy"))
    return jsSyntaxCheck


def initBatchService():
    """
    Initialize the batch service and return the entry point.

    @return the entry point for the background client (function)
    """
    return jsSyntaxBatchCheck


def jsSyntaxCheck(file, codestring):
    """
    Function to check a Javascript source file for syntax errors.

    @param file source filename (string)
    @param codestring string containing the code to check (string)
    @return dictionary with the keys 'error' and 'warnings' which
            hold a list containing details about the error/ warnings
            (file name, line number, column, codestring (only at syntax
            errors), the message, a list with arguments for the message)
    """
    return __jsSyntaxCheck(file, codestring)


def jsSyntaxBatchCheck(argumentsList, send, fx, cancelled, maxProcesses=0):
    """
    Module function to check syntax for a batch of files.

    @param argumentsList list of arguments tuples as given for jsSyntaxCheck
    @type list
    @param send reference to send function
    @type func
    @param fx registered service name
    @type str
    @param cancelled reference to function checking for a cancellation
    @type func
    @param maxProcesses number of processes to be used
    @type int
    """
    if maxProcesses == 0:
        # determine based on CPU count
        try:
            NumberOfProcesses = multiprocessing.cpu_count()
            if NumberOfProcesses >= 1:
                NumberOfProcesses -= 1
        except NotImplementedError:
            NumberOfProcesses = 1
    else:
        NumberOfProcesses = maxProcesses

    # Create queues
    taskQueue = multiprocessing.Queue()
    doneQueue = multiprocessing.Queue()

    # Submit tasks (initially two times the number of processes)
    tasks = len(argumentsList)
    initialTasks = min(2 * NumberOfProcesses, tasks)
    for _ in range(initialTasks):
        taskQueue.put(argumentsList.pop(0))

    # Start worker processes
    workers = [
        multiprocessing.Process(target=workerTask, args=(taskQueue, doneQueue))
        for _ in range(NumberOfProcesses)
    ]
    for worker in workers:
        worker.start()

    # Get and send results
    for _ in range(tasks):
        resultSent = False
        wasCancelled = False

        while not resultSent:
            try:
                # get result (waiting max. 3 seconds and send it to frontend
                filename, result = doneQueue.get()
                send(fx, filename, result)
                resultSent = True
            except queue.Empty:
                # ignore empty queue, just carry on
                if cancelled():
                    wasCancelled = True
                    break

        if wasCancelled or cancelled():
            # just exit the loop ignoring the results of queued tasks
            break

        if argumentsList:
            taskQueue.put(argumentsList.pop(0))

    # Tell child processes to stop
    for _ in range(NumberOfProcesses):
        taskQueue.put("STOP")

    for worker in workers:
        worker.join()
        worker.close()

    taskQueue.close()
    doneQueue.close()


def workerTask(inputQueue, outputQueue):
    """
    Module function acting as the parallel worker for the syntax check.

    @param inputQueue input queue (multiprocessing.Queue)
    @param outputQueue output queue (multiprocessing.Queue)
    """
    for filename, args in iter(inputQueue.get, "STOP"):
        source = args[0]
        result = __jsSyntaxCheck(filename, source)
        outputQueue.put((filename, result))


def __jsSyntaxCheck(file, codestring):
    """
    Function to check a Javascript source file for syntax errors.

    @param file source filename (string)
    @param codestring string containing the code to check (string)
    @return dictionary with the keys 'error' and 'warnings' which
            hold a list containing details about the error/ warnings
            (file name, line number, column, codestring (only at syntax
            errors), the message, a list with arguments for the message)
    """
    import jasy.script.parse.Parser as jsParser  # __IGNORE_WARNING_I102__
    import jasy.script.tokenize.Tokenizer as jsTokenizer  # __IGNORE_WARNING_I102__

    try:
        jsParser.parse(codestring, file)
    except (jsParser.SyntaxError, jsTokenizer.ParseError) as exc:
        details = exc.args[0]
        error, details = details.splitlines()
        fn, line = details.strip().rsplit(":", 1)
        error = error.split(":", 1)[1].strip()

        cline = min(len(codestring.splitlines()), int(line)) - 1
        code = codestring.splitlines()[cline]
        return [{"error": (fn, int(line), 0, code, error)}]
    except IndexError:
        error = "Incomplete source file"
        splittedCode = codestring.splitlines()
        return [
            {
                "error": (
                    file,
                    len(splittedCode) + 1,
                    len(splittedCode[-1]),
                    splittedCode[-1],
                    error,
                )
            }
        ]

    return [{}]