1701 lines
54 KiB
VimL
1701 lines
54 KiB
VimL
" 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:
|