Uploads by PUT range requests rather than WS, remove dead code WS handlers #9

Merged
LeoVasanko merged 3 commits from put-uploads into main 2026-04-25 02:55:52 +00:00
4 changed files with 3 additions and 116 deletions
Showing only changes of commit c71efe5c0d - Show all commits
+1 -61
View File
@@ -1,5 +1,4 @@
import asyncio
import typing
from pathlib import PurePosixPath
from secrets import token_bytes
@@ -9,7 +8,7 @@ from sanic.exceptions import BadRequest
from cista import __version__, auth, config, sso, watching
from cista.fileio import FileServer
from cista.protocol import ControlTypes, FileRange, StatusMsg
from cista.protocol import ControlTypes, StatusMsg
from cista.util.apphelpers import asend, websocket_wrapper
bp = Blueprint("api", url_prefix="/api")
@@ -26,65 +25,6 @@ async def stop_fileserver(app):
await fileserver.stop()
@bp.websocket("upload")
@websocket_wrapper
async def upload(req, ws):
alink = fileserver.alink
while True:
req = None
text = await ws.recv()
if not isinstance(text, str):
raise ValueError(
f"Expected JSON control, got binary len(data) = {len(text)}",
)
req = msgspec.json.decode(text, type=FileRange)
pos = req.start
while True:
data = await ws.recv()
if not isinstance(data, bytes):
break
if len(data) > req.end - pos:
raise ValueError(
f"Expected up to {req.end - pos} bytes, got {len(data)} bytes"
)
sentsize = await alink(("upload", req.name, pos, data, req.size))
pos += typing.cast(int, sentsize)
if pos >= req.end:
break
if pos != req.end:
d = f"{len(data)} bytes" if isinstance(data, bytes) else data
raise ValueError(f"Expected {req.end - pos} more bytes, got {d}")
# Signal the watcher about the uploaded file and its parent directories
path = PurePosixPath(req.name)
watching.notify_change(path, *path.parents)
# Report success
res = StatusMsg(status="ack", req=req)
await asend(ws, res)
@bp.websocket("download")
@websocket_wrapper
async def download(req, ws):
alink = fileserver.alink
while True:
req = None
text = await ws.recv()
if not isinstance(text, str):
raise ValueError(
f"Expected JSON control, got binary len(data) = {len(text)}",
)
req = msgspec.json.decode(text, type=FileRange)
pos = req.start
while pos < req.end:
end = min(req.end, pos + (1 << 20))
data = typing.cast(bytes, await alink(("download", req.name, pos, end)))
await asend(ws, data)
pos += len(data)
# Report success
res = StatusMsg(status="ack", req=req)
await asend(ws, res)
@bp.websocket("control")
@websocket_wrapper
async def control(req, ws):
+1 -42
View File
@@ -1,10 +1,8 @@
import asyncio
import os
import threading
from cista import config
from cista.util import filename
from cista.util.asynclink import AsyncLink
from cista.util.lrucache import LRUCache
@@ -63,34 +61,12 @@ class File:
class FileServer:
async def start(self):
self.alink = AsyncLink()
self.worker = asyncio.get_event_loop().run_in_executor(
None,
self.worker_thread,
self.alink.to_sync,
)
self.cache = LRUCache(File, capacity=10, maxage=5.0)
self.cache_lock = threading.Lock()
self.file_locks: dict[str, threading.Lock] = {}
async def stop(self):
await self.alink.stop()
await self.worker
def worker_thread(self, slink):
try:
for req in slink:
with req as (command, *args):
if command == "upload":
req.set_result(self.upload(*args))
elif command == "upload_info":
req.set_result(self.upload_info(*args))
elif command == "download":
req.set_result(self.download(*args))
else:
raise NotImplementedError(f"Unhandled {command=} {args}")
finally:
self.cache.close()
self.cache.close()
@staticmethod
def _stat_size(path):
@@ -99,15 +75,6 @@ class FileServer:
except FileNotFoundError:
return None
def upload(self, name, pos, data, file_size):
name = filename.sanitize(name)
with self.cache_lock:
f = self.cache[name]
lock = self.file_locks.setdefault(name, threading.Lock())
with lock:
f.write(pos, data, file_size=file_size)
return len(data)
def upload_info(self, name, pos, data, file_size):
name = filename.sanitize(name)
with self.cache_lock:
@@ -123,11 +90,3 @@ class FileServer:
"size_before": size_before,
"size_after": size_after,
}
def download(self, name, start, end):
name = filename.sanitize(name)
with self.cache_lock:
f = self.cache[name]
lock = self.file_locks.setdefault(name, threading.Lock())
with lock:
return f[start:end]
+1 -12
View File
@@ -12,7 +12,6 @@ from cista.util import filename
## Control commands
class ControlBase(msgspec.Struct, tag_field="op", tag=str.lower):
def __call__(self):
raise NotImplementedError
@@ -118,19 +117,9 @@ class Cp(ControlBase):
ControlTypes = MkDir | Rename | Rm | Mv | Cp
## File uploads and downloads
class FileRange(msgspec.Struct):
name: str
size: int
start: int
end: int
class StatusMsg(msgspec.Struct):
status: str
req: FileRange
req: Any
class ErrorMsg(msgspec.Struct):
-1
View File
@@ -3,7 +3,6 @@ import { showAuthIframe, AuthCancelledError, isAuthIframeOpen } from 'paskia'
import type { FileEntry, UpdateEntry, errorEvent } from "./Document"
export const controlUrl = '/api/control'
export const uploadUrl = '/api/upload'
export const watchUrl = '/api/watch'
let tree = [] as FileEntry[]