-- mod-version:3
--
-- Crossplatform file based IPC system for lite-xl.
-- @copyright Jefferson Gonzalez <jgmdev@gmail.com>
-- @license MIT
--
local core = require "core"
local config = require "core.config"
local common = require "core.common"
local command = require "core.command"
local Object = require "core.object"
local RootView = require "core.rootview"
local settings_found, settings = pcall(require, "plugins.settings")
---The maximum amount of seconds a message will be broadcasted.
---@type integer
local MESSAGE_EXPIRATION=3
---@class config.plugins.ipc
---@field single_instance boolean
---@field dirs_instance '"new"' | '"add"' | '"change"'
config.plugins.ipc = common.merge({
single_instance = true,
dirs_instance = "new",
-- The config specification used by the settings gui
config_spec = {
name = "Inter-process communication",
{
label = "Single Instance",
description = "Run a single instance of lite-xl.",
path = "single_instance",
type = "toggle",
default = true
},
{
label = "Directories Instance",
description = "Control how to open directories in single instance mode.",
path = "dirs_instance",
type = "selection",
default = "new",
values = {
{"Create a New Instance", "new"},
{"Add to Current Instance", "add"},
{"Change Current Instance Project Directory", "change"}
}
}
}
}, config.plugins.ipc)
---@alias plugins.ipc.onmessageread fun(message: plugins.ipc.message) | nil
---@alias plugins.ipc.onreplyread fun(reply: plugins.ipc.reply) | nil
---@alias plugins.ipc.onmessage fun(message: plugins.ipc.message, reply: plugins.ipc.reply) | nil
---@alias plugins.ipc.onreply fun(reply: plugins.ipc.reply) | nil
---@alias plugins.ipc.function fun(...)
---@alias plugins.ipc.messagetype
---| '"message"'
---| '"method"'
---| '"signal"'
---@class plugins.ipc.message
---Id of the message
---@field id string
---The id of process that sent the message
---@field sender string
---Name of the message
---@field name string
---Type of message.
---@field type plugins.ipc.messagetype | string
---List with id of the instance that should receive the message.
---@field destinations table<integer,string>
---A list of named values sent to receivers.
---@field data table<string,any>
---Time in seconds when the message was sent for automatic expiration purposes.
---@field timestamp number
---Optional callback executed by the receiver when the message is read.
---@field on_read plugins.ipc.onmessageread
---Optional callback executed when a reply to the message is received.
---@field on_reply plugins.ipc.onreply
---The received replies for the message.
---@field replies plugins.ipc.reply[]
---@class plugins.ipc.reply
---Id of the message
---@field id string
---The id of process that sent the message
---@field sender string
---The id of the replier
---@field replier string
---A list of named values sent back to sender.
---@field data table<string,any>
---Time in seconds when the reply was sent for automatic expiration purposes.
---@field timestamp number
---Optional callback executed by the sender when the reply is read.
---@field on_read plugins.ipc.onreplyread
---@class plugins.ipc.instance
---Process id of the instance.
---@field id string
---The position in which the instance was launched.
---@field position integer
---Flag that indicates if this instance was the first started.
---@field primary boolean
---Indicates the last time this instance updated its session file.
---@field last_update integer
---The messages been broadcasted.
---@field messages plugins.ipc.message[]
---The replies been broadcasted.
---@field replies plugins.ipc.reply[]
---Table of properties associated with the instance. (NOT IMPLEMENTED)
---@field properties table
---@class plugins.ipc : core.object
---@field protected id string
---@field protected user_dir string
---@field protected running boolean
---@field protected file string
---@field protected primary boolean
---@field protected position integer
---@field protected messages plugins.ipc.message[]
---@field protected replies plugins.ipc.reply[]
---@field protected listeners table<string,table<integer,plugins.ipc.onmessage>>
---@field protected signals table<string,integer>
---@field protected methods table<string,integer>
---@field protected signal_definitions table<integer,string>
---@field protected method_definitions table<integer,string>
local IPC = Object:extend()
---@class plugins.ipc.threads
---@field cr thread
---@field wake number
---List of threads belonging to all instantiated IPC objects.
---@type plugins.ipc.threads[]
local threads = {}
---Register a new thread to be run on the background.
---@param f function
local function add_thread(f)
local key = #threads + 1
threads[key] = { cr = coroutine.create(f), wake = 0 }
return key
end
---Updates the session file of an IPC object.
---@param self plugins.ipc
local function update_file(self)
local file, errmsg = io.open(self.file, "w+")
if file then
local output = "-- Warning: Generated by IPC system do not edit manually!\n"
.. "return " .. common.serialize(
{
id = self.id,
primary = self.primary,
position = self.position,
last_update = os.time(),
messages = self.messages,
replies = self.replies,
signals = self.signal_definitions,
methods = self.method_definitions
},
{
pretty = true
}
)
output = output:gsub("%s+%[\"on_reply\"%].-\n", "")
file:write(output)
file:close()
else
core.error("IPC Error: failed updating status (%s)", errmsg)
end
end
---Constructor
---@param id? string Defaults to current lite-xl process id.
function IPC:new(id)
self.id = id or tostring(system.get_process_id())
self.user_dir = USERDIR .. "/ipc"
self.file = self.user_dir .. "/" .. self.id .. ".lua"
self.primary = false
self.running = false
self.messages = {}
self.replies = {}
self.listeners = {}
self.signals = {}
self.methods = {}
self.signal_definitions = {}
self.method_definitions = {}
local ipc_dir_status = system.get_file_info(self.user_dir)
if not ipc_dir_status then
local created, errmsg = common.mkdirp(self.user_dir)
if not created then
core.error("Error initializing IPC system: %s", errmsg)
return
end
end
local file, errmsg = io.open(self.file, "w+")
if not file then
core.error("Error initializing IPC system: %s", errmsg)
return
else
file:close()
os.remove(self.file)
end
-- Execute to set the instance position and primary attribute if no other running.
local instances = self:get_instances()
self.primary = #instances == 0 and true or false
self.position = #instances + 1
self:start()
end
---Starts and registers the ipc session and monitoring.
function IPC:start()
if not self.running then
self.running = true
update_file(self)
local wait_time = 0.3
local this = self
self.coroutine_key = add_thread(function()
coroutine.yield(wait_time)
while(true) do
this:read_messages()
this:read_replies()
update_file(this)
coroutine.yield(wait_time)
end
end)
end
end
---Stop and unregister the ipc session and monitoring.
function IPC:stop()
self.running = false
table.remove(threads, self.coroutine_key)
os.remove(self.file)
end
---Get a list of running lite-xl instances.
---@return plugins.ipc.instance[]
function IPC:get_instances()
---@type plugins.ipc.instance[]
local instances = {}
local files, errmsg = system.list_dir(self.user_dir)
if files then
for _, file in ipairs(files) do
if string.match(file, "^%d+%.lua$") then
local path = self.user_dir .. "/" .. file
local file_info = system.get_file_info(path)
if file_info and file_info.type == "file" then
::read_instance_file::
---@type plugins.ipc.instance
local instance = dofile(path)
if instance and instance.id ~= self.id then
if instance.last_update + 2 > os.time() then
table.insert(instances, instance)
else
-- Delete expired instance session maybe result of a crash
os.remove(path)
end
elseif not instance and path ~= self.file then
--We retry reading the file since it was been modified
--by its owner instance.
goto read_instance_file
end
end
end
end
else
core.error("IPC Error: failed getting running instances (%s)", errmsg)
end
local instances_count = #instances
if instances_count > 0 then
table.sort(instances, function(ia, ib)
return ia.position < ib.position
end)
end
if not self.primary and self.position then
if instances_count == 0 or instances[1].position > self.position then
self.primary = true
end
end
return instances
end
---@class plugins.ipc.vardecl
---@field name string
---@field type string
---@field optional boolean
---Generate a string representation of a function
---@param name string
---@param params? plugins.ipc.vardecl[]
---@param returns? plugins.ipc.vardecl[]
---@return string function_definition
local function generate_definition(name, params, returns)
local declaration = name .. "("
if params and #params > 0 then
local params_string = ""
for _, param in ipairs(params) do
params_string = params_string .. param.name
if param.optional then
params_string = params_string .. "?: "
else
params_string = params_string .. ": "
end
params_string = params_string .. param.type .. ", "
end
local params_stripped = params_string:gsub(", $", "")
declaration = declaration .. params_stripped
end
declaration = declaration .. ")"
if returns and #returns > 0 then
declaration = declaration .. " -> "
local returns_string = ""
for _, ret in ipairs(returns) do
if ret.name then
returns_string = returns_string .. ret.name .. ": "
end
returns_string = returns_string .. ret.type
if ret.optional then
returns_string = returns_string .. "?, "
else
returns_string = returns_string .. ", "
end
end
local returns_stripped = returns_string:gsub(", $", "")
declaration = declaration .. returns_stripped
end
return declaration
end
---Retrieve the id of the primary instance if found.
---@return string | nil
function IPC:get_primary_instance()
local instances = self:get_instances()
for _, instance in ipairs(instances) do
if instance.primary then
return instance.id
end
end
return nil
end
---Get a queued message.
---@param message_id string
---@return plugins.ipc.message | nil
function IPC:get_message(message_id)
for _, message in ipairs(self.messages) do
if message.id == message_id then
return message
end
end
return nil
end
---Remove a message from the queue.
---@param message_id string
function IPC:remove_message(message_id)
for m, message in ipairs(self.messages) do
if message.id == message_id then
table.remove(self.messages, m)
break
end
end
end
---Get the reply sent to a specific message.
---@param message_id string
---@return plugins.ipc.reply | nil
function IPC:get_reply(message_id)
for _, reply in ipairs(self.replies) do
if reply.id == message_id then
return reply
end
end
return nil
end
---Verify all the messages sent by running instances, read those directed
---to the currently running instance and reply to them.
function IPC:read_messages()
local instances = self:get_instances()
local awaiting_replies = {}
for _, instance in ipairs(instances) do
for _, message in ipairs(instance.messages) do
for _, destination in ipairs(message.destinations) do
if destination == self.id then
local reply = self:get_reply(message.id)
if not reply then
if message.on_read then
local on_read, errmsg = load(message.on_read)
if on_read then
local executed = core.try(function() on_read(message) end)
if not executed then
core.error(
"IPC Error: could not run message on_read\n"
.. "Message: %s\n",
common.serialize(message, {pretty = true})
)
end
else
core.error(
"IPC Error: could not run message on_read (%s)\n"
.. "Message: %s\n",
errmsg,
common.serialize(message, {pretty = true})
)
end
end
---@type plugins.ipc.reply
reply = {}
reply.id = message.id
reply.sender = message.sender
reply.replier = self.id
reply.data = {}
reply.on_read = nil
local type_name = message.type .. "." .. message.name
-- Allow listeners to react to message and modify reply
if self.listeners[type_name] and #self.listeners[type_name] > 0 then
for _, on_message in ipairs(self.listeners[type_name]) do
on_message(message, reply)
end
end
if reply.on_read then
reply.on_read = string.dump(reply.on_read)
end
reply.timestamp = os.time()
end
table.insert(awaiting_replies, reply)
break
end
end
end
end
self.replies = awaiting_replies
end
---Reads replies directed to messages sent by the currently running instance
---and if any returns them.
---@return plugins.ipc.reply[] | nil
function IPC:read_replies()
if #self.messages == 0 then
return
end
local instances = self:get_instances()
local replies = {}
local messages_removed = 0;
for m=1, #self.messages do
local message = self.messages[m-messages_removed]
local message_removed = false
local destinations_removed = 0
for d=1, #message.destinations do
local destination = message.destinations[d-destinations_removed]
local found = false
for _, instance in ipairs(instances) do
if instance.id == destination then
found = true
for _, reply in ipairs(instance.replies) do
if reply.id == message.id then
local reply_registered = false
for _, message_reply in ipairs(message.replies) do
if message_reply.replier == instance.id then
reply_registered = true
break
end
end
if not reply_registered then
if message.on_reply then
message.on_reply(reply)
end
if reply.on_read then
local on_read, errmsg = load(reply.on_read)
if on_read then
local executed = core.try(function() on_read(reply) end)
if not executed then
core.error(
"IPC Error: could not run reply on_read\n"
.. "Message: %s\n"
.. "Reply: %s",
common.serialize(message, {pretty = true}),
common.serialize(reply, {pretty = true})
)
end
else
core.error(
"IPC Error: could not run reply on_read (%s)\n"
.. "Message: %s\n"
.. "Reply: %s",
errmsg,
common.serialize(message, {pretty = true}),
common.serialize(reply, {pretty = true})
)
end
end
table.insert(replies, reply)
table.insert(message.replies, reply)
end
end
end
break
end
end
if not found then
table.remove(message.destinations, d-destinations_removed)
destinations_removed = destinations_removed + 1
if #message.destinations == 0 then
table.remove(self.messages, m-messages_removed)
messages_removed = messages_removed + 1
message_removed = true
end
end
end
if
not message_removed
and
(
#message.replies == #message.destinations
or
message.timestamp + MESSAGE_EXPIRATION < os.time()
)
then
table.remove(self.messages, m-messages_removed)
messages_removed = messages_removed + 1
end
end
return replies
end
---Blocks execution of current instance to wait for all replies by the
---specified message and when finished returns them.
---@param message_id string
---@return plugins.ipc.reply[] | nil
function IPC:wait_for_replies(message_id)
local message_data = self:get_message(message_id)
update_file(self)
if message_data then
self:read_replies()
while true do
if
message_data.replies
and
#message_data.replies == #message_data.destinations
then
return message_data.replies
elseif not self:get_message(message_id) then
return message_data.replies
end
self:read_replies()
end
end
return nil
end
---Blocks execution of current instance to wait for all messages to
---be replied to.
function IPC:wait_for_messages()
update_file(self)
while #self.messages > 0 do
self:read_replies()
system.sleep(0.1)
end
end
---@class plugins.ipc.sendmessageoptions
---@field data table<string,any> @Optional data given to the receiver.
---@field on_reply plugins.ipc.onreply @Callback that allows monitoring all the replies received for this message.
---@field on_read plugins.ipc.onmessage @Function executed by the message receiver.
---@field destinations string | table<integer,string> | nil @Id of the running instances to receive the message, if not set all running instances will receive the message.
---Queue a new message to be sent to other lite-xl instances.
---@param name string
---@param options? plugins.ipc.sendmessageoptions
---@param message_type? plugins.ipc.messagetype
---@return string | nil message_id
function IPC:send_message(name, options, message_type)
options = options or {}
local found_destinations = {}
local instances = self:get_instances()
local destinations = options.destinations
if type(destinations) == "string" then
destinations = { destinations }
end
if not destinations then
for _, instance in ipairs(instances) do
table.insert(found_destinations, instance.id)
end
else
for _, destination in ipairs(destinations) do
for _, instance in ipairs(instances) do
if instance.id == destination then
table.insert(found_destinations, destination)
end
end
end
end
if #found_destinations <= 0 then
return nil
end
---@type plugins.ipc.message
local message = {}
message.id = self.id .. "." .. tostring(system.get_time())
message.name = name
message.type = message_type or "message"
message.sender = self.id
message.data = options.data or {}
message.destinations = found_destinations
message.timestamp = os.time()
message.on_reply = options.on_reply or nil
message.on_read = options.on_read and string.dump(options.on_read) or nil
message.replies = {}
table.insert(self.messages, message)
update_file(self)
return message.id
end
---Add a listener for a given type of message.
---@param name string
---@param callback plugins.ipc.onmessage
---@param message_type? plugins.ipc.messagetype
---@return integer listener_position
function IPC:listen_message(name, callback, message_type)
message_type = message_type or "message"
local type_name = message_type .. "." .. name
if not self.listeners[type_name] then
self.listeners[type_name] = {}
end
table.insert(self.listeners[type_name], callback)
return #self.listeners[type_name]
end
---Listen for a given signal.
---@param name string
---@param callback plugins.ipc.function
---@return integer listener_position
function IPC:listen_signal(name, callback)
local signal_cb = function(message)
callback(table.unpack(message.data))
end
return self:listen_message(name, signal_cb, "signal")
end
---Add a new signal that can be sent to other instances.
---@param name string A unique name for the signal.
---@param params? plugins.ipc.vardecl[] Parameters that are going to be passed into callback.
function IPC:register_signal(name, params)
if self.signals[name] then
core.log_quiet("IPC: Overriding signal '%s'", name)
table.remove(self.signal_definitions, self.signals[name])
end
self.signals[name] = table.insert(
self.signal_definitions,
generate_definition(name, params)
)
table.sort(self.signal_definitions)
end
---Add a new method that can be invoked from other instances.
---@param name string A unique name for the method.
---@param method fun(...) Function invoked when the method is called.
---@param params? plugins.ipc.vardecl[] Parameters that are going to be passed into method.
---@param returns? plugins.ipc.vardecl[] Return values of the method.
function IPC:register_method(name, method, params, returns)
if self.methods[name] then
core.log_quiet("IPC: Overriding method '%s'", name)
table.remove(self.method_definitions, self.methods[name])
end
self.methods[name] = table.insert(
self.method_definitions,
generate_definition(name, params, returns)
)
table.sort(self.method_definitions)
self:listen_message(name, function(message, reply)
local ret = table.pack(method(table.unpack(message.data)))
reply.data = ret
end, "method")
end
---Broadcast a signal to running instances.
---@param destinations string | table<integer, string> | nil
---@param name string
---@vararg any signal_parameters
function IPC:signal(destinations, name, ...)
self:send_message(name, {
destinations = destinations,
data = table.pack(self.id, ...)
}, "signal")
end
---Call a method on another instance and wait for reply.
---@param destinations string | table<integer, string> | nil
---@param name string
---@return any | table<string,table> return_of_called_method
function IPC:call(destinations, name, ...)
local message_id = self:send_message(name, {
destinations = destinations,
data = table.pack(...)
}, "method")
local ret = nil
if message_id then
local replies = self:wait_for_replies(message_id)
if replies and #replies > 1 then
ret = {}
for _, reply in ipairs(replies) do
ret[reply.replier] = reply.data
end
elseif replies and #replies > 0 then
return table.unpack(replies[1].data)
end
else
core.error("IPC Error: could not make call to '%s'", name)
end
return ret
end
---Call a method on another instance asynchronously waiting for the replies.
---@param destinations string | table<integer, string> | nil
---@param name string
---@param callback fun(id: string, ret: table) | nil Called with the returned values
---@return string | nil message_id
function IPC:call_async(destinations, name, callback, ...)
return self:send_message(name, {
destinations = destinations,
data = table.pack(...),
on_reply = callback and function(reply)
callback(reply.replier, reply.data)
end or nil
}, "method")
end
---Main ipc session for current instance.
---@type plugins.ipc
local ipc = IPC()
---Get the IPC session for the running lite-xl instance.
---@return plugins.ipc
function IPC.current()
return ipc
end
--------------------------------------------------------------------------------
-- Override system.wait_event to allow ipc monitoring on the background.
--------------------------------------------------------------------------------
local system_wait_event = system.wait_event
local run_threads = coroutine.wrap(function()
while true do
for k, thread in pairs(threads) do
if thread.wake < system.get_time() then
local _, wait = assert(coroutine.resume(thread.cr))
if coroutine.status(thread.cr) == "dead" then
table.remove(threads, k)
elseif wait then
thread.wake = system.get_time() + wait
end
end
coroutine.yield()
end
end
end)
system.wait_event = function(timeout)
run_threads()
if not timeout then
if not system.window_has_focus() then
local t = system.get_time()
local h = 0.5 / 2
local dt = math.ceil(t / h) * h - t
return system_wait_event(dt + 1 / config.fps)
else
return system_wait_event()
end
else
return system_wait_event(timeout)
end
end
--------------------------------------------------------------------------------
-- Override system.show_fatal_error to be able and destroy session file on crash.
--------------------------------------------------------------------------------
local system_show_fatal_error = system.show_fatal_error
system.show_fatal_error = function(title, message)
if title == "Lite XL internal error" then
ipc:stop()
end
system_show_fatal_error(title, message)
end
--------------------------------------------------------------------------------
-- Override core.run to destroy ipc session file on exit.
--------------------------------------------------------------------------------
local core_run = core.run
core.run = function()
core_run()
ipc:stop()
end
--------------------------------------------------------------------------------
-- Override system.get_time temporarily as first function called on core.run
-- to allow settings gui to properly load ipc config options.
--------------------------------------------------------------------------------
local system_get_time = system.get_time
system.get_time = function()
if settings_found and settings and not settings.ui then
return system_get_time()
end
if config.plugins.ipc.single_instance then
system.get_time = system_get_time
local primary_instance = ipc:get_primary_instance()
if primary_instance and ARGS[2] then
local open_directory = false
for i=2, #ARGS do
local path = system.absolute_path(ARGS[i])
if path then
local path_info = system.get_file_info(path)
if path_info then
if path_info.type == "file" then
ipc:call_async(primary_instance, "core.open_file", nil, path)
else
if config.plugins.ipc.dirs_instance == "add" then
ipc:call_async(primary_instance, "core.open_directory", nil, path)
elseif config.plugins.ipc.dirs_instance == "change" then
ipc:call_async(primary_instance, "core.change_directory", nil, path)
else
if #ARGS > 2 then
system.exec(string.format("%q %q", EXEFILE, path))
else
open_directory = true
end
end
end
end
end
end
ipc:wait_for_messages()
if not open_directory then
os.exit()
end
end
else
system.get_time = system_get_time
end
return system_get_time()
end
--------------------------------------------------------------------------------
-- Register methods for opening files and directories.
--------------------------------------------------------------------------------
ipc:register_method("core.open_file", function(file)
if system.get_file_info(file) then
if system.raise_window then system.raise_window() end
core.root_view:open_doc(core.open_doc(file))
end
end, {{name = "file", type = "string"}})
ipc:register_method("core.open_directory", function(directory)
if system.get_file_info(directory) then
if system.raise_window then system.raise_window() end
core.add_project_directory(directory)
end
end, {{name = "directory", type = "string"}})
ipc:register_method("core.change_directory", function(directory)
if system.get_file_info(directory) then
if system.raise_window then system.raise_window() end
if directory == core.project_dir then return end
core.confirm_close_docs(core.docs, function(dirpath)
core.open_folder_project(dirpath)
end, directory)
end
end, {{name = "directory", type = "string"}})
--------------------------------------------------------------------------------
-- Register file dragging signals from instance to instance
--------------------------------------------------------------------------------
ipc:register_signal("core.tab_drag_start", {{name = "file", type = "string"}})
ipc:register_signal("core.tab_drag_stop")
ipc:register_signal("core.tab_drag_received", {{name = "file", type = "string"}})
local rootview_tab_dragging = false
local rootview_dragged_node = nil
local rootview_waiting_drop_file = ""
local rootview_waiting_drop_instance = ""
local rootview_on_mouse_moved = RootView.on_mouse_moved
function RootView:on_mouse_moved(x, y, dx, dy)
rootview_on_mouse_moved(self, x, y, dx, dy)
if
self.dragged_node and self.dragged_node.dragging
and
not rootview_tab_dragging
then
---@type core.doc
local doc = core.active_view.doc
if doc and doc.abs_filename then
rootview_tab_dragging = true
ipc:signal(nil, "core.tab_drag_start", doc.abs_filename)
rootview_dragged_node = self.dragged_node
end
elseif rootview_dragged_node then
local w, h, wx, wy = system.get_window_size()
if x < 0 or x > w or y < 0 or y > h then
self.dragged_node = nil
self:set_show_overlay(self.drag_overlay, false)
elseif not self.dragged_node then
self.dragged_node = rootview_dragged_node
self:set_show_overlay(self.drag_overlay, true)
end
core.request_cursor("hand")
elseif rootview_waiting_drop_file ~= "" then
ipc:signal(
rootview_waiting_drop_instance,
"core.tab_drag_received",
rootview_waiting_drop_file
)
core.root_view:open_doc(core.open_doc(rootview_waiting_drop_file))
rootview_waiting_drop_file = ""
end
end
local rootview_on_mouse_released = RootView.on_mouse_released
function RootView:on_mouse_released(button, x, y, ...)
rootview_on_mouse_released(self, button, x, y, ...)
if rootview_tab_dragging then
rootview_tab_dragging = false
rootview_dragged_node = nil
ipc:signal(nil, "core.tab_drag_stop")
end
end
ipc:listen_signal("core.tab_drag_start", function(instance, file)
rootview_waiting_drop_instance = instance
rootview_waiting_drop_file = file
end)
ipc:listen_signal("core.tab_drag_stop", function()
rootview_waiting_drop_instance = ""
rootview_waiting_drop_file = ""
end)
ipc:listen_signal("core.tab_drag_received", function()
command.perform("root:close")
end)
return IPC
