<commit_before>/* @flow */

import fs from "fs";
import { Message, Socket } from "jmp";
import v4 from "uuid/v4";

import Kernel from "./kernel";
import InputView from "./input-view";
import kernelManager from "./kernel-manager";
import { log } from "./utils";

type Connection = {
  control_port: number,
  hb_port: number,
  iopub_port: number,
  ip: string,
  key: string,
  shell_port: number,
  signature_scheme: string,
  stdin_port: number,
  transport: string,
  version: number
};
/* eslint-disable camelcase */
export default class ZMQKernel extends Kernel {
  executionCallbacks: Object = {};
  connection: Connection;
  connectionFile: string;
  kernelProcess: child_process$ChildProcess;

  shellSocket: Socket;
  controlSocket: Socket;
  stdinSocket: Socket;
  ioSocket: Socket;

  constructor(
    kernelSpec: Kernelspec,
    grammar: atom$Grammar,
    connection: Connection,
    connectionFile: string,
    kernelProcess: child_process$ChildProcess
  ) {
    super(kernelSpec, grammar);
    this.connection = connection;
    this.connectionFile = connectionFile;
    this.kernelProcess = kernelProcess;

    this._connect();

    if (this.kernelProcess) {
      log("ZMQKernel: @kernelProcess:", this.kernelProcess);

      this.kernelProcess.stdout.on("data", (data: string | Buffer) => {
        data = data.toString();

        if (atom.config.get("Hydrogen.kernelNotifications")) {
          atom.notifications.addInfo(this.kernelSpec.display_name, {
            description: data,
            dismissable: true
          });
        } else {
          log("ZMQKernel: stdout:", data);
        }
      });

      this.kernelProcess.stderr.on("data", (data: string | Buffer) => {
        atom.notifications.addError(this.kernelSpec.display_name, {
          description: data.toString(),
          dismissable: true
        });
      });
    } else {
      log("ZMQKernel: connectionFile:", this.connectionFile);
      atom.notifications.addInfo("Using an existing kernel connection");
    }
  }

  _connect() {
    const scheme = this.connection.signature_scheme.slice("hmac-".length);
    const { key } = this.connection;

    this.shellSocket = new Socket("dealer", scheme, key);
    this.controlSocket = new Socket("dealer", scheme, key);
    this.stdinSocket = new Socket("dealer", scheme, key);
    this.ioSocket = new Socket("sub", scheme, key);

    const id = v4();
    this.shellSocket.identity = `dealer${id}`;
    this.controlSocket.identity = `control${id}`;
    this.stdinSocket.identity = `dealer${id}`;
    this.ioSocket.identity = `sub${id}`;

    const address = `${this.connection.transport}://${this.connection.ip}:`;
    this.shellSocket.connect(address + this.connection.shell_port);
    this.controlSocket.connect(address + this.connection.control_port);
    this.ioSocket.connect(address + this.connection.iopub_port);
    this.ioSocket.subscribe("");
    this.stdinSocket.connect(address + this.connection.stdin_port);

    this.shellSocket.on("message", this.onShellMessage.bind(this));
    this.ioSocket.on("message", this.onIOMessage.bind(this));
    this.stdinSocket.on("message", this.onStdinMessage.bind(this));

    this.shellSocket.on("connect", () => log("shellSocket connected"));
    this.controlSocket.on("connect", () => log("controlSocket connected"));
    this.ioSocket.on("connect", () => log("ioSocket connected"));
    this.stdinSocket.on("connect", () => log("stdinSocket connected"));

    try {
      this.shellSocket.monitor();
      this.controlSocket.monitor();
      this.ioSocket.monitor();
      this.stdinSocket.monitor();
    } catch (err) {
      console.error("Kernel:", err);
    }
  }

  interrupt() {
    if (process.platform === "win32") {
      atom.notifications.addWarning("Cannot interrupt this kernel", {
        detail: "Kernel interruption is currently not supported in Windows."
      });
    } else if (this.kernelProcess) {
      log("ZMQKernel: sending SIGINT");
      this.kernelProcess.kill("SIGINT");
    } else {
      log("ZMQKernel: cannot interrupt an existing kernel");
      atom.notifications.addWarning("Cannot interrupt an existing kernel");
    }
  }

  _kill() {
    if (this.kernelProcess) {
      log("ZMQKernel: sending SIGKILL");
      this.kernelProcess.kill("SIGKILL");
    } else {
      log("ZMQKernel: cannot kill an existing kernel");
      atom.notifications.addWarning("Cannot kill this kernel");
    }
  }
<commit_msg>ZMQKernel: Unmonitor sockets after connection<commit_after>/* @flow */

import fs from "fs";
import { Message, Socket } from "jmp";
import v4 from "uuid/v4";

import Kernel from "./kernel";
import InputView from "./input-view";
import kernelManager from "./kernel-manager";
import { log } from "./utils";

type Connection = {
  control_port: number,
  hb_port: number,
  iopub_port: number,
  ip: string,
  key: string,
  shell_port: number,
  signature_scheme: string,
  stdin_port: number,
  transport: string,
  version: number
};
/* eslint-disable camelcase */
export default class ZMQKernel extends Kernel {
  executionCallbacks: Object = {};
  connection: Connection;
  connectionFile: string;
  kernelProcess: child_process$ChildProcess;

  shellSocket: Socket;
  controlSocket: Socket;
  stdinSocket: Socket;
  ioSocket: Socket;

  constructor(
    kernelSpec: Kernelspec,
    grammar: atom$Grammar,
    connection: Connection,
    connectionFile: string,
    kernelProcess: child_process$ChildProcess
  ) {
    super(kernelSpec, grammar);
    this.connection = connection;
    this.connectionFile = connectionFile;
    this.kernelProcess = kernelProcess;

    this._connect();

    if (this.kernelProcess) {
      log("ZMQKernel: @kernelProcess:", this.kernelProcess);

      this.kernelProcess.stdout.on("data", (data: string | Buffer) => {
        data = data.toString();

        if (atom.config.get("Hydrogen.kernelNotifications")) {
          atom.notifications.addInfo(this.kernelSpec.display_name, {
            description: data,
            dismissable: true
          });
        } else {
          log("ZMQKernel: stdout:", data);
        }
      });

      this.kernelProcess.stderr.on("data", (data: string | Buffer) => {
        atom.notifications.addError(this.kernelSpec.display_name, {
          description: data.toString(),
          dismissable: true
        });
      });
    } else {
      log("ZMQKernel: connectionFile:", this.connectionFile);
      atom.notifications.addInfo("Using an existing kernel connection");
    }
  }

  _connect(done) {
    const scheme = this.connection.signature_scheme.slice("hmac-".length);
    const { key } = this.connection;

    this.shellSocket = new Socket("dealer", scheme, key);
    this.controlSocket = new Socket("dealer", scheme, key);
    this.stdinSocket = new Socket("dealer", scheme, key);
    this.ioSocket = new Socket("sub", scheme, key);

    const id = v4();
    this.shellSocket.identity = `dealer${id}`;
    this.controlSocket.identity = `control${id}`;
    this.stdinSocket.identity = `dealer${id}`;
    this.ioSocket.identity = `sub${id}`;

    const address = `${this.connection.transport}://${this.connection.ip}:`;
    this.shellSocket.connect(address + this.connection.shell_port);
    this.controlSocket.connect(address + this.connection.control_port);
    this.ioSocket.connect(address + this.connection.iopub_port);
    this.ioSocket.subscribe("");
    this.stdinSocket.connect(address + this.connection.stdin_port);

    this.shellSocket.on("message", this.onShellMessage.bind(this));
    this.ioSocket.on("message", this.onIOMessage.bind(this));
    this.stdinSocket.on("message", this.onStdinMessage.bind(this));

    this.shellSocket.on("connect", () => log("shellSocket connected"));
    this.controlSocket.on("connect", () => log("controlSocket connected"));
    this.ioSocket.on("connect", () => log("ioSocket connected"));

    try {
      let socketNames = ["shellSocket", "controlSocket", "ioSocket"];

      let waitGroup = socketNames.length;
      function onConnect() {
        log("ZMQKernel: " + this.socketName + " connected");
        this.socket.unmonitor();

        waitGroup--;
        if (waitGroup === 0) {
          if (done) done();
        }
      }

      socketNames.forEach(socketName => {
        const socket = this[socketName];
        socket.on("connect", onConnect.bind({ socketName, socket }));
        socket.monitor();
      });
    } catch (err) {
      console.error("ZMQKernel:", err);
    }
  }

  interrupt() {
    if (process.platform === "win32") {
      atom.notifications.addWarning("Cannot interrupt this kernel", {
        detail: "Kernel interruption is currently not supported in Windows."
      });
    } else if (this.kernelProcess) {
      log("ZMQKernel: sending SIGINT");
      this.kernelProcess.kill("SIGINT");
    } else {
      log("ZMQKernel: cannot interrupt an existing kernel");
      atom.notifications.addWarning("Cannot interrupt an existing kernel");
    }
  }

  _kill() {
    if (this.kernelProcess) {
      log("ZMQKernel: sending SIGKILL");
      this.kernelProcess.kill("SIGKILL");
    } else {
      log("ZMQKernel: cannot kill an existing kernel");
      atom.notifications.addWarning("Cannot kill this kernel");
    }
  }

{'fraction_non_alphanumeric': 0.14194387257557697, 'fraction_numerical': 0.0011918951132300357, 'mean_word_length': 3.303030303030303, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 15, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5032529', 'n_tokens_mistral': 2755, 'n_tokens_neox': 2699, 'n_words': 567}