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
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
|
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, chan?: integer } #
--- - `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.
--- - `chan`: only handle responses from this channel.
---@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 = require('vim._core.util').nvim_on(
'TermResponse',
opts.group,
{ nested = true },
---@param ev {data: vim.event.termresponse.data}
function(ev)
if opts.chan and ev.data.chan ~= opts.chan then
return
end
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 opts? { timeout?: integer, on_timeout?: fun(), group?: integer|string, chan?: integer }
--- @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.
---@overload fun(caps:string|table, on_response:fun(cap:string, found:boolean, seq:string?))
function M.query(caps, opts, on_response)
if type(opts) == 'function' then
on_response = opts
opts = {}
end
vim.validate('caps', caps, { 'string', 'table' })
vim.validate('opts', opts, 'table', true)
vim.validate('on_response', on_response, 'function')
opts = opts or {}
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, {
timeout = opts.timeout,
group = opts.group,
chan = opts.chan,
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
if opts.on_timeout then
opts.on_timeout()
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, chan?: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
--- Displays error even when this code runs in the context of the client
--- @param msg string
local function broadcast_error(msg)
vim.rpcnotify(0, 'nvim_echo', {
{
msg,
},
}, true, { err = true })
end
--- Get user overrides for terminfo entries as a table. See |$NVIM_TERMDEFS|
function M._get_termdefs()
local termdefs_raw = os.getenv('NVIM_TERMDEFS')
if termdefs_raw ~= nil then
local ok, termdefs_or_err = pcall(vim.json.decode, termdefs_raw)
if not ok then
broadcast_error('E557: Failed to parse $NVIM_TERMDEFS: ' .. vim.inspect(termdefs_or_err))
return
elseif type(termdefs_or_err) ~= 'table' or vim.isarray(termdefs_or_err) then
broadcast_error('E557: expected $NVIM_TERMDEFS to be table. :help $NVIM_TERMDEFS')
return
end
return termdefs_or_err
end
end
return M
|
