2
0
mirror of https://github.com/koreader/koreader synced 2024-11-04 12:00:25 +00:00
koreader/frontend/ui/message/streammessagequeue.lua
NiLuJe fadee1f5dc
Clarify our OOP semantics across the codebase (#9586)
Basically:

* Use `extend` for class definitions
* Use `new` for object instantiations

That includes some minor code cleanups along the way:

* Updated `Widget`'s docs to make the semantics clearer.
* Removed `should_restrict_JIT` (it's been dead code since https://github.com/koreader/android-luajit-launcher/pull/283)
* Minor refactoring of LuaSettings/LuaData/LuaDefaults/DocSettings to behave (mostly, they are instantiated via `open` instead of `new`) like everything else and handle inheritance properly (i.e., DocSettings is now a proper LuaSettings subclass).
* Default to `WidgetContainer` instead of `InputContainer` for stuff that doesn't actually setup key/gesture events.
* Ditto for explicit `*Listener` only classes, make sure they're based on `EventListener` instead of something uselessly fancier.
* Unless absolutely necessary, do not store references in class objects, ever; only values. Instead, always store references in instances, to avoid both sneaky inheritance issues, and sneaky GC pinning of stale references.
  * ReaderUI: Fix one such issue with its `active_widgets` array, with critical implications, as it essentially pinned *all* of ReaderUI's modules, including their reference to the `Document` instance (i.e., that was a big-ass leak).
* Terminal: Make sure the shell is killed on plugin teardown.
* InputText: Fix Home/End/Del physical keys to behave sensibly.
* InputContainer/WidgetContainer: If necessary, compute self.dimen at paintTo time (previously, only InputContainers did, which might have had something to do with random widgets unconcerned about input using it as a baseclass instead of WidgetContainer...).
* OverlapGroup: Compute self.dimen at *init* time, because for some reason it needs to do that, but do it directly in OverlapGroup instead of going through a weird WidgetContainer method that it was the sole user of.
* ReaderCropping: Under no circumstances should a Document instance member (here, self.bbox) risk being `nil`ed!
* Kobo: Minor code cleanups.
2022-10-06 02:14:48 +02:00

98 lines
3.2 KiB
Lua

local ffi = require("ffi")
local logger = require("logger")
local MessageQueue = require("ui/message/messagequeue")
local _ = require("ffi/zeromq_h")
local zmq = ffi.load("libs/libzmq.so.4")
local czmq = ffi.load("libs/libczmq.so.1")
local C = ffi.C
local StreamMessageQueue = MessageQueue:extend{
host = nil,
port = nil,
}
function StreamMessageQueue:start()
self.context = czmq.zctx_new()
self.socket = czmq.zsocket_new(self.context, C.ZMQ_STREAM)
self.poller = czmq.zpoller_new(self.socket, nil)
local endpoint = string.format("tcp://%s:%d", self.host, self.port)
logger.warn("connect to endpoint", endpoint)
local rc = czmq.zsocket_connect(self.socket, endpoint)
if rc ~= 0 then
error("cannot connect to " .. endpoint)
end
local id_size = ffi.new("size_t[1]", 256)
local buffer = ffi.new("uint8_t[?]", id_size[0])
--- @todo: Check return of zmq_getsockopt()
zmq.zmq_getsockopt(self.socket, C.ZMQ_IDENTITY, buffer, id_size)
self.id = ffi.string(buffer, id_size[0])
logger.dbg("id", #self.id, self.id)
end
function StreamMessageQueue:stop()
if self.poller ~= nil then
czmq.zpoller_destroy(ffi.new('zpoller_t *[1]', self.poller))
end
if self.socket ~= nil then
czmq.zsocket_destroy(self.context, self.socket)
end
if self.context ~= nil then
czmq.zctx_destroy(ffi.new('zctx_t *[1]', self.context))
end
end
function StreamMessageQueue:handleZframe(frame)
local size = czmq.zframe_size(frame)
local data = nil
if size > 0 then
local frame_data = czmq.zframe_data(frame)
if frame_data ~= nil then
data = ffi.string(frame_data, size)
end
end
czmq.zframe_destroy(ffi.new('zframe_t *[1]', frame))
return data
end
function StreamMessageQueue:waitEvent()
-- Successive zframes may come in batches of tens or hundreds in some cases.
-- Since we buffer each frame's data in a Lua string,
-- and then let the caller concatenate those,
-- it may consume a significant amount of memory.
-- And it's fairly easy to trigger when receiving file data from Calibre.
-- So, throttle reception to 256 packages at most in one waitEvent loop,
-- after which we immediately call receiveCallback.
local wait_packages = 256
-- In a similar spirit, much like LuaSocket,
-- we store the data as ropes of strings in an array,
-- to be concatenated by the caller.
local t = {}
while czmq.zpoller_wait(self.poller, 0) ~= nil and wait_packages > 0 do
local id_frame = czmq.zframe_recv(self.socket)
if id_frame ~= nil then
self:handleZframe(id_frame)
end
local frame = czmq.zframe_recv(self.socket)
if frame ~= nil then
local data = self:handleZframe(frame)
if data then
table.insert(t, data)
end
end
wait_packages = wait_packages - 1
end
if self.receiveCallback and #t ~= 0 then
self.receiveCallback(t)
end
end
function StreamMessageQueue:send(data)
local msg = czmq.zmsg_new()
czmq.zmsg_addmem(msg, self.id, #self.id)
czmq.zmsg_addmem(msg, data, #data)
czmq.zmsg_send(ffi.new('zmsg_t *[1]', msg), self.socket)
end
return StreamMessageQueue