diff --git a/.gitignore b/.gitignore index 3f0a142..015dacc 100644 --- a/.gitignore +++ b/.gitignore @@ -1,3 +1,4 @@ + # Compiled source # ################### *.com diff --git a/app.js b/app.js index 002f017..fd9bb88 100644 --- a/app.js +++ b/app.js @@ -1,126 +1,119 @@ const tar = require('tar-stream'); const fs = require('fs'); const commandLineArgs = require('command-line-args'); - - -/** - * Command line arguments definition - * @type {[*]} - */ -const commandLineDef = [ - { name: 'port', alias: 'p', type: Number, defaultValue: 5016}, - { name: 'mongoAddress', alias: 'a', type: String, defaultValue: '127.0.0.1/queue'}, - { name: 'mongoCollection', alias: 'c', type: String, defaultValue: 'agendaJobs'}, - { name: 'defaultConcurrency', alias: 'd', type: Number, defaultValue: 40}, - { name: 'maxConcurrency', alias: 'm', type: Number, defaultValue: 70} -]; -/** - * Object that for keys has command line argument names - * and for values its value. - * @type {Object} - */ -const commandLine = commandLineArgs(commandLineDef); - - -/** - * Server specified in server.js - * @type {Server} - */ -const server = require('./server')( - commandLine.port, - commandLine.mongoAddress, - commandLine.mongoCollection, - commandLine.defaultConcurrency, - commandLine.maxConcurrency); -/** - * Object controlling docker specified in javaBox.js - * @type {EventEmitter} - */ -const javaBox = require('./javaBox'); - -/** - * Specifying the server and javaBox control flow - */ -server.on('runJunit', runJunit); -server.on('runJava', runJava); -javaBox.on('result', giveFeedBack); - - -/** - * Given code to execute and info about it, - * creates a tar with the code and passed the info and the tar to javaBox. - * - * @param messageId {String}: id of the given message/request. - * @param main {String}: entry point class name. - * @param files [{name: {String}, data: {String]: array of objects with filename and its content - * @param timeLimitCompileMs {Number}: Compilation timeout. - * @param timeLimitExecutionMs {Number}: Execution timeout. - */ -function runJava(messageId, main, files, timeLimitCompileMs, timeLimitExecutionMs) { - - let filesToAdd = files.length; - - const pack = tar.pack(); - - const tryTarAndRun = () => { - - filesToAdd--; - - if (filesToAdd === 0) { +const Promise = require('bluebird'); +const coroutine = Promise.coroutine; + + +const main = coroutine(function*() { + /** + * Command line arguments definition + * @type {[*]} + */ + const commandLineDef = [ + {name: 'port', alias: 'p', type: Number, defaultValue: 5016}, + {name: 'mongoAddress', alias: 'a', type: String, defaultValue: '127.0.0.1/queue'}, + {name: 'mongoCollection', alias: 'c', type: String, defaultValue: 'agendaJobs'}, + {name: 'defaultConcurrency', alias: 'd', type: Number, defaultValue: 40}, + {name: 'maxConcurrency', alias: 'm', type: Number, defaultValue: 70} + ]; + /** + * Object that for keys has command line argument names + * and for values its value. + * @type {Object} + */ + const commandLine = commandLineArgs(commandLineDef); + + + /** + * Server specified in server.js + * @type {Server} + */ + const server = yield require('./server')( + commandLine.port, + commandLine.mongoAddress, + commandLine.mongoCollection, + commandLine.defaultConcurrency, + commandLine.maxConcurrency + ); + + /** + * Object controlling docker specified in javaBox.js + * @type {EventEmitter} + */ + const javaBox = require('./javaBox'); + + /** + * Given code to execute and info about it, + * creates a tar with the code and passed the info and the tar to javaBox. + * + * @param messageId {String}: id of the given message/request. + * @param main {String}: entry point class name. + * @param files {{name: String, data: String}[]}: array of objects with filename and its content + * @param timeLimitCompileMs {Number}: Compilation timeout. + * @param timeLimitExecutionMs {Number}: Execution timeout. + */ + const runJava = coroutine(function*(messageId, main, files, timeLimitCompileMs, timeLimitExecutionMs) { + + const pack = tar.pack(); + //const packEntry = Promise.promisify(pack.entry, {context: pack}); + + Promise.map(files, function (file) { + //return packEntry({name: file.name}, file.data); + return pack.entry({name: file.name}, file.data); + + }).then(function () { const tarBuffer = pack.read(); javaBox.emit('runJava', messageId, main, tarBuffer, timeLimitCompileMs, timeLimitExecutionMs); - } - }; - - files.forEach((file) => { - pack.entry({ name: file.name }, file.data, tryTarAndRun); - }); -} - -/** - * Given code to execute and to test and info about it, - * creates a tar with the code and passed the info and the tar to javaBox. - * - * @param messageId {String}: id of the given message/request. - * @param tests [{name: {String}, data: {String]: array of objects with filename and its content. - * @param files [{name: {String}, data: {String]: array of objects with filename and its content. - * @param timeLimitCompileMs {Number}: Compilation timeout. - * @param timeLimitExecutionMs {Number}: Execution timeout. - */ -function runJunit(messageId, tests, files, timeLimitCompileMs, timeLimitExecutionMs) { - let filesToAdd = files.length + tests.length; + }); - const pack = tar.pack(); - - const tryTarAndRun = () => { - - filesToAdd--; - - if (filesToAdd === 0) { + }); + /** + * Given code to execute and to test and info about it, + * creates a tar with the code and passed the info and the tar to javaBox. + * + * @param messageId {String}: id of the given message/request. + * @param tests [{name: {String}, data: {String]: array of objects with filename and its content. + * @param files [{name: {String}, data: {String]: array of objects with filename and its content. + * @param timeLimitCompileMs {Number}: Compilation timeout. + * @param timeLimitExecutionMs {Number}: Execution timeout. + */ + const runJunit = coroutine(function *(messageId, tests, files, timeLimitCompileMs, timeLimitExecutionMs) { + + files = files.concat(tests); + + const pack = tar.pack(); + const packEntry = Promise.promisify(pack.entry, {context: pack}); + + Promise.map(files, function (file) { + return packEntry({name: file.name}, file.data); + + }).then(function () { const tarBuffer = pack.read(); javaBox.emit('runJunit', messageId, tests, tarBuffer, timeLimitCompileMs, timeLimitExecutionMs); - } - }; - files.forEach((file) => { - pack.entry({ name: file.name }, file.data, tryTarAndRun); - }); + }); - tests.forEach((test) => { - pack.entry({ name: test.name }, test.data, tryTarAndRun); }); -} -/** - * Given the feedback object passes it to the server. - * @param feedback {Object}, feedback/result object as specified in readme or javaBox.js - */ -function giveFeedBack(feedback) { + /** + * Given the feedback object passes it to the server. + * @param feedback {Object}, feedback/result object as specified in readme or javaBox.js + */ + function giveFeedBack(feedback) { - server.emit('result', feedback); -} + server.emit('result', feedback); + } + /** + * Specifying the server and javaBox control flow + */ + server.on('runJunit', runJunit); + server.on('runJava', runJava); + javaBox.on('result', giveFeedBack); +}); +main(); \ No newline at end of file diff --git a/containerOutputStream.js b/containerOutputStream.js new file mode 100644 index 0000000..638fa18 --- /dev/null +++ b/containerOutputStream.js @@ -0,0 +1,30 @@ +/** + * Module for creating custom writable streams in order to get the stdout and stderr from the containers + */ + +const stream = require('stream'); +const util = require('util'); +const Writable = stream.Writable; + + +function OutputStream(options) { + if (!(this instanceof OutputStream)) return new OutputStream(options); + + Writable.call(this, options); + this.memStore = new Buffer(''); + + this.toString = function () { + return this.memStore.toString() + } +} + +util.inherits(OutputStream, Writable); + +OutputStream.prototype._write = function (chunk, enc, cb) { + + const buffer = (Buffer.isBuffer(chunk)) ? chunk : new Buffer(chunk, enc); + this.memStore = Buffer.concat([this.memStore, buffer]); + cb(); +}; + +module.exports = OutputStream; \ No newline at end of file diff --git a/javaBox.js b/javaBox.js index 7dac31c..fdbaa7a 100644 --- a/javaBox.js +++ b/javaBox.js @@ -1,66 +1,97 @@ const EventEmitter = require('events'); +const Promise = require('bluebird'); +const coroutine = Promise.coroutine; +//const Docker = Promise.promisifyAll(require('dockerode')); const Docker = require('dockerode'); const concat = require('concat-stream'); +const OutputStream = require('./containerOutputStream'); + + + /** - * Docker object, connected on /var/run/docker.socket or default localhost docker port. - * @type {Docker} + * @typedef {Object} ExecutionOutput + * @type ExecutionOutput.messageId {String} + * @type ExecutionOutput.success {Boolean} + * @type ExecutionOutput.output {String} + * @type ExecutionOutput.errorMessage {String} + * @type ExecutionOutput.timeout {Boolean} + * */ -const docker = new Docker(); /** - * Time to wait between checking that the command has been executed (milliseconds). - * @type {number} + * Docker object, connected on /var/run/docker.socket or default localhost docker port. + * @type {Docker} */ -const EXEC_WAIT_TIME_MS = 250; - +const docker = Promise.promisifyAll(new Docker()); /** * JavaBox eventEmitter. * @type {EventEmitter} */ const javaBox = new EventEmitter(); -/** - * Initialising javaBox. - */ -javaBox.on('runJava', runJava); -javaBox.on('runJunit', runJunit); + /** * Creates a docker container with newly created execution * to run the Main.java specified inside the tar. * - * @param messageId {String}: id of the given message/request. - * @param main {String}: entry point class name. - * @param tarBuffer {String}: the buffer of the tar containing java files - * @param timeLimitCompileMs - * @param timeLimitExecutionMs + * @param {String} messageId : id of the given message/request. + * @param {String} main : entry point class name. + * @param {String} tarBuffer : the buffer of the tar containing java files + * @param {Number} timeLimitCompileMs + * @param {Number} timeLimitExecutionMs */ -function runJava(messageId, main, tarBuffer, timeLimitCompileMs, timeLimitExecutionMs) { +const runJava = coroutine(function*(messageId, main, tarBuffer, timeLimitCompileMs, timeLimitExecutionMs) { + const className = main.split('.')[0]; + let container = yield initializeContainer(tarBuffer, messageId); + const javacCmd = ['javac', '-cp', 'home', `home/${main}`]; const javaCmd = ['java', '-Djava.security.manager', '-cp', 'home', className]; - const sourceLocation = tarBuffer; - const execution = dockerCommand(javacCmd, timeLimitCompileMs, dockerCommand(javaCmd, timeLimitExecutionMs)); + try { + const compileOutput = yield runCommand(javacCmd, container, timeLimitCompileMs, 'compilation'); - createJContainer(messageId, sourceLocation, false, execution); -} -/** - * Creates a docker container with newly created execution - * to run the tests on files to test specified inside the tar. - * - * @param messageId {String}: id of the given message/request. - * @param junitFileNames [String]: array of different junit tests filename. - * @param tarBuffer {String}: the buffer of the tar containing java files (both test and testing). - * @param timeLimitCompileMs - * @param timeLimitExecutionMs - */ -function runJunit(messageId, junitFileNames, tarBuffer, timeLimitCompileMs, timeLimitExecutionMs) { + if (executedCorrectly(compileOutput)) { + + + const runtimeOutput = yield runCommand(javaCmd, container, timeLimitExecutionMs, 'execution'); + + if (executedCorrectly(runtimeOutput)) { + //TODO: maybe also return compileOutput for warnings? + emitSuccess(runtimeOutput); + + } else { + if (runtimeOutput.timeout) emitTimeout(runtimeOutput, 'execution'); + else emitWrong(runtimeOutput, 'Runtime'); + } + + } else { + if (compileOutput.timeout) emitTimeout(compileOutput, 'compilation'); + else emitWrong(compileOutput, 'Compile'); + } + + yield container.stopAsync(); + yield container.removeAsync({f: true}); + + + } catch (e) { // internal error + //TODO: specify javaBox behavior for internal error + console.log('500: Internal error with Docker!'); + emitServerError(e); + yield container.killAsync({t: 0}); + yield container.removeAsync({f: true}); + } + +}); + + +const runJunit = coroutine(function*(messageId, junitFileNames, tarBuffer, timeLimitCompileMs, timeLimitExecutionMs) { let junitFiles = []; junitFileNames.forEach((f)=>{ @@ -69,6 +100,8 @@ function runJunit(messageId, junitFileNames, tarBuffer, timeLimitCompileMs, time const className = 'TestRunner'; + let container = yield initializeContainer(tarBuffer, messageId, true); + const javacCmd = ['javac', '-cp', 'home:libs/junit-4.12:libs/hamcrest-core-1.3:libs/json-simple-1.1.1']; let javaCmd = ['java', '-cp', 'home:libs/junit-4.12:libs/hamcrest-core-1.3:libs/json-simple-1.1.1', className]; @@ -77,248 +110,114 @@ function runJunit(messageId, junitFileNames, tarBuffer, timeLimitCompileMs, time javaCmd.push(file); }); + try { + const compileOutput = yield runCommand(javacCmd, container, timeLimitCompileMs, 'compilation'); + if (executedCorrectly(compileOutput)) { - const sourceLocation = tarBuffer; - const execution = dockerCommand(javacCmd, timeLimitCompileMs, dockerCommand(javaCmd, timeLimitExecutionMs)); - - createJContainer(messageId, sourceLocation, true, execution); -} - - -/** - * Creates and starts a container with bash, JDK SE, maybe junit and executes the callback - * passing the container or error to it. - * - * @param messageId {String}: id of the given message/request. - * @param tarBuffer {String}: the buffer of the tar containing java files. - * @param isJunit {Boolean}: true if need to create container with junit support. - * @param callback {function(error, container)}: callback to operate on error and container or data in case of error, - * should accept two arguments. - */ -function createJContainer(messageId, tarBuffer, isJunit, callback) { - - let copyToCall = 4; - - const tryCallback = (container) => { + const runtimeOutput = yield runCommand(javaCmd, container, timeLimitExecutionMs, 'execution'); + if (executedCorrectly(runtimeOutput)) { + emitSuccess(runtimeOutput, true); - if (copyToCall === 0) { + } else { + if (runtimeOutput.timeout) emitTimeout(runtimeOutput, 'execution'); + else emitWrong(runtimeOutput, 'Runtime'); + } - callback(null, container); + } else { + if (compileOutput.timeout) emitTimeout(compileOutput, 'compilation'); + else emitWrong(compileOutput, 'Compile'); } - }; - const createOpts = {Image: 'openjdk:8u111-jdk', Tty: true, Cmd: ['/bin/bash']}; - docker.createContainer(createOpts, (err, container) => { - if (err) {callback(err, container); return} + yield container.stopAsync(); + yield container.removeAsync({f: true}); - container['messageId'] = messageId; + } catch (e) { // internal error + //TODO: specify javaBox behavior for internal error + console.log('500: Internal error with Docker!'); + emitServerError(e); + yield container.killAsync({t: 0}); + yield container.removeAsync({f: true}); + } - const startOpts = {}; - container.start(startOpts, (err, data) => { +}); - if (err) {callback(err, data); return} - if (isJunit) { +const initializeContainer = coroutine(function*(tarBuffer, messageId, junit) { - const tarOptsRunner = {path: 'home'}; - container.putArchive('./archives/TestRunner.class.tar', tarOptsRunner, (err, data) => { - copyToCall--; - if (err) callback(err, data); - else tryCallback(container); + junit = junit || false; - }); + const createOpts = {Image: 'openjdk:8u111-jdk', Tty: true, Cmd: ['/bin/bash']}; + let container = Promise.promisifyAll(yield docker.createContainerAsync(createOpts)); + container['messageId'] = messageId; - const tarOptsSecureTest = {path: 'home'}; - container.putArchive('./archives/SecureTest.class.tar', tarOptsSecureTest, (err, data) => { - copyToCall--; - if (err) callback(err, data); - else tryCallback(container); + const startOps = {}; + yield container.startAsync(startOps); - }); + const tarOptsSource = {path: 'home'}; + yield container.putArchiveAsync(tarBuffer, tarOptsSource); - const tarOptsLibs = {path: '/'}; - container.putArchive('./archives/libs.tar', tarOptsLibs, (err, data) => { - copyToCall--; - if (err) callback(err, data); - else tryCallback(container); - }); - } else { - copyToCall = 1; - } + if (junit) { + yield container.putArchiveAsync('./archives/SecureTest.class.tar', {path: 'home'}); + yield container.putArchive('./archives/libs.tar', {path: '/'}); + } - const tarOptsSource = {path: 'home'}; - container.putArchive(tarBuffer, tarOptsSource, (err, data) => { - copyToCall--; - if (err) callback(err, data); - else tryCallback(container); - }); - }); - }); -} + return container; +}); -/** - * Given a command, the timeout and the callback, returns a function that on some given container, - * executes the command, attaches the output listener and runs waitCmdExit. - * - * @param command {String}: command to be passed to container runtime environment. - * @param commandTimeLimitMs {Number}: execution timeout. - * @param callback {function(err, container)}: function to be executed after the command has finished. - * - * @return {function(err, container)}: function that executes the command on a container. - */ -function dockerCommand(command, commandTimeLimitMs, callback) { - - const opts = {Cmd: command, AttachStdout: true, AttachStderr: true}; - - const execution = (err, container) => { - - if (!err) container.exec(opts, (err, exec) => { - exec.start((err, stream) => { - - let stdOut = ''; - let stdErr = ''; - const stdOutConcat = concat({}, (data) => { - try { - stdOut += data; - } catch (e) { - stdOut = 'Output larger than 268435440 bytes.'; - } - }).on('error', (err) => {}); - const stdErrConcat = concat({}, (data) => { - try { - stdErr += data; - } catch (e) { - stdOut = 'Output larger than 268435440 bytes.'; - } - }).on('error', (err) => {}); - - container.modem.demuxStream(stream, stdOutConcat, stdErrConcat); - - - const streamInfo = { - - getOut: () => {return stdOut}, - getErr: () => {return stdErr}, - endStream: () => { - stdOutConcat.end(); - stdErrConcat.end(); - } - }; - - waitCmdExit(container, exec, callback, streamInfo, commandTimeLimitMs); - }); - }); - }; - - return execution; -} /** - * Given an execution of a command on a container, a stream handler and command timeout - * waits for the command to be finished or timeout to be expired and calls the callback - * if it isn't null, otherwise it calls the feedbackAndClose function. + * Run a command inside a container, capture and return output streams and successful execution. + * + * @param command {String[]}: A command with arguments to execute + * @param container {Container}: The container, properly initialised (from initializeContainer), for execution + * @param executionTimeLimit {Number} [Optional]: The number in milliseconds after which, execution should stop + * @return The result of the execution indicating if the execution was successful and the output of the stdout/ stderr. + * @return result {ExecutionOutput} + * @throws Error * - * @param container {Container}: Active docker container. - * @param exec {Object}: Docker execution object. - * @param callback {function(err, container)}: function to be executed after the command has finished. - * @param streamInfo {Object}: returns stdOut, stdIn and closes concat stream - * @param commandTimeOutMs {Number}: execution timeout in ms. - * @param previousTimeMs {Number}: ms already spent on this execution, called when the function is called - * recursively, should not be passed otherwise. */ -function waitCmdExit(container, exec, callback, streamInfo, commandTimeOutMs, previousTimeMs){ - - let timeSpentMs = previousTimeMs || 0; - - const checkExit = (err, data) => { - - if (data.Running) { // command is still running, check later or send time out - - timeSpentMs += EXEC_WAIT_TIME_MS; // count time spent - - if (timeSpentMs >= commandTimeOutMs){ // time period expired - - feedbackAndClose(container, streamInfo, false, true); +const runCommand = coroutine(function *(command, container, executionTimeLimit) { - } else { + let timeoutCallback = null; + let timeoutHappend = false; + const stdoutStream = new OutputStream(); + const stderrStream = new OutputStream(); - waitCmdExit(container, exec, callback, streamInfo, commandTimeOutMs, timeSpentMs); - } + const execOpts = {Cmd: command, AttachStdout: true, AttachStderr: true, Tty: false}; // execution options + const exec = Promise.promisifyAll(yield container.execAsync(execOpts)); // create execution of command - } else if ((data.ExitCode === 0) && (callback)) { // command successful, has next command + const stream = yield exec.startAsync(); // start execution (get an output stream) + container.modem.demuxStream(stream, stdoutStream, stderrStream); // intercept container output to our streams - callback(null, container); + if (executionTimeLimit) { // start keeping time + timeoutCallback = setTimeout(() => { + timeoutHappend = true + }, executionTimeLimit); // prepare timeout termination + } - } else if (data.ExitCode === 0) { // command successful, it was the last command - feedbackAndClose(container, streamInfo, true, false) + let executionData = yield exec.inspectAsync(); + while (executionData.Running && timeoutHappend == false) { // loop (asynchronously) until execution stops + executionData = yield exec.inspectAsync(); + } - } else { // command failed - feedbackAndClose(container, streamInfo, false, false); - } - }; + const executedSuccessfully = (executionData.ExitCode == 0 && timeoutHappend == false); // set successful execution mark - setTimeout(() => exec.inspect(checkExit), EXEC_WAIT_TIME_MS); -} + if (executionTimeLimit) clearTimeout(timeoutCallback); -/** - * Closes the stream, parses it assuming it could be a specific junit output, - * accordingly creates a feedback and emits it, closing and deleting docker container - * at the end. - * - * @param container {Container}: Active docker container. - * @param streamInfo Object, returns stdOut, stdIn and closes concat stream - * @param passed Boolean, true if no compile or runtime error during normal execution - * @param timeOut Boolean, true if timeout time elapsed - * - * @feedback - * if test files exist (junit output) and passed is true: - * { - * messageId, - * passed: Boolean (false if compile/runtime errors true otherwise), - * output: String, - * errorMessage: String (empty if `passed` is true), - * timeOut: Boolean, - * totalNumberOfTests: Integer, - * numberOfTestsPassed: Integer, - * testsOutput: String (output of all failed tests) - * } - * otherwise: - * { - * messageId, - * passed: Boolean (false if compile/runtime errors true otherwise), - * output: String, - * errorMessage: String (empty if `passed` is false), - * timeOut: Boolean - * } - */ -function feedbackAndClose(container, streamInfo, passed, timeOut) { - - streamInfo.endStream(); - - const feedback = { + return { messageId: container.messageId, - passed: passed, - output: (!timeOut) ? streamInfo.getOut() : '', - errorMessage: (!timeOut) ? streamInfo.getErr() : "Reached maximum time limit", - timeOut: timeOut - - }; + success: executedSuccessfully, + output: stdoutStream.toString(), + errorMessage: stderrStream.toString(), + timeout: timeoutHappend + } - const parsed = parseOutput(feedback.output); - feedback.output = parsed.normalOutput; - feedback.totalNumberOfTests = parsed.totalNumberOfTests; - feedback.numberOfTestsPassed = parsed.numberOfTestsPassed; - feedback.testsOutput = parsed.testsOutput; - javaBox.emit('result', feedback); - - container.kill({}, () => - container.remove({v: true}, () => {})); -} +}); /** * Given a possibly junit output, parses it and if it's junit, adds @@ -352,5 +251,78 @@ function parseOutput(output){ return outputObject; } +/** + * Check if a command executed correctly, by analyzing the result object. + * + * @param result {Object || ExecutionOutput} + * @param result.timeout: {Boolean} + * @param result.success: {Boolean} + * @return {boolean}: `true` if command exited with code 0 and no time out occurred, `false` otherwise. + */ +function executedCorrectly(result) { + return (result.success && !result.timeout); +} + +function emitServerError(e) { + throw e; +} + +/** + * Emit `result` event with success values. + * + * @param executionOutput {ExecutionOutput}: Output of a run command execution + * @param junit {Boolean} [Optional][Default: false]: Set to true if execution output is from junit orchestrator class for parsing. + */ +function emitSuccess(executionOutput, junit) { + junit = junit || false; + const feedback = executionOutput; + + if (junit) { + const parsed = parseOutput(feedback.output); + feedback.output = parsed.normalOutput; + feedback.totalNumberOfTests = parsed.totalNumberOfTests; + feedback.numberOfTestsPassed = parsed.numberOfTestsPassed; + feedback.testsOutput = parsed.testsOutput; + } + javaBox.emit('result', feedback); +} + +/** + * Emit `result` event with timeout values and error message. + * + * @param feedback {Object}: The feedback with all the info to be emitted back. + * @param stage {String}[Optional]: In which stage this function is called. + */ +function emitTimeout(feedback, stage) {//messageId, stdoutStream, stderrStream, stage) { + + + feedback.errorMessage += '\n>>>JavaBox: Timeout reached'; + + if (stage) { + feedback.errorMessage += ` during ${stage}.`; + } else { + feedback.errorMessage += '.'; + } + + javaBox.emit('result', feedback); + +} + +/** + * Emit `result` event with error values. + * + * @param executionOutput {ExecutionOutput}: Output of a run command execution. + * @param stage {String}[Optional]: In which stage this function is called. + */ +function emitWrong(executionOutput, stage) { + javaBox.emit('result', executionOutput); +} + + +/** + * Initialising javaBox. + */ +javaBox.on('runJava', runJava); +javaBox.on('runJunit', runJunit); -module.exports = javaBox; +module.exports = javaBox; \ No newline at end of file diff --git a/package.json b/package.json index 368bc85..eecceff 100644 --- a/package.json +++ b/package.json @@ -1,7 +1,7 @@ { "name": "javabox", - "version": "1.0.0", - "description": "Server to procede java code", + "version": "1.1.0", + "description": "Server for compiling and running Java code", "main": "index.js", "scripts": { "test": "echo \"Error: no test specified\" && exit 1", @@ -15,6 +15,7 @@ "concat-stream": "^1.5.2", "dockerode": "^2.3.1", "json-socket": "^0.2.0", - "tar-stream": "^1.5.2" + "tar-stream": "^1.5.2", + "bluebird": "^2.8.2" } } diff --git a/queue.js b/queue.js new file mode 100644 index 0000000..e14ad93 --- /dev/null +++ b/queue.js @@ -0,0 +1,71 @@ +const Agenda = require('agenda'); + +/** + * @typedef {Agenda} queue + */ +const queue = new Agenda(); + +/** + * @typedef {Object} JsonSocket + * @typedef {Object} Request + * + * @typedef {Object} StoredMessage + * @type StoredMessage.socket: {JsonSocket} + * @type StoredMessage.request: {Request} + * @type StoredMessage.done: Function + */ + +/** + * Initialize queue, connect to the database and define actions + * @param queueParams {Object} + * @param queueParams.mongoFullAddress {string} + * @param queueParams.mongoCollection {string} + * @param queueParams.maxConcurrency {int} + * @param queueParams.defaultConcurrency {int} + * @param processMessageJob {Function}: A callback for Agenda.now() for 'process_message' action + * + * @return {Promise}: A promise for starting the queue + */ +Agenda.prototype.initialize = function (queueParams, processMessageJob) { + + return new Promise((resolve, reject) => { + + queue.database(queueParams.mongoFullAddress, queueParams.mongoCollection) + .defaultConcurrency(queueParams.defaultConcurrency) + .maxConcurrency(queueParams.maxConcurrency); + + queue.define('process_message', processMessageJob); + + queue.on('ready', resolve); // use .then(queue.start()) + queue.on('error', reject); + }); +}; + +Agenda.prototype.addMessage = function (request) { + + const messageId = createMessageId(request); + const jobData = { + messageId: messageId, + request: request + }; + queue.now('process_message', jobData); + return messageId; +}; + +/** + * Given a clientId creates an unique messageId. + * + * @param message {Object}: connection message + * @param message.clientId: client UID + * @return {string} clientId + ::: + current date in milliseconds + */ +function createMessageId(message) { + return `${message.clientId}:::${Date.now()}`; +} + +module.exports = queue; + + + + + diff --git a/server.js b/server.js index 32fc7bd..7b918b0 100644 --- a/server.js +++ b/server.js @@ -1,29 +1,44 @@ const net = require('net'); const JsonSocket = require('json-socket'); -const Agenda = require('agenda'); +const Promise = require('bluebird'); +const coroutine = Promise.coroutine; +/** + * @typdef {Object} JsonSocket + * @typedef {Object} Request + */ /** - * Contains preceding messages and its related information as such - * {messageId = {socket: JsonSocket, request: Object, done: Function}, ...}. + * "Dictionary" containing preceding messages and its related information as such + * "messageId" {String} -> { + * socket: {JsonSocket}, + * request: {Request}, + * done: Function + * } * @type {Object} */ const messages = {}; /** * Agenda object, will be initialised in init function. - * @type {Agenda} + * @type {Queue} + */ +const queue = require('./queue'); + +/** + * @typedef {Object} Server */ -const queue = new Agenda(); /** * Tcp server. * @type {Server} */ const server = net.createServer(); +//TODO: is this correct? +server.listen = Promise.promisify(server.listen); + /** - * Initialises server object, connects to mongo database, - * and starts the server on given port. + * Initialises server and queue for storing requests. * * @param port {Number}: tcp port number. * @param mongoAddress {String}: mongo database url. @@ -33,51 +48,57 @@ const server = net.createServer(); * * @return {Server}: server event emitter object () */ -function init(port, mongoAddress, mongoCollection, defaultConcurrency, maxConcurrency) { +const init = coroutine(function*(port, mongoAddress, mongoCollection, defaultConcurrency, maxConcurrency) { const mongoFullAddress = `mongodb://${mongoAddress}`; - queue.database(mongoFullAddress, mongoCollection) - .defaultConcurrency(defaultConcurrency) - .maxConcurrency(maxConcurrency); - queue.define('process_message', processMessageJob); - queue.on('ready', () => queue.start()); + const queueInitParams = { + mongoFullAddress, + mongoCollection, + maxConcurrency, + defaultConcurrency + }; + yield queue.initialize(queueInitParams, processMessageJob); + queue.start(); + server.on('connection', initSocket); server.on('result', sendResult); - server.listen(port); + + yield server.listen(port); + console.log(`JavaBox listening on port ${port}...`); + return server; -} +}); /** - * Given a connection creates putMessageInQueue function, - * creates a json socket and initialises it to accept - * the request with putMessageInQueue function. + * When a new connection is established, + * creates a new socket + * and inserts request object into the queue for execution. * @param connection {Object}: A tcp connection. */ function initSocket(connection) { - const socket = new JsonSocket(connection); - const putMessageInQueue = (request) => { + const socket = new JsonSocket(connection); + socket.on('message', addMessageInQueue.bind(null, socket)); +} - const messageId = createMessageId(request); - const jobData = { - messageId: messageId, - request: request - }; - messages[messageId] = {socket: socket, request: request}; - queue.now('process_message', jobData); +/** + * Adds a request from a socket to the queue for execution. + * @param socket: {JsonSocket} + * @param request: {Request} + */ +function addMessageInQueue(socket, request) { - }; - socket.on('message', putMessageInQueue); + const messageId = queue.addMessage(request); + messages[messageId] = {socket: socket, request: request}; } /** - * Given a job and done function, stores done function, - * reads jobs attributes and call parseRequestAndSend function on them. - * @param job {Object}: Agenda job. - * @param done {Object}: job terminating function. + * Process the execution of a job-request. + * @param job {Object}: An `Agenda` job. + * @param done {Function}: A callback to execute when finished, in case the function is asynchronous. */ function processMessageJob(job, done) { @@ -85,18 +106,23 @@ function processMessageJob(job, done) { const messageId = job.attrs.data.messageId; messages[messageId]['done'] = done; - parseRequestAndSend(request, messageId); -} + if (isRequestValid(request)) { + executeRequest(request, messageId); + + } else { + sendResultForBadRequest(messageId); + } + + //done(); // not needed, useful for async code +} /** - * Given a request and its id, parses the request, checks if it's ok, - * and emits the right event. - * @param request {Object}, request object as specified in readme. - * @param messageId {String}, id of the given message/request. + * Checks whether a request is valid or not, given the request specification + *(See: https://github.com/ASQ-USI/asq-java-q-backend/blob/master/README.md#communication-api) + * @param request {Request}: The request to be checked. + * @return {boolean} : Returns `false` if request has one or more missing properties or illogical values, `true` otherwise. */ -function parseRequestAndSend(request, messageId) { - - const clientId = request.clientId; +function isRequestValid(request) { const main = request.submission.main; const files = request.submission.files; const tests = request.submission.tests; @@ -104,16 +130,43 @@ function parseRequestAndSend(request, messageId) { const timeLimitExecutionMs = request.executionTimeoutMs; const charactersMaxLength = request.charactersMaxLength; - - if (!((main || tests) && files && timeLimitCompileMs && timeLimitExecutionMs && charactersMaxLength)) { - sendResult({clientId: clientId}); - return; + if (!((main || tests) && files && timeLimitCompileMs && timeLimitExecutionMs && charactersMaxLength)) { // absence of parameter + return false; + } else if (!(timeLimitCompileMs > 0 && timeLimitExecutionMs > 0)) { // illogical values for timeout + return false; + } else { + return true; } +} - if (!(timeLimitCompileMs > 0 && timeLimitExecutionMs > 0)) { // illogical values for timeout - sendResult({clientId: clientId}); - return; +/** + * Sends back to corresponding client error code and closes connection. + * @param messageId {String}: The id in the queue of the bad request message. + */ +function sendResultForBadRequest(messageId) { + const message = messages[messageId]; + const socket = message.socket; + const clientId = message.request.clientId; + try { + socket.sendEndMessage({}); //TODO: define output for bad requests + } catch (e) { + console.log(`73: Socket with client ${clientId} closed before sending result back.`); } +} + +/** + * Executes the request by emitting 'runJunit' or 'runJava' event. + * @param request {Request} + * @param messageId {String}, id of the given message/request. + */ +function executeRequest(request, messageId) { + + console.log('executing request for ' + messageId); + const main = request.submission.main; + const files = request.submission.files; + const tests = request.submission.tests; + const timeLimitCompileMs = request.compileTimeoutMs; + const timeLimitExecutionMs = request.executionTimeoutMs; if (tests && Array.isArray(tests) && tests.length > 0) { // run junit tests @@ -126,25 +179,15 @@ function parseRequestAndSend(request, messageId) { } } - /** - * Given a clientId creates an unique messageId. - * - * @param message {string} Some random clientId - * @return {string} clientId + ::: + current date in milliseconds - */ -function createMessageId(message) { - - return `${message.clientId}:::${Date.now()}`; -} - - -/** - * Correct and truncates the feedback and sends it back. - * @param feedback {Object}, feedback/result object as specified in readme or javaBox.js + * Correct and truncates the feedback and sends it back. Also closes connection. + * @param feedback {Object}: feedback/result object as specified in readme or javaBox.js + * @param feedback.messageId {string} */ function sendResult(feedback) { + console.log('>>sending back to ' + feedback.messageId); + const message = messages[feedback.messageId]; const socket = message.socket; const clientId = message.request.clientId; @@ -163,7 +206,7 @@ function sendResult(feedback) { try { socket.sendEndMessage(feedback); } catch (e) { - console.log('socket closed before sending result back'); + console.log(`73: Socket with client ${clientId} closed before sending result back.`); } done(); diff --git a/test/client.js b/test/client.js index 6513e6e..66718ed 100644 --- a/test/client.js +++ b/test/client.js @@ -11,8 +11,8 @@ const host = '127.0.0.1'; * @type {[*]} */ const commandLineDef = [ - { name: 'clients', alias: 'c', type: Number, defaultValue: 1}, - { name: 'submission', alias: 's', type: String, defaultValue: 'hwSub'} + {name: 'clients', alias: 'c', type: Number, defaultValue: 50}, + {name: 'submission', alias: 's', type: String, defaultValue: 'infiniteSub'} ]; /** * Object that for keys has command line argument names @@ -53,6 +53,7 @@ function simpleSubmission(command) { const submissions = { hwSub: simpleSubmission(`System.out.println("Hello world."); `), infiniteSub: simpleSubmission(`while (true) { System.out.println("To infinity and beyond!"); }`), + exceptionPlusOutput: simpleSubmission("System.out.println(\"Standard output message\" ); String s=null; s.toString();"), rmSub: { main: 'RemoveSub.java', files: [{ @@ -79,7 +80,8 @@ const submissions = { name: 'TestJunit2.java', data: fs.readFileSync('src/TestJunit2.java', 'utf8'), }] - } + }, + }; @@ -110,8 +112,8 @@ function makeConnection(clientId) { const message = { clientId : clientId, submission : submission, - compileTimeoutMs : 60000, - executionTimeoutMs : 60000, + compileTimeoutMs: 15000, + executionTimeoutMs: 6000, charactersMaxLength: 1000 }; @@ -131,9 +133,8 @@ function makeConnection(clientId) { console.log(`Clients left: ${clientsLeft}\n \n`); }); }); -}; - +} for (var i = 0; i < clientsNumber; i++) { const clientId = 'client' + i; makeConnection(clientId); -}; +}