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

function disposeAllStreams (streams) {
  Object.keys(streams).forEach(key => {
    const value = streams[key];

    delete streams[key];

    value.dispose();
  });
}

function makeDispose ({streams}, originalDispose) {
  return function dispose () {
    originalDispose();
    disposeAllStreams(streams);
  };
}

function onDispose (observable, disposeHandler) {
  const oldDispose = observable.dispose;

  observable.dispose = () => {
    disposeHandler();
    oldDispose();
  };

  return observable;
}

function record ({streams, log, logEntry$}, streamToRecord, identifier) {
  if (streams[identifier] === undefined) {
    streams[identifier] = new ReplaySubject();
  }

  const stream = streams[identifier];

  const subscription = streamToRecord.subscribe(event => {
    log.push({event, time: new Date(), identifier, stream});
    logEntry$.onNext({event, time: new Date(), identifier, stream});

    stream.onNext(event);
  });

  onDispose(stream, () => subscription.dispose());

  return stream;
}

function recordObservableSource ({streams, log, logEntry$}, source) {
  const source$ = new ReplaySubject(1);

  streams[':root'] = source$;

  const subscription = source.subscribe(event => {
    if (typeof event.subscribe === 'function') {
      const loggedEvent$ = event.do(response => {
        log.push({
          event: Object.assign(
            Observable.just(response),
            event
          ),
          time: new Date(),
          identifier: ':root'
        });

        logEntry$.onNext({
          event: Object.assign(
            Observable.just(response),
            event
          ),
          time: new Date(),
          identifier: ':root'
        });
      });

      loggedEvent$.request = event.request;

      source$.onNext(loggedEvent$);
    } else {
      source$.onNext(event);

      log.push({event, time: new Date(), identifier: ':root'});
      logEntry$.onNext({event, time: new Date(), identifier: ':root'});
    }
  });

  onDispose(source$, () => subscription.dispose());

  return source$;
}

function wrapSourceFunction ({streams, log, logEntry$}, name, f, context, scope = []) {
  return function newSource (...args) {
    const newScope = scope.concat(args);

    const returnValue = f.bind(context, ...args)();

    if (name.indexOf('isolate') !== -1 || typeof returnValue !== 'object') {
      return returnValue;
    }

    if (typeof returnValue.subscribe !== 'function') {
      return wrapSource({streams, log, logEntry$}, returnValue, newScope);
    }

    const identifier = newScope.join('/');

    return record({streams, log, logEntry$}, returnValue, identifier);
  };
}

function wrapSource ({streams, log, logEntry$}, source, scope = []) {
  const returnValue = {};

  Object.keys(source).forEach(key => {
    const value = source[key];

    if (key === 'dispose') {
      returnValue[key] = makeDispose({streams}, value);
    } else if (typeof value === 'function') {
      returnValue[key] = wrapSourceFunction({streams, log, logEntry$}, key, value, returnValue, scope);
    } else {
      returnValue[key] = value;
    }
  });

  return returnValue;
}

export default function restartable (driver, opts = {}) {
  const log = [];
  const logEntry$ = new Subject();
  const log$ = logEntry$
    .startWith([])
    .scan((log, entry) => log.concat([entry]))
    .shareReplay(1);

  log$.subscribe();

  const streams = {};

  const pauseSinksWhileReplaying = opts.pauseSinksWhileReplaying === undefined ? true : opts.pauseSinksWhileReplaying;

  let replaying;

  function restartableDriver (sink$) {
    let filteredSink$ = sink$;

    if (sink$ && pauseSinksWhileReplaying) {
      filteredSink$ = sink$.filter(_ => !replaying);
    }

    const source = driver(filteredSink$);

    let returnValue;

    if (source === undefined || source === null) {
      return source;
    } else if (typeof source.subscribe === 'function') {
      returnValue = recordObservableSource({streams, log, logEntry$}, source);
    } else {
      returnValue = wrapSource({streams, log, logEntry$}, source);
    }

    returnValue.log = () => [];
    returnValue.log$ = log$;

    return returnValue;
  }

  function replayable (driver) {
    driver.onPreReplay = function () {
      replaying = true;
    };

    driver.replayLog = function (scheduler, newLog$) {
      newLog$.take(1).subscribe(newLog => {
        function scheduleEvent (historicEvent) {
          scheduler.scheduleAbsolute({}, historicEvent.time, () => {
            streams[historicEvent.identifier].onNext(historicEvent.event);
          });
        }

        newLog.forEach(scheduleEvent);
      });
    };

    driver.onPostReplay = function () {
      replaying = false;
    };

    return restartableDriver;
  }

  return replayable(restartableDriver);
}

@@ -574,37 +574,32 @@
 ecord (%7Bstreams,
- log,
  logEntry$%7D, str
@@ -822,69 +822,8 @@
 %3E %7B%0A
-    log.push(%7Bevent, time: new Date(), identifier, stream%7D);%0A
     
@@ -1027,37 +1027,32 @@
 ource (%7Bstreams,
- log,
  logEntry$%7D, sou
@@ -1284,200 +1284,8 @@
 %3E %7B%0A
-        log.push(%7B%0A          event: Object.assign(%0A            Observable.just(response),%0A            event%0A          ),%0A          time: new Date(),%0A          identifier: ':root'%0A        %7D);%0A%0A
     
@@ -1618,72 +1618,8 @@
 );%0A%0A
-      log.push(%7Bevent, time: new Date(), identifier: ':root'%7D);%0A
     
@@ -1803,37 +1803,32 @@
 ction (%7Bstreams,
- log,
  logEntry$%7D, nam
@@ -2181,37 +2181,32 @@
 Source(%7Bstreams,
- log,
  logEntry$%7D, ret
@@ -2296,37 +2296,32 @@
 record(%7Bstreams,
- log,
  logEntry$%7D, ret
@@ -2373,37 +2373,32 @@
 ource (%7Bstreams,
- log,
  logEntry$%7D, sou
@@ -2690,37 +2690,32 @@
 nction(%7Bstreams,
- log,
  logEntry$%7D, key
@@ -2889,26 +2889,8 @@
 ) %7B%0A
-  const log = %5B%5D;%0A
   co
@@ -3629,37 +3629,32 @@
 Source(%7Bstreams,
- log,
  logEntry$%7D, sou
@@ -3716,13 +3716,8 @@
 ams,
- log,
  log
@@ -3745,40 +3745,8 @@
  %7D%0A%0A
-    returnValue.log = () =%3E %5B%5D;%0A
     

Remove dead log code