<commit_before>// connect :: Streams... -> Stream
// connect function takes one or more streams
// and sequentially pipes them to each other,
// returning the result of the last pipe operation.
//
// `ReadableStream::pipeThrough` is used to
// connect the streams.

export default function connect(origin, ...streams) {
  // Check for transform streams
  let
    pipeThrough,
    end = origin.readable || origin;

  for ( let stream of streams ) {

    // Null stream
    if ( !end )
      return end;

    // Extract piping function
    if ( end.pipeThrough ) {
      pipeThrough = end.pipeThrough.bind(end);

    } else if ( end.readable ) {
      pipeThrough = end.readable.pipeThrough.bind(end.readable);

    } else {
      throw new Error("Only readable and transform streams can be chained");

    }

    // Pipe streams into each other
    // If transform stream then pipe readable
    end = pipeThrough( stream );
  }

  return end;
}<commit_msg>Fixed connect<commit_after>// connect :: Streams... -> ReadableStream | Promise
// connect function takes one or more streams
// and sequentially pipes them to each other,
// returning the result of the last pipe operation.
//

// Utils
const
  isTransform = s => s && s.writable && s.readable,
  isWritable = s => s && s.write;

export default function connect(origin, ...streams) {
  // Check origin
  if ( !origin )
    throw new Error("No streams passed");

  let sink, end;

  // Get the last stream
  sink = streams.pop();

  // if origin is a transform$, take it's readable part
  end = origin.readable || origin;

  // Connect the streams
  for ( let stream of streams ) {

    // Check for transform streams
    if ( !isTransform( stream ))
      throw new Error("Only transform streams allowed in the center");

    // piping through a transform returns it's readable part
    end = end.pipeThrough( stream );
  }

  // Handle sink
  if ( isWritable( sink ))
    end = end.pipeTo( sink );

  else if ( isTransform( sink ))
    end = end.pipeThrough( sink );

  else
    throw new Error("Only writable and transform streams allowed at the end.");

  // Return result
  return end;
}

{'fraction_non_alphanumeric': 0.10852352119236143, 'fraction_numerical': 0.0, 'mean_word_length': 2.6780821917808217, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 7, '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': '2076276', 'n_tokens_mistral': 628, 'n_tokens_neox': 582, 'n_words': 237}