<commit_before>import * as jmp from 'jmp';
import * as uuid from 'uuid';

import { Observer, Observable, Subject } from 'rx';

function formConnectionString(config, channel) {
  const portDelimiter = config.transport === 'tcp' ? ':' : '-';
  const port = config[channel + '_port'];
  if (! port) {
    throw new Error(`Port not found for channel "${channel}"`);
  }
  return `${config.transport}://${config.ip}${portDelimiter}${port}`;
}

function sendMessage(socket, messageObject) {
  const message = new jmp.Message(messageObject);
  socket.send(message);
}

function createObserver(jmpSocket) {
  return Observer.create(messageObject => {
    sendMessage(jmpSocket, messageObject);
  }, err => {
    // We don't expect to send errors to the kernel
    console.error(err);
  }, () => {
    jmpSocket.close();
  });
}

function createObservable(jmpSocket) {
  return Observable.fromEvent(jmpSocket, 'message')
                   .publish()
                   .refCount();
}

function createSubject(jmpSocket) {
  const subj = Subject.create(createObserver(jmpSocket),
                              createObservable(jmpSocket));
  subj.send = subj.onNext; // Adapt naming to fit our parlance
  return subj;
}

function enchannelZMQ(config) {
  const scheme = config.signature_scheme.slice('hmac-'.length);

  const shellSocket = new jmp.Socket('dealer', scheme, config.key);
  const controlSocket = new jmp.Socket('dealer', scheme, config.key);
  const ioSocket = new jmp.Socket('sub', scheme, config.key);

  shellSocket.identity = 'shell' + uuid.v4();
  controlSocket.identity = 'control' + uuid.v4();
  ioSocket.identity = 'iopub' + uuid.v4();

  shellSocket.connect(formConnectionString(config, 'shell'));<commit_msg>On subj.onCompleted(), removeAllListeners<commit_after>import * as jmp from 'jmp';
import * as uuid from 'uuid';

import { Observer, Observable, Subject } from 'rx';

function formConnectionString(config, channel) {
  const portDelimiter = config.transport === 'tcp' ? ':' : '-';
  const port = config[channel + '_port'];
  if (! port) {
    throw new Error(`Port not found for channel "${channel}"`);
  }
  return `${config.transport}://${config.ip}${portDelimiter}${port}`;
}

function sendMessage(socket, messageObject) {
  const message = new jmp.Message(messageObject);
  socket.send(message);
}

function createObserver(jmpSocket) {
  return Observer.create(messageObject => {
    sendMessage(jmpSocket, messageObject);
  }, err => {
    // We don't expect to send errors to the kernel
    console.error(err);
  }, () => {
    // tear it down, tear it *all* down
    jmpSocket.removeAllListeners();
    jmpSocket.close();
  });
}

function createObservable(jmpSocket) {
  return Observable.fromEvent(jmpSocket, 'message')
                   .publish()
                   .refCount();
}

function createSubject(jmpSocket) {
  const subj = Subject.create(createObserver(jmpSocket),
                              createObservable(jmpSocket));
  subj.send = subj.onNext; // Adapt naming to fit our parlance
  return subj;
}

function enchannelZMQ(config) {
  const scheme = config.signature_scheme.slice('hmac-'.length);

  const shellSocket = new jmp.Socket('dealer', scheme, config.key);
  const controlSocket = new jmp.Socket('dealer', scheme, config.key);
  const ioSocket = new jmp.Socket('sub', scheme, config.key);

  shellSocket.identity = 'shell' + uuid.v4();
  controlSocket.identity = 'control' + uuid.v4();
  ioSocket.identity = 'iopub' + uuid.v4();

  shellSocket.connect(formConnectionString(config, 'shell'));
{'fraction_non_alphanumeric': 0.145152198421646, 'fraction_numerical': 0.0016910935738444193, 'mean_word_length': 3.848360655737705, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 9, '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': '2039544', 'n_tokens_mistral': 1087, 'n_tokens_neox': 1084, 'n_words': 253}