import _ from 'lodash';
import { isPipeline } from './Pipeline';

/**
 * Pipes in pipeline can be functions (pipes) or pipeline (multiple pipes connected).
 * When piping stream pipe is needed.
 * @param pipeline
 * @returns {*}
 */
function getPipe(pipeline) { // TODO - better name
  if (isPipeline(pipeline)) {
    return pipeline.pipe;
  } else if (_.isFunction(pipeline)) {
    return pipeline;
  }
  // pipeline = pipelineDescriptor
  return getPipe(pipeline.pipe);
}

function transformOutStream(newStream, stream, pipeDescriptor) {
  const { outTransformer } = pipeDescriptor;
  return outTransformer ? outTransformer(newStream, stream) : newStream;
}

function transformInStream(stream, pipeDescriptor) {
  const { inTransformer } = pipeDescriptor;
  return inTransformer ? inTransformer(stream) : stream;
}

/**
 * Premise.
 * Instead passing next as argument, close is passed. Reason behind it is optimistic approach.
 * More frequent case should be continuing flow, not closing, for that reason,
 * simpler should be to continue then to stop.
 */

export default (stream, currentPipeDescriptor) => {
  return new Promise((resolve, reject) => {
    const currentPipe = getPipe(currentPipeDescriptor);
    const inStream = transformInStream(stream, currentPipeDescriptor);

    const resolvePipe = nextStream =>
      resolve(transformOutStream(nextStream, stream, currentPipeDescriptor));

    let closed = false;
    // Close flow so that newStream doesn't go further.
    const closePipe = closedStream => {
      // Helps handle both async and sync flows.
      closed = true;
      reject(closedStream);
    };

    const newStream = currentPipe(inStream, closePipe);

    if (closed) {
      // Closed with closePipe function (It means close flow already started).
      // It is a bit hidden relation but it simplifies closing pattern.
      // Removes need to return anything when closing.
      return;
    }

    // TODO - invalidate close after stream is piped (either sync or async)
    //  Once stream is piped close SHOULD NOT be called!

    if (_.isUndefined(newStream)) {
      // TODO - newStream === stream?
      //  Trying to preserve as much as possible JS practices
      //  Early return of `undefined` is one of them when nothing is done?
      //  However, returning same object would be more explicit.
      resolvePipe(stream);
    } else if (newStream instanceof Promise) {
      newStream.then(resolvePipe).catch(closePipe);
    } else if (_.isPlainObject(newStream)) {
      resolvePipe(newStream);
    } else {
      console.log('Invalid stream, must be object, undefined or promise.');
      console.log('Stream value: ', newStream);
      console.log('Pipe: ', currentPipe);
      closePipe({ error: 'Invalid stream', stream: newStream});
    }
  });
}

@@ -464,24 +464,504 @@
 e.pipe);%0A%7D%0A%0A
+// TODO - use enhancers for the stream transformation?%0A// Using enhancers for transformation makes pipeline writing more complex%0A// but %22pipe%22 is getting simpler.%0A// Transformers are for now special case and they exists for every stream type.%0A// Transformers should be used as AdHoc solution for reusing pipes more easily.%0Afunction transformInStream(stream, pipeDescriptor) %7B%0A  const %7B inTransformer %7D = pipeDescriptor;%0A  return inTransformer ? inTransformer(stream) : stream;%0A%7D%0A%0A
 function tra
@@ -1148,33 +1148,45 @@
 on transform
-In
+Err
 Stream(
+errStream, 
 stream, pipe
@@ -1201,34 +1201,35 @@
 or) %7B%0A  const %7B 
-in
+err
 Transformer %7D = 
@@ -1245,34 +1245,35 @@
 iptor;%0A  return 
-in
+err
 Transformer ? in
@@ -1262,34 +1262,35 @@
 rrTransformer ? 
-in
+err
 Transformer(stre
@@ -1277,32 +1277,43 @@
  errTransformer(
+errStream, 
 stream) : stream
@@ -1298,33 +1298,36 @@
 ream, stream) : 
-s
+errS
 tream;%0A%7D%0A%0A/**%0A *
@@ -1474,16 +1474,20 @@
 tinuing 
+the 
 flow, no
@@ -1512,17 +1512,16 @@
 t reason
-,
 %0A * simp
@@ -2109,28 +2109,79 @@
  reject(
-closedStream
+transformErrStream(closedStream, stream, currentPipeDescriptor)
 );%0A    %7D

Add support for errTransformers to pipe