<commit_before>'use strict';

import { isError } from '../utilites';
import { arrayEmitter } from '../emitters/array';
import { toArrayOperator } from './toArray';

export function joinOperator(flow, predicate) {

  return emitter => (next, done, context) => {
  let source, array = [];
  toArrayOperator()(emitter)(
      result => {
        source = result;
        return false;
      },
      result => {        
        if (isError(result)) done(result);
        else {
          flow.run(value => {
              array.push(source
                .filter(x => !predicate || predicate(x, value))
                .map(x => {
                  return [x, value];
                }));
              }, 
            () => { 
              arrayEmitter(array)(next, done, context) 
            }
          );
        }
      },
      context
    );
  }
}<commit_msg>Add adapterEmitter<commit_after>'use strict';

import { isError } from '../utilites';
import { arrayEmitter } from '../emitters/array';
import { adapterEmitter } from '../emitters/adapter';
import { toArrayOperator } from './toArray';

export function joinOperator(right, predicate) {
  let array = [], source;

  return emitter => (next, done, context) => {
  let source, array = [];
  var rightEmitter = adapterEmitter(right, true);
  toArrayOperator()(rightEmitter)(
      result => {
        source = result;
        return false;
      },
      result => {        
        if (isError(result)) done(result);
        else {          
          context.flow.run(value => {
              source
                .filter(x => !predicate || predicate(x, value))
                .map(x => array.push([x, value]));
              }, 
            () => { 
              arrayEmitter(array)(next, done, context) 
            }
          );
        }
      },
      context
    );
  }
}
{'fraction_non_alphanumeric': 0.14579945799457994, 'fraction_numerical': 0.0, 'mean_word_length': 1.6522988505747127, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 19, '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': '4705696', 'n_tokens_mistral': 512, 'n_tokens_neox': 499, 'n_words': 113}