kinvey-code-task-runner/lib/receivers/tcpReceiver.js

// Copyright (c) 2018 Kinvey Inc.
//
// Licensed 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.

const net = require('net');
const log = require('../log/logger');
const { serialize } = require('../utils');

function getSerializableError(err) {
  let errorToLog = err.toString();

  if (!(err instanceof Error)) {
    log.debug('Error argument not an instance of Error');
    try {
      serialize(err); // if err has circular references, this would fail
      errorToLog = err;
    } catch (err) {
      const msg = 'Error argument not instance of Error and not stringifiable';
      errorToLog = msg;
      log.debug(msg);
    }
  }

  return errorToLog;
}

module.exports = (() => {
  let server = {};

  function composeErrorReply(errorMessage, debugMessage, err, metadata) {
    // build an error response as generated by blApi
    // FIXME:  error response format should be handled by the outermost layer
    // and not be duplicated in various places.

    return {
      isError: true,
      error: getSerializableError(err),
      message: errorMessage,
      debugMessage,
      // note: stack is not a stack array, it's a formatted text string that contains a printed stack trace
      stackTrace: err.stack,
      metadata: metadata || {}
    };
  }

  function parseTask(task, callback) {
    log.debug('parseTask Invoked');
    log.debug(task);

    if (!task) {
      return callback(new Error('Invalid or missing task object'));
    }

    if (typeof task === 'object') {
      return callback(null, task);
    }

    let parsedTask;

    try {
      parsedTask = JSON.parse(task);
      log.debug('Task parsed');
    } catch (e) {
      log.debug('Task Parse failed');
      log.debug(e);
      return callback(new Error('invalid task: unable to parse task json'));
    }

    return callback(null, parsedTask);
  }

  function startServer(taskReceivedCallback, startedCallback, options) {
    server = net.createServer((socket) => {
      log.info('Connection established');
      let clientDisconnected = false;
      let data = '';
      let lineEnd;
      let task;

      function validateSocketReadiness() {
        if (clientDisconnected) {
          return 'Connection ended by client.';
        } else if (socket.destroyed) {
          return 'Connection lost - cannot write response.';
        }
        return null;
      }

      function writeResponse(data) {
        const msg = validateSocketReadiness();
        if (msg) {
          log.error(msg);
          return msg;
        }
        socket.write(serialize(data));
        socket.write('\n');
        return null;
      }

      function processTask(task) {
        setImmediate(() => {
          log.debug('About to parse task');
          log.debug(`Task: ${task}`);

          parseTask(task, (parseError, parsedTask) => {
            log.debug('Task parsing complete');

            if (parseError) {
              log.error(`Parse error! ${parseError}`);
              parseError.isError = true;
              writeResponse(composeErrorReply('Internal Error', parseError.toString(), parseError));
              return;
            }

            // Set task request.query to request.params to account for difference in BL tasks
            if (parsedTask && parsedTask.request && parsedTask.request.params && !parsedTask.request.query) {
              parsedTask.request.query = parsedTask.request.params;
            }

            log.debug('About to invoke taskReceivedCallback');
            log.debug(parsedTask);

            taskReceivedCallback(parsedTask, (err, result) => {
              log.debug('About to respond');
              // processBL returns  pre-assembled response object

              if (err) {
                log.debug('Responding with error');
                log.error(err);
                return writeResponse(composeErrorReply('Internal Error', 'Unable to execute Flex method', err));
              }

              log.debug('Responding with success');
              log.debug(result);

              // Make sure result is present, undefined is not stringified into JSON
              if (typeof result === 'undefined') {
                result = null;
              }

              // TODO: Check the size of ret and use a non-blocking serializer if too large
              return writeResponse(result);
            });
          });
        });
      }

      socket.on('error', (err) => {
        // Avoid crashing the process on errors for a single connection
        log.error(err.stack);
      });

      socket.on('data', (chunk) => {
        log.debug('chunk received');

        // Always append the new data, then process task below
        data += chunk.toString();

        // process one task at a time in arrival order, sending back responses in the same order
        // an already running processing loop will handle the newly arrived task as well
        lineEnd = data.indexOf('\n');

        while (lineEnd > -1) {
          task = data.slice(0, lineEnd + 1);
          data = data.slice(lineEnd + 1);

          if (task == null) {
            // Task is invalid, can't process.  Can't send an error back because we don't have a task Id.
            // Break out of the loop as no more data will be in data.slice
            break;
          }

          // TODO: do not ^^^ recopy all the data for each task line, index into data instead
          // skip blank lines, processBL non-blank ones

          if (task.indexOf('{"healthCheck":1}') > -1) {
            log.info('healthcheck!');
            writeResponse({ status: 'ready' });
            return;
          }

          processTask(task);
          lineEnd = data.indexOf('\n');
        }
      });

      socket.on('end', () => {
        clientDisconnected = true;
      });
    });

    setImmediate(() => {
      server.listen(options.port || 7000, () => {
        startedCallback();
      });
    });
  }

  function stop(callback) {
    server.close(callback);
  }

  return {
    startServer,
    stop
  };
})();