1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
|
local M = {}
--- Send `payload` to the host terminal and listen for `TermResponse`, calling `on_response` for
--- each response. Cleans up after `opts.timeout` ms if the callback never returns `true`.
---
--- The autocommand is removed when:
--- - `on_response()` returns `true`
--- - the timeout fires (and `opts.on_timeout` is called, if given)
--- - the caller explicitly deletes the returned autocmd id
---
---@param payload string Sequence to send via nvim_ui_send(). Use empty string ('') to just register
--- a listener (no sending).
---@param opts? { timeout?: integer, on_timeout?: fun(), group?: integer|string }
--- - `timeout` (default: 1000) ms to wait before giving up, or 0 for never (caller must remove the autocmd).
--- - `on_timeout` optional fn called when the timeout fires.
--- - `group`: augroup for the TermResponse autocmd.
---@param on_response fun(resp:string):boolean? Called for each TermResponse. Return `true` to stop listening.
---@return integer # autocmd id of the TermResponse handler.
function M.request(payload, opts, on_response)
vim.validate('payload', payload, 'string')
vim.validate('opts', opts, 'table', true)
vim.validate('on_response', on_response, 'function')
opts = opts or {}
local timeout = opts.timeout or 1000
local timer ---@type uv.uv_timer_t?
if timeout > 0 then
timer = assert(vim.uv.new_timer())
end
local id = vim.api.nvim_create_autocmd('TermResponse', {
group = opts.group,
nested = true,
callback = function(ev)
local stop = on_response(ev.data.sequence)
-- If on_response is done, cancel the timeout so on_timeout doesn't fire spuriously.
if stop and timer and not timer:is_closing() then
timer:close()
end
return stop
end,
})
if payload ~= '' then
vim.api.nvim_ui_send(payload)
end
if timer then
timer:start(timeout, 0, function()
vim.schedule(function()
pcall(vim.api.nvim_del_autocmd, id)
if opts.on_timeout then
opts.on_timeout()
end
end)
if not timer:is_closing() then
timer:close()
end
end)
end
return id
end
--- Query the host terminal emulator for terminfo capabilities.
---
--- This function sends the XTGETTCAP DCS sequence to the host terminal emulator asking the terminal
--- to send us its terminal capabilities. These are strings that are normally taken from a terminfo
--- file, however an up to date terminfo database is not always available (particularly on remote
--- machines), and many terminals continue to misidentify themselves or do not provide their own
--- terminfo file, making the terminfo database unreliable.
---
--- Querying the terminal guarantees that we get a truthful answer, but only if the host terminal
--- emulator supports the XTGETTCAP sequence.
---
--- @param caps string|table A terminal capability or list of capabilities to query
--- @param on_response fun(cap:string, found:boolean, seq:string?) Called for each capability in
--- `caps`. `found` is true if the capability was found, else false. `seq` is the control
--- sequence if found, or nil for boolean capabilities.
function M.query(caps, on_response)
vim.validate('caps', caps, { 'string', 'table' })
vim.validate('on_response', on_response, 'function')
if type(caps) ~= 'table' then
caps = { caps }
end
local pending = {} ---@type table<string, boolean>
for _, v in ipairs(caps) do
pending[v] = true
end
local encoded = {} ---@type string[]
for i = 1, #caps do
encoded[i] = vim.text.hexencode(caps[i])
end
local payload = ('\027P+q%s\027\\'):format(table.concat(encoded, ';'))
M.request(payload, {
on_timeout = function()
-- Call the callback for all capabilities that were not found.
for k in pairs(pending) do
on_response(k, false, nil)
end
end,
}, function(resp)
local k, rest = resp:match('^\027P1%+r(%x+)(.*)$')
if not k or not rest then
return
end
local cap = vim.text.hexdecode(k)
if not cap or not pending[cap] then
-- Received a response for a capability we didn't request. This can happen if there are
-- multiple concurrent XTGETTCAP requests
return
end
local seq ---@type string?
if rest:match('^=%x+$') then
seq = vim.text
.hexdecode(rest:sub(2))
:gsub('\\E', '\027')
:gsub('%%p%d', '')
:gsub('\\(%d+)', string.char)
end
on_response(cap, true, seq)
pending[cap] = nil
return next(pending) == nil
end)
end
--- Send an APC sequence to the terminal and call `on_response` for each APC response received.
--- Cleans up after {timeout} milliseconds if no response is received.
---
--- `on_response` receives the full APC sequence including the `\027_` prefix.
--- Return `true` from `on_response` to stop listening.
---
---@param payload string APC sequence to send (full escape sequence including prefix/suffix)
---@param opts {timeout?:integer} Options table (timeout in milliseconds, default 1000)
---@param on_response fun(resp:string):boolean? Callback invoked for each APC TermResponse
---@overload fun(payload:string, on_response:fun(resp:string):boolean?)
function M.query_apc(payload, opts, on_response)
if type(opts) == 'function' then
on_response = opts
opts = {}
end
vim.validate('payload', payload, 'string')
vim.validate('opts', opts, 'table')
vim.validate('on_response', on_response, 'function')
M.request(payload, opts, function(resp)
if resp:match('^\027_') then
return on_response(resp)
end
end)
end
return M
|
