module.exports = function (RED) {
  var spawn = require("child_process").spawn;
  var util = require("util");

  function indentLines(fnCode, depth) {
    return fnCode
      .split("\n")
      .map((line) => Array(depth).join(" ") + line)
      .join("\n");
  }

  function spawnFn(self) {
    self.child = spawn("python", ["-uc", self.func.code], {
      stdio: ["pipe", "pipe", "pipe", "pipe", "ipc"],
    });
    self.child.stdout.on("data", function (data) {
      self.log(data.toString());
    });
    self.child.stderr.on("data", function (data) {
      self.error(data.toString());
    });
    self.child.on("close", function (exitCode) {
      if (exitCode) {
        self.error(`Python Function process exited with code ${exitCode}`);
        if (self.func.attempts) {
          spawnFn(self);
          self.func.attempts--;
        } else {
          self.error(
            `Function '${self.name}' has failed more than 10 times. Fix it and deploy again`
          );
          self.status({
            fill: "red",
            shape: "dot",
            text: "Stopped, see debug panel",
          });
        }
      }
    });
    self.child.on("message", function (response) {
      switch (response.ctx) {
        case "send":
          sendResults(self, response.msgid, response.value);
          clearCachedMsg(self, response.msgid);
          break;
        case "log":
        case "warn":
        case "error":
        case "status":
          self[response.ctx].apply(self, response.value);
          break;
        default:
          clearCachedMsg(self, response.msgid);
          throw new Error(`Don't know what to do with ${response.ctx}`);
      }
    });
    self.log(`Python function '${self.name}' running on PID ${self.child.pid}`);
    self.status({ fill: "green", shape: "dot", text: "Python" });
  }

  function clearCachedMsg(self, msgid) {
    if (!self.cache || !msgid) return;
    self.cache.delete(msgid);
  }

  function sendResults(self, _msgid, msgs) {
    if (msgs == null) {
      return;
    } else if (!util.isArray(msgs)) {
      msgs = [msgs];
    }
    var msgCount = 0;
    for (var m = 0; m < msgs.length; m++) {
      if (msgs[m]) {
        if (util.isArray(msgs[m])) {
          for (var n = 0; n < msgs[m].length; n++) {
            msgs[m][n]._msgid = _msgid;
            msgCount++;
          }
        } else {
          msgs[m]._msgid = _msgid;
          msgCount++;
        }
      }
    }
    if (msgCount > 0) {
      const msgCache = self.cache?.get(_msgid);
      if (msgCache) {
        // Restore REQ object if it exists.
        if (msgCache.req !== undefined) {
          msgs[0].req = msgCache.req;
        }
        // Restore RES object if it exists.
        if (msgCache.res !== undefined) {
          msgs[0].res = msgCache.res;
        }
      }

      self.send(msgs);
    }
  }

  function PythonFunction(config) {
    var self = this;
    RED.nodes.createNode(self, config);
    self.name = config.name;
    self.cache = new Map();
    self.func = {
      code:
        `
import os
import json
import asyncio
import functools
import concurrent.futures


class Msg(object):
    SEND = 'send'
    LOG = 'log'
    WARN = 'warn'
    ERROR = 'error'
    STATUS = 'status'

    def __init__(self, ctx, value, msgid):
        self.ctx = ctx
        self.value = value
        self.msgid = msgid

    def dumps(self):
        return json.dumps(vars(self)) + "\\n"

    @classmethod
    def loads(cls, json_string):
        return cls(**json.loads(json_string))


class Node(object):
    def __init__(self, msgid, writer, writer_sync):
        self.__msgid = msgid
        self.__writer = writer
        self.__writer_sync = writer_sync

    async def send(self, msg):
        msg = Msg(Msg.SEND, msg, self.__msgid)
        await self.send_to_node(msg)

    def log(self, *args):
        msg = Msg(Msg.LOG, args, self.__msgid)
        self.send_to_node_sync(msg)

    def warn(self, *args):
        msg = Msg(Msg.WARN, args, self.__msgid)
        self.send_to_node_sync(msg)

    def error(self, *args):
        msg = Msg(Msg.ERROR, args, self.__msgid)
        self.send_to_node_sync(msg)

    def status(self, *args):
        msg = Msg(Msg.STATUS, args, self.__msgid)
        self.send_to_node_sync(msg)

    async def send_to_node(self, msg):
        self.__writer.write(msg.dumps().encode('utf-8'))
        await self.__writer.drain()

    def send_to_node_sync(self, msg):
        # 当数据比较大时，下面方法将会报错
        # self.__writer_sync.write(msg.dumps().encode('utf-8'))

        self.__writer.write(msg.dumps().encode('utf-8'))
        # 是否需要 drain，可以不用 drain，或者在合适的地方 drain？
        # 比如 send_to_node 时 drain
        # https://stackoverflow.com/questions/53779956/why-should-asyncio-streamwriter-drain-be-explicitly-called
        # self.__writer.drain()


def python_function(msg, node):
` +
        indentLines(config.func, 4) +
        `
async def connect_stdin_stdout(loop):
    r_channel = os.fdopen(3, "r+b", buffering=0)
    # w_fd = os.dup(3)
    w_channel = os.fdopen(4, "r+b", buffering=0)
    
    ## Connect reader with pipe
    reader = asyncio.StreamReader(limit=2**27)
    r_protocol = asyncio.StreamReaderProtocol(reader)
    await loop.connect_read_pipe(lambda: r_protocol, r_channel)
    
    ## Connect writer with pipe
    # Method 1:
    # w_protocol = asyncio.StreamReaderProtocol(asyncio.StreamReader())
    # w_transport, _ = await loop.connect_write_pipe(
    #     lambda: w_protocol,
    #     w_channel
    # )
    # writer = asyncio.StreamWriter(w_transport, w_protocol, None, loop)
    # Method 2:
    w_transport, w_protocol = await loop.connect_write_pipe(
        asyncio.streams.FlowControlMixin,
        w_channel
    )
    writer = asyncio.StreamWriter(w_transport, w_protocol, reader, loop)

    return reader, writer, w_channel

async def run_task(msg, node):
    loop = asyncio.get_event_loop()
    python_function_with_msg = functools.partial(python_function, msg, node)

    # executor 选择
    # 1. 文件操作使用 thread pool
    # 2. CPU-bound 操作使用一般情况更适合使用 process pool

    # 方法一 process pool
    # 在 process pool 中使用 node 报错：
    # TypeError: cannot pickle '_io.FileIO' object
    # with concurrent.futures.ProcessPoolExecutor(max_workers=10) as pool:
    #     res_msgs = await loop.run_in_executor(
    #         pool, python_function_with_msg)
    #     await node.send(res_msgs)

    # 方法二 thread pool
    with concurrent.futures.ThreadPoolExecutor(max_workers=10) as pool:
        res_msgs = await loop.run_in_executor(
            pool, python_function_with_msg)
        await node.send(res_msgs)
  
    # 方法三 pathos.multiprocessing

async def main():
    loop = asyncio.get_event_loop()
    reader, writer, writer_sync = await connect_stdin_stdout(loop)
    while True:
        raw_msg = await reader.readline()
        if not raw_msg:
            raise RuntimeError('Received EOF!')
        msg = json.loads(raw_msg)
        msgid = msg["_msgid"]
        node = Node(msgid, writer, writer_sync)

        asyncio.create_task(run_task(msg, node))


asyncio.run(main())
`,
      attempts: 10,
    };

    spawnFn(self);

    self.on("input", function (msg) {
      const msgCache = {};
      // Save REQ object if it exists.
      if (msg.req !== undefined) {
        msgCache.req = msg.req;
      }
      // Save RES object if it exists.
      if (msg.res !== undefined) {
        msgCache.res = msg.res;
      }
      self.cache.set(msg._msgid, msgCache);

      const jsonMsg = serializeMsg(msg);

      // 这里没有使用 self.child.send(JSON.parse(jsonMsg))
      // python 的 asyncio 不支持在同一个管道上读写，所以 nodejs
      // 的 ipc 管道用来接收数据，使用普通的 pipe 发送数据。
      self.child.stdio[3].write(`${jsonMsg}\n`);
    });
    self.on("close", function () {
      self.child.kill();
      self.cache.clear();
    });
  }
  RED.nodes.registerType("python-function", PythonFunction);
};

function serializeMsg(msg) {
  // 避免 payload 属性因为循环引用被删除
  if (
    msg.req &&
    (msg.payload === msg.req.query || msg.payload === msg.req.body)
  ) {
    msg.payload = { ...msg.payload };
  }

  let cache = [];

  // 删除重复引用的数据
  // 其它方法：https://github.com/douglascrockford/JSON-js/blob/master/cycle.js
  const stringified = JSON.stringify(msg, function (_, value) {
    if (typeof value === "object" && value !== null) {
      if (cache.indexOf(value) !== -1) {
        // Circular reference found, discard key
        return;
      }
      // Store value in our collection
      cache.push(value);
    }
    return value;
  });

  cache = null; // Enable garbage collection

  return stringified;
}