X-Git-Url: https://git.madduck.net/etc/vim.git/blobdiff_plain/5a4872f466ebd76ddd532bdf2798554421c53df4..fe3919e725e156d751069662d11e38f7b4791de1:/.vim/bundle/vim-lsp/autoload/lsp/callbag.vim?ds=sidebyside

diff --git a/.vim/bundle/vim-lsp/autoload/lsp/callbag.vim b/.vim/bundle/vim-lsp/autoload/lsp/callbag.vim
new file mode 100644
index 00000000..f08c4e2b
--- /dev/null
+++ b/.vim/bundle/vim-lsp/autoload/lsp/callbag.vim
@@ -0,0 +1,1700 @@
+" 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: