" https://github.com/prabirshrestha/callbag.vim#82f96a7d97342fbf0286e6578b65a60f2bc1ce33
"    :CallbagEmbed path=autoload/lsp/callbag.vim namespace=lsp#callbag

let s:undefined_token = '__callbag_undefined__'
let s:str_type = type('')

function! lsp#callbag#undefined() abort
    return s:undefined_token
endfunction

function! lsp#callbag#isUndefined(d) abort
    return type(a:d) == s:str_type && a:d ==# s:undefined_token
endfunction

function! s:noop(...) abort
endfunction

function! s:createArrayWithSize(size, defaultValue) abort
    let l:i = 0
    let l:array = []
    while l:i < a:size
        call add(l:array, a:defaultValue)
        let l:i = l:i + 1
    endwhile
    return l:array
endfunction

" pipe() {{{
function! lsp#callbag#pipe(...) abort
    let l:Res = a:1
    let l:i = 1
    while l:i < a:0
        let l:Res = a:000[l:i](l:Res)
        let l:i = l:i + 1
    endwhile
    return l:Res
endfunction
" }}}

" operate() {{{
function! lsp#callbag#operate(...) abort
    let l:data = { 'cbs': a:000 }
    return function('s:operateFactory', [l:data])
endfunction

function! s:operateFactory(data, src) abort
    let l:Res = a:src
    let l:n = len(a:data['cbs'])
    let l:i = 0
    while l:i < l:n
        let l:Res = a:data['cbs'][l:i](l:Res)
        let l:i = l:i + 1
    endwhile
    return l:Res
endfunction
" }}}

" makeSubject() {{{
function! lsp#callbag#makeSubject() abort
    let l:data = { 'sinks': [] }
    return function('s:makeSubjectFactory', [l:data])
endfunction

function! s:makeSubjectFactory(data, t, d) abort
    if a:t == 0
        let l:Sink = a:d
        call add(a:data['sinks'], l:Sink)
        call l:Sink(0, function('s:makeSubjectSinkCallback', [a:data, l:Sink]))
    else
        let l:zinkz = copy(a:data['sinks'])
        let l:i = 0
        let l:n = len(l:zinkz)
        while l:i < l:n
            let l:Sink = l:zinkz[l:i]
            let l:j = -1
            let l:found = 0
            for l:Item in a:data['sinks']
                let l:j += 1
                if l:Item == l:Sink
                    let l:found = 1
                    break
                endif
            endfor

            if l:found
                call l:Sink(a:t, a:d)
            endif
            let l:i += 1
        endwhile
    endif
endfunction

function! s:makeSubjectSinkCallback(data, Sink, t, d) abort
    if a:t == 2
        let l:i = -1
        let l:found = 0
        for l:Item in a:data['sinks']
            let l:i += 1
            if l:Item == a:Sink
                let l:found = 1
                break
            endif
        endfor
        if l:found
            call remove(a:data['sinks'], l:i)
        endif
    endif
endfunction
" }}}

" create() {{{
function! lsp#callbag#create(...) abort
    let l:data = {}
    if a:0 > 0
        let l:data['prod'] = a:1
    endif
    return function('s:createProd', [l:data])
endfunction

function! s:createProd(data, start, sink) abort
    if a:start != 0 | return | endif
    let a:data['sink'] = a:sink
    if !has_key(a:data, 'prod') || type(a:data['prod']) != type(function('s:noop'))
        call a:sink(0, function('s:noop'))
        call a:sink(2, lsp#callbag#undefined())
        return
    endif
    let a:data['end'] = 0
    call a:sink(0, function('s:createSinkCallback', [a:data]))
    if a:data['end'] | return | endif
    let a:data['clean'] = a:data['prod'](function('s:createNext', [a:data]), function('s:createError', [a:data]), function('s:createComplete', [a:data]))
endfunction

function! s:createSinkCallback(data, t, ...) abort
    if !a:data['end']
        let a:data['end'] = (a:t == 2)
        if a:data['end'] && has_key(a:data, 'clean') && type(a:data['clean']) == type(function('s:noop'))
            call a:data['clean']()
        endif
    endif
endfunction

function! s:createNext(data, d) abort
    if !a:data['end'] | call a:data['sink'](1, a:d) | endif
endfunction

function! s:createError(data, e) abort
    if !a:data['end'] && !lsp#callbag#isUndefined(a:e)
        let a:data['end'] = 1
        call a:data['sink'](2, a:e)
    endif
endfunction

function! s:createComplete(data) abort
    if !a:data['end']
        let a:data['end'] = 1
        call a:data['sink'](2, lsp#callbag#undefined())
    endif
endfunction
" }}}

" lazy() {{{
function! lsp#callbag#lazy(F) abort
    let l:data = { 'F': a:F }
    return function('s:lazyFactory', [l:data])
endfunction

function! s:lazyFactory(data, start, sink) abort
    if a:start != 0 | return | endif
    let a:data['sink'] = a:sink
    let a:data['unsubed'] = 0
    call a:data['sink'](0, function('s:lazySinkCallback', [a:data]))
    call a:data['sink'](1, a:data['F']())
    if !a:data['unsubed'] | call a:data['sink'](2, lsp#callbag#undefined()) | endif
endfunction

function! s:lazySinkCallback(data, t, d) abort
    if a:t == 2 | let a:data['unsubed'] = 1 | endif
endfunction
" }}}

" empty() {{{
function! lsp#callbag#empty() abort
    let l:data = {}
    return function('s:emptyStart', [l:data])
endfunction

function! s:emptyStart(data, start, sink) abort
    if a:start != 0 | return | endif
    let a:data['disposed'] = 0
    call a:sink(0, function('s:emptySinkCallback', [a:data]))
    if a:data['disposed'] | return | endif
    call a:sink(2, lsp#callbag#undefined())
endfunction

function! s:emptySinkCallback(data, t, ...) abort
    if a:t != 2 | return | endif
    let a:data['disposed'] = 1
endfunction

function! s:empty_sink_callback(data, t, ...) abort
    if a:t == 2 | call timer_stop(a:data['timer']) | endif
endfunction
" }}}

" never() {{{
function! lsp#callbag#never() abort
    return function('s:never')
endfunction

function! s:never(start, sink) abort
    if a:start != 0 | return | endif
    call a:sink(0, function('s:noop'))
endfunction
" }}}

" forEach() {{{
function! lsp#callbag#forEach(operation) abort
    let l:data = { 'operation': a:operation }
    return function('s:forEachOperation', [l:data])
endfunction

function! s:forEachOperation(data, source) abort
    return a:source(0, function('s:forEachOperationSource', [a:data]))
endfunction

function! s:forEachOperationSource(data, t, d) abort
    if a:t == 0 | let a:data['talkback'] = a:d | endif
    if a:t == 1 | call a:data['operation'](a:d) | endif
    if (a:t == 1 || a:t == 0) | call a:data['talkback'](1, lsp#callbag#undefined()) | endif
endfunction
" }}}

" tap() {{{
function! lsp#callbag#tap(...) abort
    let l:data = {}
    if a:0 > 0 && type(a:1) == type({}) " a:1 { next, error, complete }
        if has_key(a:1, 'next') | let l:data['next'] = a:1['next'] | endif
        if has_key(a:1, 'error') | let l:data['error'] = a:1['error'] | endif
        if has_key(a:1, 'complete') | let l:data['complete'] = a:1['complete'] | endif
    else " a:1 = next, a:2 = error, a:3 = complete
        if a:0 >= 1 | let l:data['next'] = a:1 | endif
        if a:0 >= 2 | let l:data['error'] = a:2 | endif
        if a:0 >= 3 | let l:data['complete'] = a:3 | endif
    endif
    return function('s:tapFactory', [l:data])
endfunction

function! s:tapFactory(data, source) abort
    let a:data['source'] = a:source
    return function('s:tapSouceFactory', [a:data])
endfunction

function! s:tapSouceFactory(data, start, sink) abort
    if a:start != 0 | return | endif
    let a:data['sink'] = a:sink
    call a:data['source'](0, function('s:tapSourceCallback', [a:data]))
endfunction

function! s:tapSourceCallback(data, t, d) abort
    if a:t == 1 && has_key(a:data, 'next') | call a:data['next'](a:d) | endif
    if a:t == 2 && lsp#callbag#isUndefined(a:d) && has_key(a:data, 'complete') | call a:data['complete']() | endif
    if a:t == 2 && !lsp#callbag#isUndefined(a:d) && has_key(a:data, 'error') | call a:data['error'](a:d) | endif
    call a:data['sink'](a:t, a:d)
endfunction
" }}}

" interval() {{{
function! lsp#callbag#interval(period) abort
    let l:data = { 'period': a:period }
    return function('s:intervalPeriod', [l:data])
endfunction

function! s:intervalPeriod(data, start, sink) abort
    if a:start != 0 | return | endif
    let a:data['i'] = 0
    let a:data['sink'] = a:sink
    let a:data['timer'] = timer_start(a:data['period'], function('s:interval_callback', [a:data]), { 'repeat': -1 })
    call a:sink(0, function('s:interval_sink_callback', [a:data]))
endfunction

function! s:interval_callback(data, ...) abort
    let l:i = a:data['i']
    let a:data['i'] = a:data['i'] + 1
    call a:data['sink'](1, l:i)
endfunction

function! s:interval_sink_callback(data, t, ...) abort
    if a:t == 2 | call timer_stop(a:data['timer']) | endif
endfunction
" }}}

" delay() {{{
function! lsp#callbag#delay(period) abort
    let l:data = { 'period': a:period }
    return function('s:delayPeriod', [l:data])
endfunction

function! s:delayPeriod(data, source) abort
    let a:data['source'] = a:source
    return function('s:delayFactory', [a:data])
endfunction

function! s:delayFactory(data, start, sink) abort
    if a:start != 0 | return | endif
    let a:data['sink'] = a:sink
    call a:data['source'](0, function('s:delaySourceCallback', [a:data]))
endfunction

function! s:delaySourceCallback(data, t, d) abort
    if a:t != 1
        call a:data['sink'](a:t, a:d)
        return
    endif
    let a:data['d'] = a:d
    call timer_start(a:data['period'], function('s:delayTimerCallback', [a:data]))
endfunction

function! s:delayTimerCallback(data, ...) abort
    call a:data['sink'](1, a:data['d'])
endfunction
" }}}

" take() {{{
function! lsp#callbag#take(max) abort
    let l:data = { 'max': a:max }
    return function('s:takeMax', [l:data])
endfunction

function! s:takeMax(data, source) abort
    let a:data['source'] = a:source
    return function('s:takeMaxSource', [a:data])
endfunction

function! s:takeMaxSource(data, start, sink) abort
    if a:start != 0 | return | endif
    let a:data['taken'] = 0
    let a:data['end'] = 0
    let a:data['sink'] = a:sink
    let a:data['talkback'] = function('s:takeTalkback', [a:data])
    call a:data['source'](0, function('s:takeSourceCallback', [a:data]))
endfunction

function! s:takeTalkback(data, t, d) abort
    if a:t == 2
        let a:data['end'] = 1
        call a:data['sourceTalkback'](a:t, a:d)
    elseif a:data['taken'] < a:data['max']
        call a:data['sourceTalkback'](a:t, a:d)
    endif
endfunction

function! s:takeSourceCallback(data, t, d) abort
    if a:t == 0
        let a:data['sourceTalkback'] = a:d
        call a:data['sink'](0, a:data['talkback'])
    elseif a:t == 1
        if a:data['taken'] < a:data['max']
            let a:data['taken'] = a:data['taken'] + 1
            call a:data['sink'](a:t, a:d)
            if a:data['taken'] == a:data['max'] && !a:data['end']
                let a:data['end'] = 1
                call a:data['sink'](2, lsp#callbag#undefined())
                call a:data['sourceTalkback'](2, lsp#callbag#undefined())
            endif
        endif
    else
        call a:data['sink'](a:t, a:d)
    endif
endfunction
" }}}

" skip() {{{
function! lsp#callbag#skip(max) abort
    let l:data = { 'max': a:max }
    return function('s:skipMax', [l:data])
endfunction

function! s:skipMax(data, source) abort
    let a:data['source'] = a:source
    return function('s:skipMaxSource', [a:data])
endfunction

function! s:skipMaxSource(data, start, sink) abort
    if a:start != 0 | return | endif
    let a:data['sink'] = a:sink
    let a:data['skipped'] = 0
    call a:data['source'](0, function('s:skipSouceCallback', [a:data]))
endfunction

function! s:skipSouceCallback(data, t, d) abort
    if a:t == 0
        let a:data['talkback'] = a:d
        call a:data['sink'](a:t, a:d)
    elseif a:t == 1
        if a:data['skipped'] < a:data['max']
            let a:data['skipped'] = a:data['skipped'] + 1
            call a:data['talkback'](1, lsp#callbag#undefined())
        else
            call a:data['sink'](a:t, a:d)
        endif
    else
        call a:data['sink'](a:t, a:d)
    endif
endfunction
" }}}

" map() {{{
function! lsp#callbag#map(F) abort
    let l:data = { 'f': a:F }
    return function('s:mapF', [l:data])
endfunction

function! s:mapF(data, source) abort
    let a:data['source'] = a:source
    return function('s:mapFSource', [a:data])
endfunction

function! s:mapFSource(data, start, sink) abort
    if a:start != 0 | return | endif
    let a:data['sink'] = a:sink
    call a:data['source'](0, function('s:mapFSourceCallback', [a:data]))
endfunction

function! s:mapFSourceCallback(data, t, d) abort
    call a:data['sink'](a:t, a:t == 1 ? a:data['f'](a:d) : a:d)
endfunction
" }}}

" filter() {{{
function! lsp#callbag#filter(condition) abort
    let l:data = { 'condition': a:condition }
    return function('s:filterCondition', [l:data])
endfunction

function! s:filterCondition(data, source) abort
    let a:data['source'] = a:source
    return function('s:filterConditionSource', [a:data])
endfunction

function! s:filterConditionSource(data, start, sink) abort
    if a:start != 0 | return | endif
    let a:data['sink'] = a:sink
    call a:data['source'](0, function('s:filterSourceCallback', [a:data]))
endfunction

function! s:filterSourceCallback(data, t, d) abort
    if a:t == 0
        let a:data['talkback'] = a:d
        call a:data['sink'](a:t, a:d)
    elseif a:t == 1
        if a:data['condition'](a:d)
            call a:data['sink'](a:t, a:d)
        else
            call a:data['talkback'](1, lsp#callbag#undefined())
        endif
    else
        call a:data['sink'](a:t, a:d)
    endif
endfunction
" }}}

" fromEvent() {{{
let s:event_prefix_index = 0
function! lsp#callbag#fromEvent(events, ...) abort
    let l:data = { 'events': a:events }
    if a:0 > 0
        let l:data['augroup'] = a:1
    else
        let l:data['augroup'] = '__callbag_fromEvent_prefix_' . s:event_prefix_index . '__'
        let s:event_prefix_index = s:event_prefix_index + 1
    endif
    return function('s:fromEventFactory', [l:data])
endfunction

let s:event_handler_index = 0
let s:event_handlers_data = {}
function! s:fromEventFactory(data, start, sink) abort
    if a:start != 0 | return | endif
    let a:data['sink']  = a:sink
    let a:data['disposed'] = 0
    let a:data['handler'] = function('s:fromEventHandlerCallback', [a:data])
    let a:data['handler_index'] = s:event_handler_index
    let s:event_handler_index = s:event_handler_index + 1
    call a:sink(0, function('s:fromEventSinkHandler', [a:data]))

    if a:data['disposed'] | return | endif
    let s:event_handlers_data[a:data['handler_index']] = a:data

    execute 'augroup ' . a:data['augroup']
    execute 'autocmd!'
    let l:events = type(a:data['events']) == type('') ? [a:data['events']] : a:data['events']
    for l:event in l:events
        let l:exec =  'call s:notify_event_handler(' . a:data['handler_index'] . ')'
        if type(l:event) == type('')
            execute 'au ' . l:event . ' * ' . l:exec
        else
            execute 'au ' . join(l:event, ' ') .' ' .  l:exec
        endif
    endfor
    execute 'augroup end'
endfunction

function! s:fromEventHandlerCallback(data) abort
    " send v:event if it exists
    call a:data['sink'](1, lsp#callbag#undefined())
endfunction

function! s:fromEventSinkHandler(data, t, ...) abort
    if a:t != 2 | return | endif
    let a:data['disposed'] = 1
    execute 'augroup ' a:data['augroup']
    autocmd!
    execute 'augroup end'
    if has_key(s:event_handlers_data, a:data['handler_index'])
        call remove(s:event_handlers_data, a:data['handler_index'])
    endif
endfunction

function! s:notify_event_handler(index) abort
    let l:data = s:event_handlers_data[a:index]
    call l:data['handler']()
endfunction
" }}}

" fromPromise() {{{
function! lsp#callbag#fromPromise(promise) abort
    let l:data = { 'promise': a:promise }
    return function('s:fromPromiseFactory', [l:data])
endfunction

function! s:fromPromiseFactory(data, start, sink) abort
    if a:start != 0 | return | endif
    let a:data['sink'] = a:sink
    let a:data['ended'] = 0
    call a:data['promise'].then(
        \ function('s:fromPromiseOnFulfilledCallback', [a:data]),
        \ function('s:fromPromiseOnRejectedCallback', [a:data]),
        \ )
    call a:sink(0, function('s:fromPromiseSinkCallback', [a:data]))
endfunction

function! s:fromPromiseOnFulfilledCallback(data, ...) abort
    if a:data['ended'] | return | endif
    call a:data['sink'](1, a:0 > 0 ? a:1 : lsp#callbag#undefined())
    if a:data['ended'] | return | endif
    call a:data['sink'](2, lsp#callbag#undefined())
endfunction

function! s:fromPromiseOnRejectedCallback(data, err) abort
    if a:data['ended'] | return | endif
    call a:data['sink'](2, a:err)
endfunction

function! s:fromPromiseSinkCallback(data, t, ...) abort
    if a:t == 2 | let a:data['ended'] = 1 | endif
endfunction
" }}}

" debounceTime() {{{
function! lsp#callbag#debounceTime(duration) abort
    let l:data = { 'duration': a:duration }
    return function('s:debounceTimeDuration', [l:data])
endfunction

function! s:debounceTimeDuration(data, source) abort
    let a:data['source'] = a:source
    return function('s:debounceTimeDurationSource', [a:data])
endfunction

function! s:debounceTimeDurationSource(data, start, sink) abort
    if a:start != 0 | return | endif
    let a:data['sink'] = a:sink
    call a:data['source'](0, function('s:debounceTimeSourceCallback', [a:data]))
endfunction

function! s:debounceTimeSourceCallback(data, t, d) abort
    if has_key(a:data, 'timer') | call timer_stop(a:data['timer']) | endif
    if a:t == 1
        let a:data['timer'] = timer_start(a:data['duration'], function('s:debounceTimeTimerCallback', [a:data, a:d]))
    else
        call a:data['sink'](a:t, a:d)
    endif
endfunction

function! s:debounceTimeTimerCallback(data, d, ...) abort
    call a:data['sink'](1, a:d)
endfunction
" }}}

" subscribe() {{{
function! lsp#callbag#subscribe(...) abort
    let l:data = {}
    if a:0 > 0 && type(a:1) == type({}) " a:1 { next, error, complete }
        if has_key(a:1, 'next') | let l:data['next'] = a:1['next'] | endif
        if has_key(a:1, 'error') | let l:data['error'] = a:1['error'] | endif
        if has_key(a:1, 'complete') | let l:data['complete'] = a:1['complete'] | endif
    else " a:1 = next, a:2 = error, a:3 = complete
        if a:0 >= 1 | let l:data['next'] = a:1 | endif
        if a:0 >= 2 | let l:data['error'] = a:2 | endif
        if a:0 >= 3 | let l:data['complete'] = a:3 | endif
    endif
    return function('s:subscribeListener', [l:data])
endfunction

function! s:subscribeListener(data, source) abort
    call a:source(0, function('s:subscribeSourceCallback', [a:data]))
    return function('s:subscribeDispose', [a:data])
endfunction

function! s:subscribeSourceCallback(data, t, d) abort
    if a:t == 0 | let a:data['talkback'] = a:d | endif
    if a:t == 1 && has_key(a:data, 'next') | call a:data['next'](a:d) | endif
    if a:t == 1 || a:t == 0 | call a:data['talkback'](1, lsp#callbag#undefined()) | endif
    if a:t == 2 && lsp#callbag#isUndefined(a:d) && has_key(a:data, 'complete') | call a:data['complete']() | endif
    if a:t == 2 && !lsp#callbag#isUndefined(a:d) && has_key(a:data, 'error') | call a:data['error'](a:d) | endif
endfunction

function! s:subscribeDispose(data, ...) abort
    if has_key(a:data, 'talkback') | call a:data['talkback'](2, lsp#callbag#undefined()) | endif
endfunction
" }}}

" toList() {{{
function! lsp#callbag#toList() abort
    let l:data = { 'done': 0, 'items': [], 'unsubscribed': 0 }
    return function('s:toListFactory', [l:data])
endfunction

function! s:toListFactory(data, source) abort
    let a:data['unsubscribe'] = lsp#callbag#subscribe(
        \ function('s:toListOnNext', [a:data]),
        \ function('s:toListOnError', [a:data]),
        \ function('s:toListOnComplete', [a:data])
        \ )(a:source)
    if a:data['done'] | call s:toListUnsubscribe(a:data) | endif
    return {
        \ 'unsubscribe': function('s:toListUnsubscribe', [a:data]),
        \ 'wait': function('s:toListWait', [a:data])
        \ }
endfunction

function! s:toListUnsubscribe(data) abort
    if !has_key(a:data, 'unsubscribe') | return | endif
    if !a:data['unsubscribed']
        call a:data['unsubscribe']()
        let a:data['unsubscribed'] = 1
        if !a:data['done']
            let a:data['done'] = 1
            try
                throw 'lsp#callbag toList() is already unsubscribed.'
            catch
                let a:data['error'] = v:exception . ' ' . v:throwpoint
            endtry
        endif
    endif
endfunction

function! s:toListOnNext(data, item) abort
    call add(a:data['items'], a:item)
endfunction

function! s:toListOnError(data, error) abort
    let a:data['done'] = 1
    let a:data['error'] = a:error
    call s:toListUnsubscribe(a:data)
endfunction

function! s:toListOnComplete(data) abort
    let a:data['done'] = 1
    call s:toListUnsubscribe(a:data)
endfunction

function! s:toListWait(data, ...) abort
    if a:data['done']
        if has_key(a:data, 'error')
            throw a:data['error']
        else
            return a:data['items']
        endif
    else
        let l:opt = a:0 > 0 ? copy(a:1) : {}
        let l:opt['timedout'] = 0
        let l:opt['sleep'] = get(l:opt, 'sleep', 1)
        let l:opt['timeout'] = get(l:opt, 'timeout', -1)

        if l:opt['timeout'] > -1
            let l:opt['timer'] = timer_start(l:opt['timeout'], function('s:toListTimeoutCallback', [l:opt]))
        endif

        while !a:data['done'] && !l:opt['timedout']
            exec 'sleep ' . l:opt['sleep'] . 'm'
        endwhile

        if has_key(l:opt, 'timer')
            silent! call timer_stop(l:opt['timer'])
        endif

        if l:opt['timedout']
            throw 'lsp#callbag toList().wait() timedout.'
        endif

        if has_key(a:data, 'error')
            throw a:data['error']
        else
            return a:data['items']
        endif
    endif
endfunction

function! s:toListTimeoutCallback(opt, ...) abort
    let a:opt['timedout'] = 1
endfunction
" }}}

" throwError() {{{
function! lsp#callbag#throwError(error) abort
    let l:data = { 'error': a:error }
    return function('s:throwErrorFactory', [l:data])
endfunction

function! s:throwErrorFactory(data, start, sink) abort
    if a:start != 0 | return | endif
    let a:data['disposed'] = 0
    call a:sink(0, function('s:throwErrorSinkCallback', [a:data]))
    if a:data['disposed'] | return | endif
    call a:sink(2, a:data['error'])
endfunction

function! s:throwErrorSinkCallback(data, t, ...) abort
    if a:t != 2 | return | endif
    let a:data['disposed'] = 1
endfunction
" }}}

" of() {{{
function! lsp#callbag#of(...) abort
    let l:data = { 'values': a:000 }
    return function('s:listFactory', [l:data])
endfunction
" }}}

" fromList() {{{
function! lsp#callbag#fromList(list) abort
    let l:data = { 'values': a:list }
    return function('s:listFactory', [l:data])
endfunction

function! s:listFactory(data, start, sink) abort
    if a:start != 0 | return | endif
    let a:data['disposed'] = 0
    call a:sink(0, function('s:listSinkCallback', [a:data]))
    let l:i = 0
    let l:n = len(a:data['values'])
    while l:i < l:n
        if a:data['disposed'] | break | endif
        call a:sink(1, a:data['values'][l:i])
        let l:i += 1
    endwhile
    if a:data['disposed'] | return | endif
    call a:sink(2, lsp#callbag#undefined())
endfunction


function! s:listSinkCallback(data, t, ...) abort
    if a:t != 2 | return | endif
    let a:data['disposed'] = 1
endfunction
" }}}

" merge() {{{
function! lsp#callbag#merge(...) abort
    let l:data = { 'sources': a:000 }
    return function('s:mergeFactory', [l:data])
endfunction

function! s:mergeFactory(data, start, sink) abort
    if a:start != 0 | return | endif
    let a:data['sink'] = a:sink
    let a:data['n'] = len(a:data['sources'])
    let a:data['sourceTalkbacks'] = []
    let a:data['startCount'] = 0
    let a:data['endCount'] = 0
    let a:data['ended'] = 0
    let a:data['talkback'] = function('s:mergeTalkbackCallback', [a:data])
    let l:i = 0
    while l:i < a:data['n']
        if a:data['ended'] | return | endif
        call a:data['sources'][l:i](0, function('s:mergeSourceCallback', [a:data, l:i]))
        let l:i += 1
    endwhile
endfunction

function! s:mergeTalkbackCallback(data, t, d) abort
    if a:t == 2 | let a:data['ended'] = 1 | endif
    let l:i = 0
    while l:i < a:data['n']
        if l:i < len(a:data['sourceTalkbacks']) && a:data['sourceTalkbacks'][l:i] != 0
            call a:data['sourceTalkbacks'][l:i](a:t, a:d)
        endif
        let l:i += 1
    endwhile
endfunction

function! s:mergeSourceCallback(data, i, t, d) abort
    if a:t == 0
        call insert(a:data['sourceTalkbacks'], a:d, a:i)
        let a:data['startCount'] += 1
        if a:data['startCount'] == 1 | call a:data['sink'](0, a:data['talkback']) | endif
    elseif a:t == 2 && !lsp#callbag#isUndefined(a:d)
        let a:data['ended'] = 1
        let l:j = 0
        while l:j < a:data['n']
            if l:j != a:i && l:j < len(a:data['sourceTalkbacks']) && a:data['sourceTalkbacks'][l:j] != 0
                call a:data['sourceTalkbacks'][l:j](2, lsp#callbag#undefined())
            endif
            let l:j += 1
        endwhile
        call a:data['sink'](2, a:d)
    elseif a:t == 2
        let a:data['sourceTalkbacks'][a:i] = 0
        let a:data['endCount'] += 1
        if a:data['endCount'] == a:data['n'] | call a:data['sink'](2, lsp#callbag#undefined()) | endif
    else
        call a:data['sink'](a:t, a:d)
    endif
endfunction
" }}}

" concat() {{{
function! lsp#callbag#concat(...) abort
    let l:data = { 'sources': a:000 }
    return function('s:concatFactory', [l:data])
endfunction

let s:concatUniqueToken = '__callback__concat_unique_token__'
function! s:concatFactory(data, start, sink) abort
    if a:start != 0 | return | endif
    let a:data['sink'] = a:sink
    let a:data['n'] = len(a:data['sources'])
    if a:data['n'] == 0
        call a:data['sink'](0, function('s:noop'))
        call a:data['sink'](2, lsp#callbag#undefined())
        return
    endif
    let a:data['i'] = 0
    let a:data['lastPull'] = s:concatUniqueToken
    let a:data['talkback'] = function('s:concatTalkbackCallback', [a:data])
    let a:data['next'] = function('s:concatNext', [a:data])
    call a:data['next']()
endfunction

function! s:concatTalkbackCallback(data, t, d) abort
    if a:t == 1 | let a:data['lastPull'] = a:d | endif
    call a:data['sourceTalkback'](a:t, a:d)
endfunction

function! s:concatNext(data) abort
    if a:data['i'] == a:data['n']
        call a:data['sink'](2, lsp#callbag#undefined())
        return
    endif
    call a:data['sources'][a:data['i']](0, function('s:concatSourceCallback', [a:data]))
endfunction

function! s:concatSourceCallback(data, t, d) abort
    if a:t == 0
        let a:data['sourceTalkback'] = a:d
        if a:data['i'] == 0
            call a:data['sink'](0, a:data['talkback'])
        elseif (a:data['lastPull']) != s:concatUniqueToken
            call a:data['sourceTalkback'](1, a:data['lastPull'])
        endif
    elseif a:t == 2 && a:d != lsp#callbag#undefined()
        call a:data['sink'](2, a:d)
    elseif a:t == 2 
        let a:data['i'] = a:data['i'] + 1
        call a:data['next']()
    else
        call a:data['sink'](a:t, a:d)
    endif
endfunction
" }}}

" combine() {{{
function! lsp#callbag#combine(...) abort
    let l:data = { 'sources': a:000 }
    return function('s:combineFactory', [l:data])
endfunction

let s:combineEmptyToken = '__callback__combine_empty_token__'
function! s:combineFactory(data, start, sink) abort
    if a:start != 0 | return | endif
    let a:data['sink'] = a:sink
    let a:data['n'] = len(a:data['sources'])
    if a:data['n'] == 0
        call a:data['sink'](0, function('s:noop'))
        call a:data['sink'](1, [])
        call a:data['sink'](2, lsp#callbag#undefined())
        return
    endif
    let a:data['Ns'] = a:data['n'] " start counter
    let a:data['Nd'] = a:data['n'] " data counter
    let a:data['Ne'] = a:data['n'] " end counter
    let a:data['vals'] = s:createArrayWithSize(a:data['n'], lsp#callbag#undefined())
    let a:data['sourceTalkbacks'] = s:createArrayWithSize(a:data['n'], lsp#callbag#undefined())
    let a:data['talkback'] = function('s:combineTalkbackCallback', [a:data])
    let l:i = 0
    for l:Source in a:data['sources']
        let a:data['vals'][l:i] = s:combineEmptyToken
        call l:Source(0, function('s:combineSourceCallback', [a:data, l:i]))
        let l:i = l:i + 1
    endfor
endfunction

function! s:combineTalkbackCallback(data, t, d) abort
    if a:t == 0 | return | endif
    let l:i = 0
    while l:i < a:data['n']
        call a:data['sourceTalkbacks'][l:i](a:t, a:d)
        let l:i = l:i + 1
    endwhile
endfunction

function! s:combineSourceCallback(data, i, t, d) abort
    if a:t == 0
        let a:data['sourceTalkbacks'][a:i] = a:d
        let a:data['Ns'] = a:data['Ns'] - 1
        if a:data['Ns'] == 0 | call a:data['sink'](0, a:data['talkback']) | endif
    elseif a:t == 1
        if a:data['Nd'] <= 0
            let l:_Nd = 0
        else
            if a:data['vals'][a:i] == s:combineEmptyToken
                let a:data['Nd'] = a:data['Nd'] - 1
            endif
            let l:_Nd = a:data['Nd']
        endif
        let a:data['vals'][a:i] = a:d
        if l:_Nd == 0
            let l:arr = s:createArrayWithSize(a:data['n'], lsp#callbag#undefined())
            let l:j = 0
            while l:j < a:data['n']
                let l:arr[l:j] = a:data['vals'][l:j]
                let l:j = l:j + 1
            endwhile
            call a:data['sink'](1, l:arr)
        endif
    elseif a:t == 2
        let a:data['Ne'] = a:data['Ne'] - 1
        if a:data['Ne'] == 0
            call a:data['sink'](2, lsp#callbag#undefined())
        endif
    else
        call a:data['sink'](a:t, a:d)
    endif
endfunction
" }}}

" distinctUntilChanged {{{
function! s:distinctUntilChangedDefaultCompare(a, b) abort
    return a:a == a:b
endfunction

function! lsp#callbag#distinctUntilChanged(...) abort
    let l:data = { 'compare': a:0 == 0 ? function('s:distinctUntilChangedDefaultCompare') : a:1 }
    return function('s:distinctUntilChangedSourceFactory', [l:data])
endfunction

function! s:distinctUntilChangedSourceFactory(data, source) abort
    let a:data['source'] = a:source
    return function('s:distinctUntilChangedSinkFactory', [a:data])
endfunction

function! s:distinctUntilChangedSinkFactory(data, start, sink) abort
    if a:start != 0 | return | endif
    let a:data['sink'] = a:sink
    let a:data['inited'] = 0
    call a:data['source'](0, function('s:distinctUntilChangedSourceCallback', [a:data]))
endfunction

function! s:distinctUntilChangedSourceCallback(data, t, d) abort
    if a:t == 0 | let a:data['talkback'] = a:d | endif
    if a:t != 1
        call a:data['sink'](a:t, a:d)
        return
    endif

    if a:data['inited'] && has_key(a:data, 'prev') && a:data['compare'](a:data['prev'], a:d)
        call a:data['talkback'](1, lsp#callbag#undefined())
        return
    endif

    let a:data['inited'] = 1
    let a:data['prev'] = a:d
    call a:data['sink'](1, a:d)
endfunction
" }}}

" takeUntil() {{{
function! lsp#callbag#takeUntil(notfier) abort
    let l:data = { 'notifier': a:notfier }
    return function('s:takeUntilNotifier', [l:data])
endfunction

function! s:takeUntilNotifier(data, source) abort
    let a:data['source'] = a:source
    return function('s:takeUntilFactory', [a:data])
endfunction

let s:takeUntilUniqueToken = '__callback__take_until_unique_token__'

function! s:takeUntilFactory(data, start, sink) abort
    if a:start != 0 | return | endif
    let a:data['sink'] = a:sink
    let a:data['inited'] = 1
    let a:data['sourceTalkback'] = 0
    let a:data['notiferTalkback'] = 0
    let a:data['done'] = s:takeUntilUniqueToken
    call a:data['source'](0, function('s:takeUntilSourceCallback', [a:data]))
endfunction

function! s:takeUntilSourceCallback(data, t, d) abort
    if a:t == 0
        let a:data['sourceTalkback'] = a:d
        call a:data['notifier'](0, function('s:takeUntilNotifierCallback', [a:data]))
        let a:data['inited'] = 1
        call a:data['sink'](0, function('s:takeUntilSinkCallback', [a:data]))
        if a:data['done'] != s:takeUntilUniqueToken | call a:data['sink'](2, a:data['done']) | endif
        return
    endif
    if a:t == 2
        call a:data['notifierTalkback'](2, lsp#callbag#undefined())
    endif
    if a:data['done'] == s:takeUntilUniqueToken
        call a:data['sink'](a:t, a:d)
    endif
endfunction

function! s:takeUntilNotifierCallback(data, t, d) abort
    if a:t == 0
        let a:data['notifierTalkback'] = a:d
        call a:data['notifierTalkback'](1, lsp#callbag#undefined())
        return
    endif
    if a:t == 1
        let a:data['done'] = 0
        call a:data['notifierTalkback'](2, lsp#callbag#undefined())
        call a:data['sourceTalkback'](2, lsp#callbag#undefined())
        if a:data['inited'] | call a:data['sink'](2, lsp#callbag#undefined()) | endif
        return
    endif
    if a:t ==2
        let a:data['notifierTalkback'] = 0
        let a:data['done'] = a:d
        if a:d != 0
            call a:data['sourceTalkback'](2, lsp#callbag#undefined())
            if a:data['inited'] | call a:data['sink'](a:t, a:d) | endif
        endif
    endif
endfunction

function! s:takeUntilSinkCallback(data, t, d) abort
    if a:data['done'] != s:takeUntilUniqueToken | return | endif
    if a:t == 2 && has_key(a:data, 'notifierTalkback') && a:data['notifierTalkback'] != 0 | call a:data['notifierTalkback'](2, lsp#callbag#undefined()) | endif
    call a:data['sourceTalkback'](a:t, a:d)
endfunction
" }}}

" takeWhile() {{{
function! lsp#callbag#takeWhile(predicate) abort
    let l:data = { 'predicate': a:predicate }
    return function('s:takeWhileFactory', [l:data])
endfunction

function! s:takeWhileFactory(data, source) abort
    let a:data['source'] = a:source
    return function('s:takeWhileSourceFactory', [a:data])
endfunction

function! s:takeWhileSourceFactory(data, start, sink) abort
    if a:start != 0 | return | endif
    let a:data['sink'] = a:sink
    call a:data['source'](0, function('s:takeWhileSourceCallback', [a:data]))
endfunction

function! s:takeWhileSourceCallback(data, t, d) abort
    if a:t == 0
        let a:data['sourceTalkback'] = a:d
    endif

    if a:t == 1 && !a:data['predicate'](a:d)
        call a:data['sourceTalkback'](2, lsp#callbag#undefined())
        call a:data['sink'](2, lsp#callbag#undefined())
        return
    endif

    call a:data['sink'](a:t, a:d)
endfunction
" }}}

" group() {{{
function! lsp#callbag#group(n) abort
    let l:data = { 'n': a:n }
    return function('s:groupN', [l:data])
endfunction

function! s:groupN(data, source) abort
    let a:data['source'] = a:source
    return function('s:groupFactory', [a:data])
endfunction

function! s:groupFactory(data, start, sink) abort
    if a:start != 0 | return | endif
    let a:data['sink'] = a:sink
    let a:data['chunk'] = []
    call a:data['source'](0, function('s:groupSourceCallback', [a:data]))
endfunction

function! s:groupSourceCallback(data, t, d) abort
    if a:t == 0 | let a:data['talkback'] = a:d | endif
    if a:t == 1
        call add(a:data['chunk'], a:d)
        if len(a:data['chunk']) == a:data['n']
            call a:data['sink'](a:t, remove(a:data['chunk'], 0, a:data['n'] - 1))
        endif
        call a:data['talkback'](1, lsp#callbag#undefined())
    else
        if a:t == 2 && len(a:data['chunk']) > 0
            call a:data['sink'](1, remove(a:data['chunk'], 0, len(a:data['chunk']) - 1))
        else
            call a:data['sink'](a:t, a:d)
        endif
    endif
endfunction
" }}}

" flatten() {{{
function! lsp#callbag#flatten() abort
    return function('s:flattenSource')
endfunction

function! s:flattenSource(source) abort
    let l:data = { 'source': a:source }
    return function('s:flattenFactory', [l:data])
endfunction

function! s:flattenFactory(data, start, sink) abort
    if a:start != 0 | return | endif
    let a:data['sink'] = a:sink
    let a:data['outerEnded'] = 0
    let a:data['outerTalkback'] = 0
    let a:data['innerTalkback'] = 0
    let a:data['talkback'] = function('s:flattenTalkbackCallback', [a:data])
    call a:data['source'](0, function('s:flattenSourceCallback', [a:data]))
endfunction

function! s:flattenTalkbackCallback(data, t, d) abort
    if a:t == 1
        if a:data['innerTalkback'] != 0
            call a:data['innerTalkback'](1, a:d)
        else
            call a:data['outerTalkback'](1, a:d)
        endif
    endif
    if a:t == 2
        if a:data['innerTalkback'] != 0 | call a:data['innerTalkback'](2, lsp#callbag#undefined()) | endif
        call a:data['outerTalkback'](2, lsp#callbag#undefined())
    endif
endfunction

function! s:flattenSourceCallback(data, t, d) abort
    if a:t == 0
        let a:data['outerTalkback'] = a:d
        call a:data['sink'](0, a:data['talkback'])
    elseif a:t == 1
        let l:InnerSource = a:d
        if a:data['innerTalkback'] != 0 | call a:data['innerTalkback'](2, lsp#callbag#undefined()) | endif
        call l:InnerSource(0, function('s:flattenInnerSourceCallback', [a:data]))
    elseif a:t == 2 && !lsp#callbag#isUndefined(a:d)
        if a:data['innerTalkback'] != 0 | call a:data['innerTalkback'](2, lsp#callbag#undefined()) | endif
        call a:data['outerTalkback'](1, a:d)
    elseif a:t == 2
        if a:data['innerTalkback'] == 0
            call a:data['sink'](2, lsp#callbag#undefined())
        else
            let a:data['outerEnded'] = 1
        endif
    endif
endfunction

function! s:flattenInnerSourceCallback(data, t, d) abort
    if a:t == 0
        let a:data['innerTalkback'] = a:d
        call a:data['innerTalkback'](1, lsp#callbag#undefined())
    elseif a:t == 1
        call a:data['sink'](1, a:d)
    elseif a:t == 2 && !lsp#callbag#isUndefined(a:d)
        call a:data['outerTalkback'](2, lsp#callbag#undefined())
        call a:data['sink'](2, a:d)
    elseif a:t == 2
        if a:data['outerEnded'] != 0
            call a:data['sink'](2, lsp#callbag#undefined())
        else
            let a:data['innerTalkback'] = 0
            call a:data['outerTalkback'](1, lsp#callbag#undefined())
        endif
    endif
endfunction
" }}}

" flatMap() {{{
function! lsp#callbag#flatMap(F) abort
    return lsp#callbag#operate(
        \ lsp#callbag#map(a:F),
        \ lsp#callbag#flatten(),
        \ )
endfunction
" }}}

" scan() {{{
function! lsp#callbag#scan(reducer, seed) abort
    let l:data = { 'reducer': a:reducer, 'seed': a:seed }
    return function('s:scanSource', [l:data])
endfunction

function! s:scanSource(data, source) abort
    let a:data['source'] = a:source
    return function('s:scanFactory', [a:data])
endfunction

function! s:scanFactory(data, start, sink) abort
    if a:start != 0 | return | endif
    let a:data['sink'] = a:sink
    let a:data['acc'] = a:data['seed']
    call a:data['source'](0, function('s:scanSourceCallback', [a:data]))
endfunction

function! s:scanSourceCallback(data, t, d) abort
    if a:t == 1
        let a:data['acc'] = a:data['reducer'](a:data['acc'], a:d)
        call a:data['sink'](1, a:data['acc'])
    else
        call a:data['sink'](a:t, a:d)
    endif
endfunction
" }}}

" reduce() {{{
function! lsp#callbag#reduce(reducer, seed) abort
    let l:data = { 'reducer': a:reducer, 'seed': a:seed }
    return function('s:reduceSource', [l:data])
endfunction

function! s:reduceSource(data, source) abort
    let a:data['source'] = a:source
    return function('s:reduceFactory', [a:data])
endfunction

function! s:reduceFactory(data, start, sink) abort
    if a:start != 0 | return | endif
    let a:data['sink'] = a:sink
    let a:data['acc'] = a:data['seed']
    call a:data['source'](0, function('s:reduceSourceCallback', [a:data]))
endfunction

function! s:reduceSourceCallback(data, t, d) abort
    if a:t == 1
        let a:data['acc'] = a:data['reducer'](a:data['acc'], a:d)
    elseif a:t == 2 && lsp#callbag#isUndefined(a:d)
        call a:data['sink'](1, a:data['acc'])
        call a:data['sink'](2, lsp#callbag#undefined())
    else
        call a:data['sink'](a:t, a:d)
    endif
endfunction
" }}}

" switchMap() {{{
function! lsp#callbag#switchMap(makeSource, ...) abort
    let l:data = { 'makeSource': a:makeSource }
    if a:0 == 1
        let l:data['combineResults'] = a:1
    else
        let l:data['combineResults'] = function('s:switchMapDefaultCombineResults')
    endif
    return function('s:switchMapSourceCallback', [l:data])
endfunction

function! s:switchMapDefaultCombineResults(a, b) abort
    return a:b
endfunction

function! s:switchMapSourceCallback(data, inputSource) abort
    let a:data['inputSource'] = a:inputSource
    return function('s:switchMapFactory', [a:data])
endfunction

function! s:switchMapFactory(data, start, outputSink) abort
    if a:start != 0 | return | endif
    let a:data['outputSink'] = a:outputSink
    let a:data['sourceEnded'] = 0
    call a:data['inputSource'](0, function('s:switchMapInputSourceCallback', [a:data]))
endfunction

function! s:switchMapInputSourceCallback(data, t, d) abort
    if a:t == 0 | call a:data['outputSink'](a:t, a:d) | endif
    if a:t == 1
        if has_key(a:data, 'currSourceTalkback')
            call a:data['currSourceTalkback'](2, lsp#callbag#undefined())
            call remove(a:data, 'currSourceTalkback')
        endif
        let l:CurrSource = a:data['makeSource'](a:d)
        call l:CurrSource(0, function('s:switchMapCurrSourceCallback', [a:data, a:t, a:d]))
    endif
    if a:t == 2
        let a:data['sourceEnded'] = 1
        if !has_key(a:data, 'currSourceTalkback') | call a:data['outputSink'](a:t, a:d) | endif
    endif
endfunction

function! s:switchMapCurrSourceCallback(data, t, d, currT, currD) abort
    if a:currT == 0 | let a:data['currSourceTalkback'] = a:currD | endif
    if a:currT == 1 | call a:data['outputSink'](a:t, a:data['combineResults'](a:d, a:currD)) | endif
    if (a:currT == 0 || a:currT == 1) && has_key(a:data, 'currSourceTalkback')
        call a:data['currSourceTalkback'](1, lsp#callbag#undefined())
    endif
    if a:currT == 2
        call remove(a:data, 'currSourceTalkback')
        if a:data['sourceEnded'] | call a:data['outputSink'](a:currT, a:currD) | endif
    endif
endfunction
" }}}

" {{{
function! lsp#callbag#share(source) abort
    let l:data = { 'source': a:source, 'sinks': [] }
    return function('s:shareFactory', [l:data])
endfunction

function! s:shareFactory(data, start, sink) abort
    if a:start != 0 | return | endif
    call add(a:data['sinks'], a:sink)

    let a:data['talkback'] = function('s:shareTalkbackCallback', [a:data, a:sink])

    if len(a:data['sinks']) == 1
        call a:data['source'](0, function('s:shareSourceCallback', [a:data, a:sink]))
        return
    endif

    call a:sink(0, a:data['talkback'])
endfunction

function! s:shareTalkbackCallback(data, sink, t, d) abort
    if a:t == 2
        let l:i = 0
        let l:found = 0
        while l:i < len(a:data['sinks'])
            if a:data['sinks'][l:i] == a:sink
                let l:found = 1
                break
            endif
            let l:i += 1
        endwhile

        if l:found
            call remove(a:data['sinks'], l:i)
        endif

        if empty(a:data['sinks'])
            call a:data['sourceTalkback'](2, lsp#callbag#undefined())
        endif
    else
        call a:data['sourceTalkback'](a:t, a:d)
    endif
endfunction

function! s:shareSourceCallback(data, sink, t, d) abort
    if a:t == 0
        let a:data['sourceTalkback'] = a:d
        call a:sink(0, a:data['talkback'])
    else
        for l:S in a:data['sinks']
            call l:S(a:t, a:d)
        endfor
    endif
    if a:t == 2
        let a:data['sinks'] = []
    endif
endfunction
" }}}

" materialize() {{{
function! lsp#callbag#materialize() abort
    let l:data = {}
    return function('s:materializeF', [l:data])
endfunction

function! s:materializeF(data, source) abort
    let a:data['source'] = a:source
    return function('s:materializeFSource', [a:data])
endfunction

function! s:materializeFSource(data, start, sink) abort
    if a:start != 0 | return | endif
    let a:data['sink'] = a:sink
    call a:data['source'](0, function('s:materializeFSourceCallback', [a:data]))
endfunction

function! s:materializeFSourceCallback(data, t, d) abort
    if a:t == 1
        call a:data['sink'](1, lsp#callbag#createNextNotification(a:d))
    elseif a:t == 2
        call a:data['sink'](1, lsp#callbag#isUndefined(a:d)
                    \ ? lsp#callbag#createCompleteNotification()
                    \ : lsp#callbag#createErrorNotification(a:d))
        call a:data['sink'](2, lsp#callbag#undefined())
    else
        call a:data['sink'](a:t, a:d)
    endif
endfunction
" }}}

" Notifications {{{
function! lsp#callbag#createNextNotification(d) abort
    return { 'kind': 'N', 'value': a:d }
endfunction

function! lsp#callbag#createCompleteNotification() abort
    return { 'kind': 'C' }
endfunction

function! lsp#callbag#createErrorNotification(d) abort
    return { 'kind': 'E', 'error': a:d }
endfunction

function! lsp#callbag#isNextNotification(d) abort
    return a:d['kind'] ==# 'N'
endfunction

function! lsp#callbag#isCompleteNotification(d) abort
    return a:d['kind'] ==# 'C'
endfunction

function! lsp#callbag#isErrorNotification(d) abort
    return a:d['kind'] ==# 'E'
endfunction
" }}}

" spawn {{{
" let s:Stdin = lsp#callbag#makeSubject()
" call lsp#callbag#spawn(['bash', '-c', 'read i; echo $i'], {
"   \ 'stdin': s:Stdin,
"   \ 'stdout': 0,
"   \ 'stderr': 0,
"   \ 'exit': 0,
"   \ 'start': 0, " when job starts before subscribing to stdin
"   \ 'ready': 0, " when job starts and after subscribing to stdin
"   \ 'pid': 0,
"   \ 'failOnNonZeroExitCode': 1,
"   \ 'failOnStdinError': 1,
"   \ 'normalize': 'raw' | 'string' | 'array', (defaults to raw),
"   \ 'env': {},
"   \ })
"   call s:Stdin(1, 'hi')
"   call s:Stdin(2, lsp#callbag#undefined()) " requried to close stdin
function! lsp#callbag#spawn(cmd, ...) abort
    let l:data = { 'cmd': a:cmd, 'opt': a:0 > 0 ? copy(a:000[0]) : {} }
    return lsp#callbag#create(function('s:spawnCreate', [l:data]))
endfunction

function! s:spawnCreate(data, next, error, complete) abort
    let a:data['next'] = a:next
    let a:data['error'] = a:error
    let a:data['complete'] = a:complete
    let a:data['state'] = {}
    let a:data['dispose'] = 0
    let a:data['exit'] = 0
    let a:data['close'] = 0

    let l:normalize = get(a:data['opt'], 'normalize', 'raw')

    if has('nvim')
        let a:data['jobopt'] = {
            \ 'on_exit': function('s:spawnNeovimOnExit', [a:data]),
            \ }
        if l:normalize ==# 'string'
            let a:data['normalize'] = function('s:spawnNormalizeNeovimString')
        else
            let a:data['normalize'] = function('s:spawnNormalizeRaw')
        endif
        if get(a:data['opt'], 'stdout', 0) | let a:data['jobopt']['on_stdout'] = function('s:spawnNeovimOnStdout', [a:data]) | endif
        if get(a:data['opt'], 'stderr', 0) | let a:data['jobopt']['on_stderr'] = function('s:spawnNeovimOnStderr', [a:data]) | endif
        if has_key(a:data['opt'], 'env') | let a:data['jobopt']['env'] = a:data['opt']['env'] | endif
        let a:data['jobid'] = jobstart(a:data['cmd'], a:data['jobopt'])
    else
        let a:data['jobopt'] = {
            \ 'exit_cb': function('s:spawnVimExitCb', [a:data]),
            \ 'close_cb': function('s:spawnVimCloseCb', [a:data]),
            \ }
        if get(a:data['opt'], 'stdout', 0) | let a:data['jobopt']['out_cb'] = function('s:spawnVimOutCb', [a:data]) | endif
        if get(a:data['opt'], 'stderr', 0) | let a:data['jobopt']['err_cb'] = function('s:spawnVimErrCb', [a:data]) | endif
        if has_key(a:data['opt'], 'env') | let a:data['jobopt']['env'] = a:data['opt']['env'] | endif
        if l:normalize ==# 'array'
            let a:data['normalize'] = function('s:spawnNormalizeVimArray')
        else
            let a:data['normalize'] = function('s:spawnNormalizeRaw')
        endif
        if has('patch-8.1.350') | let a:data['jobopt']['noblock'] = 1 | endif
        let a:data['stdinBuffer'] = ''
        let a:data['job'] = job_start(a:data['cmd'], a:data['jobopt'])
        let a:data['jobchannel'] = job_getchannel(a:data['job'])
        let a:data['jobid'] = ch_info(a:data['jobchannel'])['id']
    endif

    if a:data['jobid'] < 0 | return | endif " jobstart failed. on_exit will notify with error

    if get(a:data['opt'], 'pid', 0)
        if has('nvim')
            let a:data['pid'] = jobpid(a:data['jobid'])
            let l:startdata['pid'] = a:data['pid']
        else
            let l:jobinfo = job_info(a:data['job'])
            if type(l:jobinfo) == type({}) && has_key(l:jobinfo, 'process')
                let a:data['pid'] = l:jobinfo['process']
                let l:startdata['pid'] = a:data['pid']
            endif
        endif
    endif

    if get(a:data['opt'], 'start', 0)
        let l:startdata = { 'id': a:data['jobid'], 'state': a:data['state'] }
        call a:data['next']({ 'event': 'start', 'data': l:startdata })
    endif

    if has_key(a:data['opt'], 'stdin')
        let a:data['stdinDispose'] = lsp#callbag#pipe(
            \ a:data['opt']['stdin'],
            \ lsp#callbag#subscribe({
            \   'next': (has('nvim') ? function('s:spawnNeovimStdinNext', [a:data]) : function('s:spawnVimStdinNext', [a:data])),
            \   'error': (has('nvim') ? function('s:spawnNeovimStdinError', [a:data]) : function('s:spawnVimStdinError', [a:data])),
            \   'complete': (has('nvim') ? function('s:spawnNeovimStdinComplete', [a:data]) : function('s:spawnVimStdinComplete', [a:data])),
            \ }),
            \ )
    endif

    if get(a:data['opt'], 'ready', 0)
        let l:readydata = { 'id': a:data['jobid'], 'state': a:data['state'] }
        if has_key(a:data, 'pid') | let l:readydata['pid'] = a:data['pid'] | endif
        call a:data['next']({ 'event': 'ready', 'data': l:readydata })
    endif

    return function('s:spawnDispose', [a:data])
endfunction

function! s:spawnJobStop(data) abort
    if has('nvim')
        try
            call jobstop(a:data['jobid'])
        catch /^Vim\%((\a\+)\)\=:E900/
            " NOTE:
            " Vim does not raise exception even the job has already closed so fail
            " silently for 'E900: Invalid job id' exception
        endtry
    else
        call job_stop(a:data['job'])
    endif
endfunction

function! s:spawnDispose(data) abort
    let a:data['dispose'] = 1
    call s:spawnJobStop(a:data)
endfunction

function! s:spawnNeovimStdinNext(data, x) abort
    call jobsend(a:data['jobid'], a:x)
endfunction

function! s:spawnVimStdinNext(data, x) abort
    " Ref: https://groups.google.com/d/topic/vim_dev/UNNulkqb60k/discussion
    let a:data['stdinBuffer'] .= a:x
    call s:spawnVimStdinNextFlushBuffer(a:data)
endfunction

function! s:spawnVimStdinNextFlushBuffer(data) abort
    " https://github.com/vim/vim/issues/2548
    " https://github.com/natebosch/vim-lsc/issues/67#issuecomment-357469091
    sleep 1m
    if len(a:data['stdinBuffer']) <= 4096
        call ch_sendraw(a:data['jobchannel'], a:data['stdinBuffer'])
        let a:data['stdinBuffer'] = ''
    else
        let l:to_send = a:data['stdinBuffer'][:4095]
        let a:data['stdinBuffer'] = a:data['stdinBuffer'][4096:]
        call ch_sendraw(a:data['jobchannel'], l:to_send)
        call timer_start(1, function('s:spawnVimStdinNextFlushBuffer', [a:data]))
    endif
endfunction

function! s:spawnNeovimStdinError(data, x) abort
    let a:data['stdinError'] = a:x
    if get(a:data['opt'], 'failOnStdinError', 1) | call s:spawnJobStop(a:data) | endif
endfunction

function! s:spawnVimStdinError(data, x) abort
    let a:data['stdinError'] = a:x
    if get(a:data['opt'], 'failOnStdinError', 1) | call s:spawnJobStop(a:data) | endif
endfunction

function! s:spawnNeovimStdinComplete(data) abort
    call chanclose(a:data['jobid'], 'stdin')
endfunction

function! s:spawnVimStdinComplete(data) abort
   " There is no easy way to know when ch_sendraw() finishes writing data
   " on a non-blocking channels -- has('patch-8.1.889') -- and because of
   " this, we cannot safely call ch_close_in().
    while len(a:data['stdinBuffer']) != 0
        sleep 1m
    endwhile
    call ch_close_in(a:data['jobchannel'])
endfunction

function! s:spawnNormalizeRaw(data) abort
    return a:data
endfunction

function! s:spawnNormalizeNeovimString(data) abort
    " convert array to string since neovim uses array split by \n by default
    return join(a:data, "\n")
endfunction

function! s:spawnNormalizeVimArray(data) abort
    " convert string to array since vim uses string by default.
    return split(a:data, "\n", 1)
endfunction

function! s:spawnNeovimOnStdout(data, id, d, event) abort
    call a:data['next']({ 'event': 'stdout', 'data': a:data['normalize'](a:d), 'state': a:data['state'] })
endfunction

function! s:spawnNeovimOnStderr(data, id, d, event) abort
    call a:data['next']({ 'event': 'stderr', 'data': a:data['normalize'](a:d), 'state': a:data['state'] })
endfunction

function! s:spawnNeovimOnExit(data, id, d, event) abort
    let a:data['exit'] = 1
    let a:data['close'] = 1
    let a:data['exitcode'] = a:d
    call s:spawnNotifyExit(a:data)
endfunction

function! s:spawnVimOutCb(data, id, d, ...) abort
    call a:data['next']({ 'event': 'stdout', 'data': a:data['normalize'](a:d), 'state': a:data['state'] })
endfunction

function! s:spawnVimErrCb(data, id, d, ...) abort
    call a:data['next']({ 'event': 'stderr', 'data': a:data['normalize'](a:d), 'state': a:data['state'] })
endfunction

function! s:spawnVimExitCb(data, id, d) abort
    let a:data['exit'] = 1
    let a:data['exitcode'] = a:d
    " for more info refer to :h job-start
    " job may exit before we read the output and output may be lost.
    " in unix this happens because closing the write end of a pipe
    " causes the read end to get EOF.
    " close and exit has race condition, so wait for both to complete
    if a:data['close'] && a:data['exit']
        call s:spawnNotifyExit(a:data)
    endif
endfunction

function! s:spawnVimCloseCb(data, id) abort
    let a:data['close'] = 1
    if a:data['close'] && a:data['exit']
        call s:spawnNotifyExit(a:data)
    endif
endfunction

function! s:spawnNotifyExit(data) abort
    if a:data['dispose'] | return | end
    if has_key(a:data, 'stdinDispose') | call a:data['stdinDispose']() | endif
    if get(a:data['opt'], 'failOnStdinError', 1) && has_key(a:data, 'stdinError')
        call a:data['error'](a:data['stdinError'])
        return
    endif
    if get(a:data['opt'], 'exit', 0)
        call a:data['next']({ 'event': 'exit', 'data': a:data['exitcode'], 'state': a:data['state'] })
    endif
    if get(a:data['opt'], 'failOnNonZeroExitCode', 1) && a:data['exitcode'] != 0
        call a:data['error']('Spawn for job ' . a:data['jobid'] .' failed with exit code ' . a:data['exitcode'] . '. ')
    else
        call a:data['complete']()
    endif
endfunction
" }}}

" vim:ts=4:sw=4:ai:foldmethod=marker:foldlevel=0: