WinAPI Named Pipes + Effil

[chapo]

Версия MoonLoader

.026-beta

Появилась потребность кидать данные между процессами без запуска локального сервера, решил сделать через Named Pipes и effil для ассинхронности.
Все работает, но есть проблема с получением сообщений от клиента. "Сервер" видит сообщения от клиента только после того как сам отправляет сообщение.
Есть идеи где я надристал?

Сообщения от сервер -> клиент доходят сразу

Сообщение от клиента появилось в игре только после отправки еще одного от сервера

lib

---@class NPipes.Server
---@field path string
---@field onUpdate? fun(self: NPipes.Server, message: string)
---@field started boolean
---@field effil {channelIn: unknown, channelOut: unknown, thread: unknown}
---@field start fun(self: NPipes.Server, path?: string): NPipes.Error?
---@field close fun(self: NPipes.Server)
---@field onMessage fun(self: NPipes.Server, callback: fun(message: string))
---@field onError fun(self: NPipes.Server, callback: fun(err: NPipes.Error))
---@field onConnect fun(self: NPipes.Server, callback: fun())
---@field onClose fun(self: NPipes.Server, callback: fun())

local Server = {}
Server.__index = Server

---@param path? string Server path. Example: "\\\\.\\pipe\\myPipeName"
---@param serverAccessMode? NPipes.ServerAccess
---@return NPipes.Server
function Server.new(path, serverAccessMode)
    -- assert(type(path) == "string", "Server path must be a string, got" .. type(path))
    
    local self = setmetatable({}, Server)
    self.serverAccessMode = serverAccessMode or NPipes.ServerAccess.PIPE_ACCESS_DUPLEX
    self.path = path
    self.started = false
    self.callback = {}
    self.effil = nil
    return self
end

local function call(fn, ...)
    if (type(fn) == "function") then
        fn(...)
    end
end

function Server:onMessage(cb) self.callback.onMessage = cb end
function Server:onError(cb) self.callback.onError = cb end
function Server:onConnect(cb) self.callback.onConnect = cb end
function Server:onClose(cb) self.callback.onClose = cb end

---@class ChannelMessage
---@field type "stateChange" | "message" | "error"
---@field data string

local function serverThread(path, channelIn, channelOut)
    local function log(...)
        channelIn:push({ type = "log", data = table.concat({ ... }, ' ') })
    end
    local ffi = require("ffi")
    ffi.cdef [[
        typedef unsigned long DWORD;
        typedef void* HANDLE;
        typedef int BOOL;
        HANDLE CreateNamedPipeA(const char*, DWORD, DWORD, DWORD, DWORD, DWORD, DWORD, void*);
        BOOL ConnectNamedPipe(HANDLE, void*);
        BOOL ReadFile(HANDLE, void*, DWORD, DWORD*, void*);
        BOOL WriteFile(HANDLE, const void*, DWORD, DWORD*, void*);
        BOOL FlushFileBuffers(HANDLE);
        BOOL DisconnectNamedPipe(HANDLE);
        BOOL CloseHandle(HANDLE);
        DWORD GetLastError();
        BOOL PeekNamedPipe(HANDLE, void*, DWORD, DWORD*, DWORD*, DWORD*);
    ]]
    
    local PIPE_ACCESS_DUPLEX = 0x00000003
    local PIPE_TYPE_BYTE = 0x00000000
    local INVALID_PIPE_HANDLE = ffi.cast("HANDLE", -1)
    -- channelIn:push({ type = "connected" })
    
    log("start")
    -- Create pipe
    local handle = ffi.C.CreateNamedPipeA(path, PIPE_ACCESS_DUPLEX, PIPE_TYPE_BYTE, 1, 4096, 4096, 0, nil)
    if (handle == INVALID_PIPE_HANDLE) then
        log("CreateNamedPipeA failed:", ffi.C.GetLastError())
        channelIn:push({ type = "error", data = ffi.C.GetLastError() })
        return
    end
    log("pipe created")
    if (ffi.C.ConnectNamedPipe(handle, nil) == 0) then
        log("ConnectNamedPipe failed:", ffi.C.GetLastError())
        channelIn:push({ type = "error", data = ffi.C.GetLastError() })
        ffi.C.CloseHandle(handle)
        return
    end
    log("pipe connected")

    channelIn:push({ type = "started" })

    -- refresh pipe
    local buf = ffi.new("char[4096]")
    local bytesRead = ffi.new("DWORD[1]")
    local bytesWritten = ffi.new("DWORD[1]")
    local totalAvail = ffi.new("DWORD[1]")
    while (true) do
        local cmd = channelOut:pop(0)
        if (cmd) then
            log("loop cmd not null", cmd.type, cmd.data)
            
            if (cmd.type == "send") then
                local data = cmd.data .. "\n"
                ffi.C.WriteFile(handle, data, #data, bytesWritten, nil)
                ffi.C.FlushFileBuffers(handle)
                log("tried to send msg")
            elseif (cmd.type == "quit") then
                break
            end
        end
        if (ffi.C.PeekNamedPipe(handle, nil, 0, nil, totalAvail, nil) ~= 0 and totalAvail[0] > 0) then
            if (ffi.C.ReadFile(handle, buf, 4096, bytesRead, nil) ~= 0) then
                local msg = ffi.string(buf, bytesRead[0])
                channelIn:push({ type = "message", data = msg })
                ffi.C.FlushFileBuffers(handle)

                if (msg == "quit") then break end
            end
        end
    end
    ffi.C.DisconnectNamedPipe(handle)
    ffi.C.CloseHandle(handle)
    channelIn:push({ type = "closed" })
end

---@param path? string
---@return NPipe.Error | nil
function Server:start(path)
    if (not self.path) then
        if (not path) then
            return error("Path is not provided. Use NPipes.Server.new(PATH) or server:start(PATH)")
        end
        self.path = path
    end
    if (self.started) then
        return call(self.callback.onError, "Server already started")
    end
    self.effil = {}
    self.effil.channelIn = effil.channel(10)
    self.effil.channelOut = effil.channel(10)
    self.effil.thread = effil.thread(serverThread)(self.path, self.effil.channelIn, self.effil.channelOut)
end

function Server:update()
    if (not self.effil or not self.effil.channelIn) then return end
    local msg = self.effil.channelIn:pop(0)
    if (msg) then
        sampAddChatMessage(("Server:update(): %s | %s"):format(msg.type, msg.data), -1)
        if (msg.type == "message" and self.callback) then
            call(self.callback.onMessage, msg.data)
        elseif (msg.type == "error") then
            call(self.callback.onError, tonumber(msg.data))
        elseif (msg.type == "started") then
            call(self.callback.onConnect)
            self.started = true
        elseif (msg.type == "closed") then
            call(self.callback.onClose)
            self.started = false
        elseif (msg.type == "log") then
            print("LOG", msg.data)
        end
    end
end

function Server:send(message)
    if (not self.effil or not self.effil.channelOut) then return end
    self.effil.channelOut:push({ type = "send", data = message })
end

function Server:close()
    if (not self.effil or not self.effil.channelOut) then return end
    self.effil.channelOut:push({ type = "quit" })
    -- self.effil = nil
    -- self.started = false
end

return Server

script

local NPipes = require("npipes")

local server = NPipes.Server.new()

server:onMessage(function(message)
    sampAddChatMessage("[PIPE] Message received: " .. message, -1)
end)

server:onError(function(error)
    sampAddChatMessage(("[PIPE] Error: %s (%s)"):format(error, NPipes:errorName(error)), -1)
end)

server:onClose(function()
    sampAddChatMessage("[PIPE] Pipe was closed", -1)
end)

server:onConnect(function()
    sampAddChatMessage("[PIPE] Connected", -1)
end)


function main()
    while not isSampAvailable() do wait(0) end
    sampRegisterChatCommand("pipe.start", function(path)
        local err = server:start("\\\\.\\pipe\\lua_" .. (#path > 0 and path or os.clock()))
        sampAddChatMessage("Starting pipe server with name:" .. server.path, -1)
        if (err) then
            print("Error starting pipe:", NPipes.Error[err])
        end
        
    end)
    sampRegisterChatCommand("pipe.send", function(msg)
        server:send(msg)
        print("[PIPE] Sent:", msg)
    end)
    sampRegisterChatCommand("pipe.close", function(msg)
        sampAddChatMessage("Closing pipe with name:" .. server.path, -1)
        server:close()
    end)
    while true do
        wait(0)
        server:update()
    end
end

function onScriptTerminate(scr)
    if (scr == thisScript()) then
        server:close()
        print("Pipe", server.path, "closed!")
    end
end

client

package main

import (
    "bufio"
    "fmt"
    "os"
)

const pipePath = `\\.\pipe\lua_pizdazalupaochko`

func main() {
    fmt.Println("Started")
    pipe, err := os.OpenFile(pipePath, os.O_RDWR, 0666)
    if err != nil {
        fmt.Println("Error:", err)
        return
    }
    defer pipe.Close()

    go func() {
        scanner := bufio.NewScanner(pipe)
        for scanner.Scan() {
            fmt.Println("Received:", scanner.Text())
        }
    }()

    scanner := bufio.NewScanner(os.Stdin)
    for scanner.Scan() {
        msg := scanner.Text()
        if msg == "exit" {
            break
        }
        pipe.Write([]byte(msg + "\n"))
    }
}

 

[Akionka]

думаю, что связано с тем, что гошный сканер ждет \n в конце строчки. но тут надо нейроныча спрашивать

 

[chromiusj]

Проблема где-то в потоках. Лучше такую прокладку делать было бы не на луашке конечно.
Попробуй в serverThread прокинуть это, после того, как подключаешь пайплайн, и протестить:

local mode = ffi.new("DWORD[1]", PIPE_READMODE_BYTE + PIPE_NOWAIT)
    if ffi.C.SetNamedPipeHandleState(handle, mode, nil, nil) == 0 then
        channelIn:push({ type = "log", data = "SetNamedPipeHandleState failed: " .. ffi.C.GetLastError() })
    end