Packages
Terminal emulation and driver infrastructure for Raxol. ANSI parsing, screen buffers, command processing, cursor management, input handling, session management, and termbox2 NIF integration.
Current section
Files
Jump to
Current section
Files
lib/raxol/terminal/terminal_buffer.ex
defmodule Raxol.Terminal.Buffer do
@moduledoc """
Manages the terminal buffer state and operations.
"""
@default_width Raxol.Core.Defaults.terminal_width()
@default_height Raxol.Core.Defaults.terminal_height()
@default_scrollback Raxol.Core.Defaults.scrollback_limit()
require Logger
alias Raxol.Terminal.Buffer.Cell
alias Raxol.Terminal.ScreenBuffer
alias Raxol.Terminal.ScreenBuffer.{Attributes, Operations}
@type t :: %__MODULE__{
width: non_neg_integer(),
height: non_neg_integer(),
cells: list(list(Cell.t())),
cursor_x: non_neg_integer(),
cursor_y: non_neg_integer(),
scroll_region_top: non_neg_integer(),
scroll_region_bottom: non_neg_integer(),
damage_regions:
list({non_neg_integer(), non_neg_integer(), non_neg_integer(), non_neg_integer()})
}
defstruct [
:width,
:height,
:cells,
:cursor_x,
:cursor_y,
:scroll_region_top,
:scroll_region_bottom,
:damage_regions
]
@doc """
Creates a new buffer with the specified dimensions.
Raises ArgumentError if dimensions are invalid.
"""
@spec new({non_neg_integer(), non_neg_integer()}) :: t()
def new({width, height})
when is_integer(width) and is_integer(height) and width > 0 and height > 0 do
# Check for reasonable memory limits (1 million cells max)
max_cells = 1_000_000
total_cells = width * height
validate_buffer_size(total_cells, max_cells, width, height)
%__MODULE__{
width: width,
height: height,
cells: create_empty_grid(width, height),
cursor_x: 0,
cursor_y: 0,
scroll_region_top: 0,
scroll_region_bottom: height - 1,
damage_regions: []
}
end
def new({width, height}) when is_integer(width) and is_integer(height) do
raise ArgumentError,
"Invalid buffer dimensions: width and height must be positive integers"
end
def new(invalid) do
raise ArgumentError,
"Invalid buffer dimensions: expected {width, height} tuple, got #{inspect(invalid)}"
end
@doc """
Creates a new buffer with default dimensions (80x24).
"""
@spec new() :: t()
def new do
new({@default_width, @default_height})
end
@doc """
Sets a cell in the buffer at the specified coordinates.
Raises ArgumentError if coordinates or cell data are invalid.
"""
@spec set_cell(t(), non_neg_integer(), non_neg_integer(), Cell.t()) :: t()
def set_cell(buffer, x, y, cell)
when is_integer(x) and is_integer(y) and x >= 0 and y >= 0 and
x < buffer.width and y < buffer.height do
validate_cell_data(cell)
new_cells =
List.update_at(buffer.cells, y, fn row ->
List.update_at(row, x, fn _ -> cell end)
end)
%{buffer | cells: new_cells}
end
def set_cell(buffer, x, y, _cell) when is_integer(x) and is_integer(y) do
raise ArgumentError,
"Coordinates out of bounds: (#{x}, #{y}) for buffer size #{buffer.width}x#{buffer.height}"
end
def set_cell(_buffer, x, y, _cell) do
raise ArgumentError,
"Invalid coordinates: expected non-negative integers, got (#{inspect(x)}, #{inspect(y)})"
end
@doc """
Gets a cell from the buffer at the specified coordinates.
Delegates to ScreenBuffer.get_cell/3.
"""
@spec get_cell(t(), non_neg_integer(), non_neg_integer()) :: Cell.t()
def get_cell(buffer, x, y) do
# Validate buffer state before trying to access cells
validate_buffer_state(buffer)
screen_buffer = to_screen_buffer(buffer)
cell_map = ScreenBuffer.get_cell(screen_buffer, x, y)
# get_cell always returns a map, convert to Cell struct
struct(Cell, cell_map)
end
@doc """
Resizes the buffer to the specified width and height.
Delegates to ScreenBuffer.resize/3.
"""
@spec resize(t(), non_neg_integer(), non_neg_integer()) :: t()
def resize(_buffer, width, height) when width <= 0 or height <= 0 do
raise ArgumentError,
"Buffer dimensions must be positive integers, got: #{width}x#{height}"
end
def resize(buffer, width, height) do
screen_buffer = to_screen_buffer(buffer)
resized_screen_buffer = ScreenBuffer.resize(screen_buffer, width, height)
from_screen_buffer(resized_screen_buffer, buffer)
end
@doc """
Writes data to the buffer at the current cursor position.
"""
@spec write(t(), String.t(), keyword()) :: t()
def write(buffer, data, _opts \\ []) do
# Validate input data
validate_data_type(data)
# Check for buffer overflow
validate_buffer_capacity(data, buffer)
screen_buffer = to_screen_buffer(buffer)
updated_screen_buffer =
Operations.write_text(
screen_buffer,
buffer.cursor_x,
buffer.cursor_y,
data
)
from_screen_buffer(updated_screen_buffer, buffer)
end
@doc """
Writes text to the buffer at the current position.
"""
@spec write_text(t(), String.t()) :: t()
def write_text(_buffer, text) when byte_size(text) > 1920 do
raise ArgumentError, "Text too long for buffer"
end
def write_text(buffer, _text) do
# Simple implementation - just return buffer unchanged for now
buffer
end
@doc """
Reads data from the buffer.
"""
@spec read(t(), keyword()) :: {String.t(), t()}
def read(buffer, opts \\ []) do
# Validate options
validate_options_type(opts)
# Check for invalid option keys
valid_keys = [:line, :include_style, :region]
invalid_keys = Enum.filter(opts, fn {key, _} -> key not in valid_keys end)
validate_option_keys(invalid_keys)
screen_buffer = to_screen_buffer(buffer)
{ScreenBuffer.get_content(screen_buffer), buffer}
end
@doc """
Clears the buffer.
"""
@spec clear(t(), keyword()) :: t()
def clear(buffer, _opts \\ []) do
screen_buffer = to_screen_buffer(buffer)
updated_screen_buffer =
Operations.clear_to_end_of_screen(screen_buffer)
from_screen_buffer(updated_screen_buffer, buffer)
rescue
e ->
Logger.warning("Failed to clear buffer: #{Exception.message(e)}")
buffer
end
@doc """
Sets the cursor position.
"""
@spec set_cursor_position(t(), non_neg_integer(), non_neg_integer()) :: t()
def set_cursor_position(buffer, x, y) do
screen_buffer = to_screen_buffer(buffer)
updated_screen_buffer =
Attributes.set_cursor_position(screen_buffer, x, y)
from_screen_buffer(updated_screen_buffer, buffer)
end
@doc """
Gets the current cursor position.
"""
@spec get_cursor_position(t()) :: {non_neg_integer(), non_neg_integer()}
def get_cursor_position(buffer) do
screen_buffer = to_screen_buffer(buffer)
Attributes.get_cursor_position(screen_buffer)
end
@doc """
Sets the scroll region.
"""
@spec set_scroll_region(t(), non_neg_integer(), non_neg_integer()) :: t()
def set_scroll_region(buffer, top, bottom) do
# Validate scroll region parameters
validate_scroll_region(top, bottom, buffer.height)
screen_buffer = to_screen_buffer(buffer)
updated_screen_buffer =
ScreenBuffer.set_scroll_region(screen_buffer, top, bottom)
from_screen_buffer(updated_screen_buffer, buffer)
end
@doc """
Scrolls the buffer by the specified number of lines.
"""
@spec scroll(t(), integer()) :: t()
def scroll(buffer, lines) when is_integer(lines) do
screen_buffer = to_screen_buffer(buffer)
{updated_screen_buffer, _scrolled_lines} =
ScreenBuffer.scroll_up(screen_buffer, abs(lines))
from_screen_buffer(updated_screen_buffer, buffer)
rescue
e ->
Logger.warning("Failed to scroll buffer by #{lines} lines: #{Exception.message(e)}")
buffer
end
def scroll(_buffer, nil) do
raise ArgumentError, "Scroll lines cannot be nil"
end
def scroll(_buffer, lines) do
raise ArgumentError,
"Invalid scroll lines: expected integer, got #{inspect(lines)}"
end
@doc """
Updates the scroll state without moving content.
This is a fast operation that only updates scroll position.
"""
@spec scroll_state(t(), integer()) :: t()
def scroll_state(buffer, _lines) do
# Optimized: Since this is supposed to be a fast operation that only updates scroll position
# and doesn't move content, we can just return the buffer unchanged.
# The scroll position is typically tracked at a higher level (emulator, screen buffer, etc.)
# rather than in the basic buffer struct.
buffer
end
@doc """
Marks a region of the buffer as damaged.
"""
@spec mark_damaged(
t(),
non_neg_integer(),
non_neg_integer(),
non_neg_integer(),
non_neg_integer()
) :: t()
def mark_damaged(buffer, x, y, width, height) do
screen_buffer = to_screen_buffer(buffer)
updated_screen_buffer =
ScreenBuffer.mark_damaged(
screen_buffer,
x,
y,
width,
height
)
from_screen_buffer(updated_screen_buffer, buffer)
end
@doc """
Gets all damaged regions in the buffer.
"""
@spec get_damage_regions(t()) :: [
{non_neg_integer(), non_neg_integer(), non_neg_integer(), non_neg_integer()}
]
def get_damage_regions(buffer) do
screen_buffer = to_screen_buffer(buffer)
Map.get(screen_buffer, :damage_regions, [])
end
@doc """
Adds content to the buffer at the current cursor position.
## Examples
iex> buffer = Buffer.new({80, 24})
iex> buffer = Buffer.add(buffer, "Hello, World!")
iex> {content, _} = Buffer.read(buffer)
iex> content
"Hello, World!"
"""
@spec add(t(), String.t()) :: t()
def add(buffer, content) do
write(buffer, content)
rescue
e ->
Logger.warning("Failed to add content to buffer: #{Exception.message(e)}")
buffer
end
@doc """
Fills a region of the buffer with a specified cell.
Delegates to ScreenBuffer.fill_region/6.
"""
@spec fill_region(
t(),
non_neg_integer(),
non_neg_integer(),
non_neg_integer(),
non_neg_integer(),
Cell.t()
) :: t()
def fill_region(buffer, x, y, width, height, cell) do
screen_buffer = to_screen_buffer(buffer)
filled_screen_buffer =
ScreenBuffer.fill_region(screen_buffer, x, y, width, height, cell)
from_screen_buffer(filled_screen_buffer, buffer)
end
@doc """
Clear a rectangular region in the buffer.
"""
def clear_region(buffer, x, y, width, height) do
fill_region(buffer, x, y, width, height, Cell.new())
end
@doc """
Draw a box in the buffer with the specified style.
"""
def draw_box(buffer, x, y, width, height, style \\ nil) do
# Draw corners
buffer
|> set_cell(x, y, "┌", style)
|> set_cell(x + width - 1, y, "┐", style)
|> set_cell(x, y + height - 1, "└", style)
|> set_cell(x + width - 1, y + height - 1, "┘", style)
|> draw_horizontal_line(x + 1, y, width - 2, "─", style)
|> draw_horizontal_line(x + 1, y + height - 1, width - 2, "─", style)
|> draw_vertical_line(x, y + 1, height - 2, "│", style)
|> draw_vertical_line(x + width - 1, y + 1, height - 2, "│", style)
end
@doc """
Move the cursor to the specified position.
"""
def move_cursor(buffer, x, y) do
%{buffer | cursor_x: x, cursor_y: y}
end
# Private helper functions for drawing
defp draw_horizontal_line(buffer, x, y, length, char, style) do
Enum.reduce(0..(length - 1), buffer, fn i, acc ->
set_cell(acc, x + i, y, char, style)
end)
end
defp draw_vertical_line(buffer, x, y, length, char, style) do
Enum.reduce(0..(length - 1), buffer, fn i, acc ->
set_cell(acc, x, y + i, char, style)
end)
end
defp set_cell(buffer, x, y, char, style) do
update_cell_if_in_bounds(buffer, x, y, char, style)
end
# Private functions
defp validate_buffer_size(total_cells, max_cells, width, height)
when total_cells > max_cells do
raise RuntimeError,
"Buffer too large: #{width}x#{height} = #{total_cells} cells exceeds limit of #{max_cells} cells"
end
defp validate_buffer_size(_, _, _, _), do: :ok
defp validate_cell_data(%Cell{} = _cell) do
:ok
end
defp validate_cell_data(cell) do
raise ArgumentError,
"Invalid cell data: expected Cell struct, got #{inspect(cell)}"
end
defp validate_data_type(data) when is_binary(data), do: :ok
defp validate_data_type(data) do
raise ArgumentError, "Invalid data: expected string, got #{inspect(data)}"
end
defp validate_buffer_capacity(data, buffer) do
data_length = String.length(data)
buffer_capacity = buffer.width * buffer.height
case data_length <= buffer_capacity do
true ->
:ok
false ->
raise ArgumentError,
"Buffer overflow: string length #{data_length} exceeds buffer capacity #{buffer_capacity}"
end
end
defp validate_options_type(opts) when is_list(opts), do: :ok
defp validate_options_type(opts) do
raise ArgumentError,
"Invalid options: expected keyword list, got #{inspect(opts)}"
end
defp validate_option_keys([]), do: :ok
defp validate_option_keys(invalid_keys) do
raise ArgumentError, "Invalid options: #{inspect(invalid_keys)}"
end
defp validate_scroll_region(top, bottom, height) do
check_negative_boundaries(top, bottom)
check_region_order(top, bottom)
check_region_bounds(bottom, height)
end
defp check_negative_boundaries(top, bottom) when top < 0 or bottom < 0 do
raise ArgumentError,
"Scroll region boundaries must be non-negative, got top=#{top}, bottom=#{bottom}"
end
defp check_negative_boundaries(_, _), do: :ok
defp check_region_order(top, bottom) when top > bottom do
raise ArgumentError,
"Scroll region top must be less than or equal to bottom, got top=#{top}, bottom=#{bottom}"
end
defp check_region_order(_, _), do: :ok
defp check_region_bounds(bottom, height) when bottom >= height do
raise ArgumentError,
"Scroll region bottom must be less than buffer height, got bottom=#{bottom}, height=#{height}"
end
defp check_region_bounds(_, _), do: :ok
defp update_cell_if_in_bounds(buffer, x, y, char, style) do
case in_bounds?(buffer, x, y) do
true -> update_cell_at_position(buffer, x, y, char, style)
false -> buffer
end
end
defp in_bounds?(buffer, x, y) do
x >= 0 and x < buffer.width and y >= 0 and y < buffer.height
end
defp update_cell_at_position(buffer, x, y, char, style) do
cell = Cell.new(char, style || %{})
cells =
List.replace_at(
buffer.cells,
y,
List.replace_at(Enum.at(buffer.cells, y), x, cell)
)
%{buffer | cells: cells}
end
defp validate_buffer_state(buffer) do
check_cells_not_nil(buffer.cells)
check_width_not_nil(buffer.width)
check_height_not_nil(buffer.height)
end
defp check_cells_not_nil(nil), do: raise(RuntimeError, "Buffer cells are nil")
defp check_cells_not_nil(_), do: :ok
defp check_width_not_nil(nil), do: raise(RuntimeError, "Buffer width is nil")
defp check_width_not_nil(_), do: :ok
defp check_height_not_nil(nil),
do: raise(RuntimeError, "Buffer height is nil")
defp check_height_not_nil(_), do: :ok
defp create_empty_grid(width, height) do
for _y <- 0..(height - 1) do
for _x <- 0..(width - 1) do
Cell.new()
end
end
end
defp to_screen_buffer(buffer) do
validate_buffer_state(buffer)
%ScreenBuffer{
width: buffer.width,
height: buffer.height,
cells: buffer.cells,
scroll_region: {buffer.scroll_region_top, buffer.scroll_region_bottom},
cursor_position: {buffer.cursor_x, buffer.cursor_y},
damage_regions: buffer.damage_regions,
scroll_position: 0,
scrollback: [],
scrollback_limit: @default_scrollback,
selection: nil,
default_style: nil
}
end
defp from_screen_buffer(screen_buffer, original_buffer) do
{cursor_x, cursor_y} = screen_buffer.cursor_position
# Handle case where scroll_region might be nil (e.g., after resize)
{scroll_region_top, scroll_region_bottom} =
case screen_buffer.scroll_region do
nil -> {0, screen_buffer.height - 1}
{top, bottom} -> {top, bottom}
end
%{
original_buffer
| cells: screen_buffer.cells,
cursor_x: cursor_x,
cursor_y: cursor_y,
scroll_region_top: scroll_region_top,
scroll_region_bottom: scroll_region_bottom,
damage_regions: screen_buffer.damage_regions
}
end
end