<commit_before>'use strict';

import {Observable, Subscription} from '@reactivex/rxjs';
import {IExternalClient} from '../defs';

// No definitions for kafka, sad
let kafka = require('kafka-node');

export interface IConsumerConfig {
  partition?: any;
  offset?: any;
}

export class GustavKafka implements IExternalClient {
  constructor(public config?: any) {
    this.config = this.config || {
      connString: 'localhost:2181',
      clientId: 'kafka-gustav-client'
    };
  }

  getClient(): any {
    return new kafka.Client(this.config.connString, this.config.clientId);
  }

  // two methods, one called on from, other on to
  from(topic: string, offset?: number): Observable<any> {
    let client = this.getClient();

    let consumer = new kafka.Consumer(client, [{
      topic
    }]/* TODO */);

    return new Observable(o => {
      consumer.on('message', m => o.next(m.value));
      consumer.on('error', err => o.error(err));

      return () => consumer.close(() => {});
    });
  }
  to(topic: string, iO: Observable<any>): Subscription<any> {
    let client = this.getClient();
    let producer = new kafka.Producer(client);

    let buffer = [];
    let handleErr = err =>  {
      if (err) { throw err; }
    };

    producer.on('ready', () => {
      if (!buffer.length) { return; }
      producer.send([{
        topic: topic,
        messages: buffer
      }], handleErr);
    });

    return iO
    .bufferTime(50)
    .subscribe(
      msg => {
        if (!msg.length) { return; }

        for (let i = msg.length - 1; i >= 0; i--) {
          msg[i] = typeof msg[i] === 'string' ? msg[i] : JSON.stringify(msg[i]);
        }

        if (!producer.ready) {
          console.log('buffering');
          buffer = buffer.concat(msg);
          return;
        }

        producer.send([{
          topic: topic,
          messages: msg
        }], handleErr);
      },
      handleErr,
      () => {/* noop */}// client.close(console.log.bind(console, 'done', config.topic))
    );
  }
}
<commit_msg>chore: Clean up Kafka coupler<commit_after>'use strict';

import {Observable, Subscription} from '@reactivex/rxjs';
import {IExternalClient} from '../defs';

// No definitions for kafka, sad
let kafka = require('kafka-node');

export interface IConsumerConfig {
  partition?: any;
  offset?: any;
}

export class GustavKafka implements IExternalClient {
  constructor(public config?: any) {
    this.config = this.config || {
      connString: 'localhost:2181',
      clientId: 'kafka-gustav-client'
    };
  }

  getClient(): any {
    return new kafka.Client(this.config.connString, this.config.clientId);
  }

  // two methods, one called on from, other on to
  from(topic: string, offset?: number): Observable<any> {
    let client = this.getClient();

    let consumer = new kafka.Consumer(client, [{
      topic
    }]/* TODO */);

    return new Observable(o => {
      consumer.on('message', m => o.next(m.value));
      consumer.on('error', err => o.error(err));

      return () => consumer.close(() => {});
    });
  }
  to(topic: string, iO: Observable<any>): Subscription<any> {
    let client = this.getClient();
    let producer = new kafka.Producer(client);

    let buffer = [];
    let handleErr = err =>  {
      if (err) { throw err; }
    };

    producer.on('ready', () => {
      if (!buffer.length) { return; }
      producer.send([{
        topic: topic,
        messages: buffer
      }], handleErr);
    });

    return iO
    // Things run faster overall with a little buffering
    .bufferTime(50)
    .subscribe(
      msg => {
        if (!msg.length) { return; }

        for (let i = msg.length - 1; i >= 0; i--) {
          msg[i] = typeof msg[i] === 'string' ? msg[i] : JSON.stringify(msg[i]);
        }

        if (!producer.ready) {
          buffer = buffer.concat(msg);
          return;
        }

        producer.send([{
          topic: topic,
          messages: msg
        }], handleErr);
      },
      handleErr,
      () => {/* noop */}// client.close(console.log.bind(console, 'done', config.topic))
    );
  }
}

{'fraction_non_alphanumeric': 0.16352969934001466, 'fraction_numerical': 0.003911024199462234, 'mean_word_length': 2.3540983606557377, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 29, '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': '3183565', 'n_tokens_mistral': 1314, 'n_tokens_neox': 1287, 'n_words': 311}