const complete = (some$: Stream<Marble>, time: number) => {
  some$.shamefullySendNext({
    time,
    complete: true
  });
  some$.shamefullySendComplete();
}

 private __commitBySha(sha: string) {
   const response$$: Stream<Stream<Response>> = this.http.select(`commit-by-sha-${sha}`);
   return response$$
     .map(response$ => response$.replaceError(() => xs.of({ status: 500, body: {} } as Response)))
     .flatten()
     .map(response => response.body as Commit);
 }

function main(sources: Sources): Sinks {
  const stateUpdate$ = sources.websocket.get('state-update')
    .map(msg => msg.data as BootstrapMessage)
  const board = isolate(Board)(sources, stateUpdate$)

  const boardWebsocket$ = Stream.merge(
    board.moveNote$.map(noteEvent => ({
      type: 'move-note',
      data: noteEvent,
    })),
    board.addNote$.mapTo({ type: 'add-note' }),
    board.editNote$.map(({ id, label }) => ({
      type: 'change-note-label',
      data: { id, label }
    })),
    board.noteDelete$.map(id => ({
      type: 'delete-note',
      data: { id }
    }))
  )

  return {
    DOM: board.DOM,
    websocket: Stream.merge(boardWebsocket$, Stream.of({ type: 'init' }))
      .debug('websocket$'),
    preventDefault: board.preventDefault,
    focus: board.focus,
  }
}

function intent(HTTPSource: HTTPSource): Intent {
  const url = 'http://swapi.co/api/people/'

  const usersReq$: Stream<RequestOptions> = xs.of({ url, category: 'users' })

  const users$: Stream<User[]> = HTTPSource
    .select('users')
    .flatten()
    .map((res: Response<UsersResBody>) => res.body.results)

  const homeworldsReq$: Stream<RequestOptions> = users$.map((users: User[]): Stream<RequestOptions> => {
    return xs.fromArray(users.map((user: User): RequestOptions => {
      return { url: user.homeworld, category: 'homeworld' }
    }))
  }).flatten()

  const homeworlds$ = HTTPSource
    .select('homeworld')
    .compose(flattenConcurrently)
    .map((res: Response<Planet>) => res.body)
    .fold((acc: Planet[], homeworld: Planet) => {
      acc = acc.concat(homeworld)
      return acc
    }, [])

  return { users$, homeworlds$, usersReq$, homeworldsReq$ }
}

 private __commits() {
   const response$$: Stream<Stream<Response>> = this.http.select('commits');
   return response$$
     .map(response$ => response$.replaceError(() => xs.of({ status: 500, body: [] } as Response)))
     .flatten()
     .map(response => response.body as Commit[]);
 }

export const toMarbleStream = (marble$: Stream<Marble>, operate: (data$: Stream<string>) => Stream<string>): Stream<Marble> =>
  Stream.merge<Marble>(
    marble$
      .filter(({ complete }) => !complete)
      .map(({ data, time }) => operate(Stream.of(data)).map<Marble>(data => ({ time, data })))
      .flatten(),
    marble$
      .filter(({ complete }) => !!complete)
  );

 constructor(operator$: Stream<string>) {
   const xs = Stream;
   operator$.addListener(dummyListener);
   this.operator$ =
     operator$
       .map(operator => xs.of(operators[operator]))
       .flatten();
   this.operators$ =
     xs.of(keys(operators)).remember();
 }

function Button(sources: Sources): Sinks {
	let props$: Stream<ButtonProps> = sources.props$;
	const click$ = sources.DOM.select('.button').events('click')
	const delta$ = props$.map(
		(props) => click$.map((ev) => props.amount)
	).flatten();
	const vdom$ = props$.map(props => button('.button', [props.text]));

	return {
		DOM: vdom$,
		delta$: delta$
	}
}

      return function delayByOperator<T>(stream: Stream<T>): Stream<T> {
        return xs.create<T>({
          start(listener) {
            const {schedule, currentTime} = timeSource.createOperator();

            stream.addListener({
              next(t: T) {
                const delay = delaySelector(t);

                schedule.next(listener, currentTime() + delay, t);
              },

              error(err: Error) {
                schedule.error(listener, currentTime(), err);
              },

              complete() {
                schedule.complete(listener, currentTime());
              },
            });
          },

          stop() {},
        });
      };

export function totalIsolateSink(
  sink: Stream<VNode | null | undefined>,
  fullScope: string,
): Stream<VNode | null | undefined> {
  return sink.map(node => {
    if (!node) {
      return node;
    }
    // Ignore if already had up-to-date full scope in vnode.data.isolate
    if (node.data && (node.data as any).isolate) {
      const isolateData = (node.data as any).isolate as string;
      const prevFullScopeNum = isolateData.replace(/(cycle|\-)/g, '');
      const fullScopeNum = fullScope.replace(/(cycle|\-)/g, '');

      if (
        isNaN(parseInt(prevFullScopeNum)) ||
        isNaN(parseInt(fullScopeNum)) ||
        prevFullScopeNum > fullScopeNum
      ) {
        // > is lexicographic string comparison
        return node;
      }
    }

    // Insert up-to-date full scope in vnode.data.isolate, and also a key if needed
    node.data = node.data || {};
    (node.data as any).isolate = fullScope;
    if (typeof node.key === 'undefined') {
      node.key = SCOPE_PREFIX + fullScope;
    }
    return node;
  });
}