tabby/tabby-terminal/src/middleware/streamProcessing.ts

155 lines
5.0 KiB
TypeScript
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

import hexdump from 'hexer'
import bufferReplace from 'buffer-replace'
import colors from 'ansi-colors'
import binstring from 'binstring'
import { interval, debounce } from 'rxjs'
import { PassThrough, Readable, Writable } from 'stream'
import { ReadLine, createInterface as createReadline, clearLine } from 'readline'
import { SessionMiddleware } from '../api/middleware'
export type InputMode = null | 'local-echo' | 'readline' | 'readline-hex'
export type OutputMode = null | 'hex'
export type NewlineMode = null | 'cr' | 'lf' | 'crlf' | 'implicit_cr' | 'implicit_lf'
export interface StreamProcessingOptions {
inputMode?: InputMode
inputNewlines?: NewlineMode
outputMode?: OutputMode
outputNewlines?: NewlineMode
}
export class TerminalStreamProcessor extends SessionMiddleware {
forceEcho = false
private inputReadline: ReadLine|null = null
private inputPromptVisible = false
private inputReadlineInStream: Readable & Writable
private inputReadlineOutStream: Readable & Writable
private started = false
constructor (private options: StreamProcessingOptions) {
super()
this.inputReadlineInStream = new PassThrough()
this.inputReadlineOutStream = new PassThrough()
this.inputReadlineOutStream.on('data', data => {
this.outputToTerminal.next(Buffer.from(data))
})
this.outputToTerminal$.pipe(debounce(() => interval(500))).subscribe(() => {
if (this.started) {
this.onOutputSettled()
}
})
}
start (): void {
this.inputReadline = createReadline({
input: this.inputReadlineInStream,
output: this.inputReadlineOutStream,
terminal: true,
prompt: this.options.inputMode === 'readline-hex' ? 'hex> ' : '> ',
})
this.inputReadline.on('line', line => {
this.onTerminalInput(Buffer.from(line + '\n'))
this.resetInputPrompt()
})
this.started = true
}
feedFromSession (data: Buffer): void {
if (this.options.inputMode?.startsWith('readline')) {
if (this.inputPromptVisible) {
clearLine(this.inputReadlineOutStream, 0)
this.outputToTerminal.next(Buffer.from('\r'))
this.inputPromptVisible = false
}
}
data = this.replaceNewlines(data, this.options.outputNewlines)
if (this.options.outputMode === 'hex') {
this.outputToTerminal.next(Buffer.concat([
Buffer.from('\r\n'),
Buffer.from(hexdump(data, {
group: 1,
gutter: 4,
divide: colors.gray(' '),
emptyHuman: colors.gray(''),
}).replaceAll('\n', '\r\n')),
Buffer.from('\r\n\n'),
]))
} else {
this.outputToTerminal.next(data)
}
}
feedFromTerminal (data: Buffer): void {
if (this.options.inputMode === 'local-echo' || this.forceEcho) {
this.outputToTerminal.next(this.replaceNewlines(data, 'crlf'))
}
if (this.options.inputMode?.startsWith('readline')) {
this.inputReadlineInStream.write(data)
} else {
this.onTerminalInput(data)
}
}
resize (): void {
if (this.options.inputMode?.startsWith('readline')) {
this.inputReadlineOutStream.emit('resize')
}
}
close (): void {
this.inputReadline?.close()
super.close()
}
private onTerminalInput (data: Buffer) {
if (this.options.inputMode === 'readline-hex') {
const tokens = data.toString().split(/\s/g)
data = Buffer.concat(tokens.filter(t => !!t).map(t => {
if (t.startsWith('0x')) {
t = t.substring(2)
}
return binstring(t, { 'in': 'hex' })
}))
}
data = this.replaceNewlines(data, this.options.inputNewlines)
this.outputToSession.next(data)
}
private onOutputSettled () {
if (this.options.inputMode?.startsWith('readline') && !this.inputPromptVisible) {
this.resetInputPrompt()
}
}
private resetInputPrompt () {
this.outputToTerminal.next(Buffer.from('\r\n'))
this.inputReadline?.prompt(true)
this.inputPromptVisible = true
}
private replaceNewlines (data: Buffer, mode?: NewlineMode): Buffer {
if (!mode) {
return data
}
else if (mode == 'implicit_cr') {
return bufferReplace(data, '\n', '\r\n')
}
else if (mode == 'implicit_lf') {
return bufferReplace(data, '\r', '\r\n')
}
data = bufferReplace(data, '\r\n', '\n')
data = bufferReplace(data, '\r', '\n')
const replacement = {
strip: '',
cr: '\r',
lf: '\n',
crlf: '\r\n',
}[mode]
return bufferReplace(data, '\n', replacement)
}
}