Handle large payload delta from upload event handler

Fix update chunk chaining logic; try/catch wasn't catching errors from the
async inner function.
This commit is contained in:
Masen Furer 2024-12-02 12:16:19 -08:00
parent 3cf32817f5
commit 0c1dc84872
No known key found for this signature in database
GPG Key ID: B0008AD22B3B3A95
2 changed files with 30 additions and 17 deletions

View File

@ -457,7 +457,7 @@ export const connect = async (
socket.current.on("reload", async (event) => {
event_processing = false;
queueEvents([...initialEvents(), JSON5.parse(event)], socket);
})
});
document.addEventListener("visibilitychange", checkVisibility);
};
@ -490,23 +490,30 @@ export const uploadFiles = async (
return false;
}
// Track how many partial updates have been processed for this upload.
let resp_idx = 0;
const eventHandler = (progressEvent) => {
// handle any delta / event streamed from the upload event handler
const event_callbacks = socket._callbacks.$event;
// Whenever called, responseText will contain the entire response so far.
const chunks = progressEvent.event.target.responseText.trim().split("\n");
// So only process _new_ chunks beyond resp_idx.
chunks.slice(resp_idx).map((chunk) => {
try {
socket._callbacks.$event.map((f) => {
f(chunk);
});
resp_idx += 1;
} catch (e) {
if (progressEvent.progress === 1) {
// Chunk may be incomplete, so only report errors when full response is available.
console.log("Error parsing chunk", chunk, e);
}
return;
}
event_callbacks.map((f, ix) => {
f(chunk)
.then(() => {
if (ix === event_callbacks.length - 1) {
// Mark this chunk as processed.
resp_idx += 1;
}
})
.catch((e) => {
if (progressEvent.progress === 1) {
// Chunk may be incomplete, so only report errors when full response is available.
console.log("Error parsing chunk", chunk, e);
}
return;
});
});
});
};
@ -711,7 +718,7 @@ export const useEventLoop = (
const combined_name = events.map((e) => e.name).join("+++");
if (event_actions?.temporal) {
if (!socket.current || !socket.current.connected) {
return; // don't queue when the backend is not connected
return; // don't queue when the backend is not connected
}
}
if (event_actions?.throttle) {
@ -852,7 +859,7 @@ export const useEventLoop = (
if (router.components[router.pathname].error) {
delete router.components[router.pathname].error;
}
}
};
router.events.on("routeChangeStart", change_start);
router.events.on("routeChangeComplete", change_complete);
router.events.on("routeChangeError", change_error);

View File

@ -19,11 +19,14 @@ def UploadFile():
import reflex as rx
LARGE_DATA = "DUMMY" * 1024 * 512
class UploadState(rx.State):
_file_data: Dict[str, str] = {}
event_order: List[str] = []
progress_dicts: List[dict] = []
disabled: bool = False
large_data: str = ""
async def handle_upload(self, files: List[rx.UploadFile]):
for file in files:
@ -34,6 +37,7 @@ def UploadFile():
for file in files:
upload_data = await file.read()
self._file_data[file.filename or ""] = upload_data.decode("utf-8")
self.large_data = LARGE_DATA
yield UploadState.chain_event
def upload_progress(self, progress):
@ -42,13 +46,15 @@ def UploadFile():
self.progress_dicts.append(progress)
def chain_event(self):
assert self.large_data == LARGE_DATA
self.large_data = ""
self.event_order.append("chain_event")
def index():
return rx.vstack(
rx.input(
value=UploadState.router.session.client_token,
is_read_only=True,
read_only=True,
id="token",
),
rx.heading("Default Upload"),