feat: narrow thread sync context and dedupe realtime refresh
This commit is contained in:
@@ -3,6 +3,7 @@ package com.hyzq.boss;
|
||||
import androidx.annotation.Nullable;
|
||||
|
||||
import org.json.JSONException;
|
||||
import org.json.JSONArray;
|
||||
import org.json.JSONObject;
|
||||
|
||||
import java.io.BufferedReader;
|
||||
@@ -10,6 +11,9 @@ import java.io.IOException;
|
||||
import java.io.InputStream;
|
||||
import java.io.InputStreamReader;
|
||||
import java.net.HttpURLConnection;
|
||||
import java.util.ArrayList;
|
||||
import java.util.Collections;
|
||||
import java.util.Iterator;
|
||||
import java.nio.charset.StandardCharsets;
|
||||
|
||||
final class BossRealtimeClient {
|
||||
@@ -19,6 +23,7 @@ final class BossRealtimeClient {
|
||||
|
||||
private final BossApiClient apiClient;
|
||||
private final Listener listener;
|
||||
private static final String HEARTBEAT_EVENT_NAME = "heartbeat";
|
||||
private volatile boolean running;
|
||||
private @Nullable Thread workerThread;
|
||||
private @Nullable HttpURLConnection activeConnection;
|
||||
@@ -144,6 +149,9 @@ final class BossRealtimeClient {
|
||||
if (eventName.isEmpty()) {
|
||||
return null;
|
||||
}
|
||||
if (HEARTBEAT_EVENT_NAME.equals(eventName)) {
|
||||
return null;
|
||||
}
|
||||
JSONObject payload = new JSONObject();
|
||||
if (dataBuilder.length() > 0) {
|
||||
try {
|
||||
@@ -151,7 +159,65 @@ final class BossRealtimeClient {
|
||||
} catch (JSONException ignored) {
|
||||
payload = new JSONObject();
|
||||
}
|
||||
} else {
|
||||
return null;
|
||||
}
|
||||
return new BossRealtimeEvent(eventName, payload);
|
||||
}
|
||||
|
||||
static String buildEventFingerprint(@Nullable BossRealtimeEvent event) {
|
||||
if (event == null || event.eventName.isEmpty()) {
|
||||
return "";
|
||||
}
|
||||
return event.eventName + "|" + canonicalizeJson(event.payload);
|
||||
}
|
||||
|
||||
private static String canonicalizeJson(@Nullable Object value) {
|
||||
if (value == null || value == JSONObject.NULL) {
|
||||
return "null";
|
||||
}
|
||||
if (value instanceof JSONObject) {
|
||||
JSONObject object = (JSONObject) value;
|
||||
ArrayList<String> keys = new ArrayList<>();
|
||||
Iterator<String> iterator = object.keys();
|
||||
while (iterator.hasNext()) {
|
||||
String key = iterator.next();
|
||||
if (!"at".equals(key)) {
|
||||
keys.add(key);
|
||||
}
|
||||
}
|
||||
Collections.sort(keys);
|
||||
StringBuilder builder = new StringBuilder("{");
|
||||
for (int index = 0; index < keys.size(); index += 1) {
|
||||
if (index > 0) {
|
||||
builder.append(',');
|
||||
}
|
||||
String key = keys.get(index);
|
||||
builder.append(JSONObject.quote(key));
|
||||
builder.append(':');
|
||||
builder.append(canonicalizeJson(object.opt(key)));
|
||||
}
|
||||
builder.append('}');
|
||||
return builder.toString();
|
||||
}
|
||||
if (value instanceof JSONArray) {
|
||||
JSONArray array = (JSONArray) value;
|
||||
StringBuilder builder = new StringBuilder("[");
|
||||
for (int index = 0; index < array.length(); index += 1) {
|
||||
if (index > 0) {
|
||||
builder.append(',');
|
||||
}
|
||||
builder.append(canonicalizeJson(array.opt(index)));
|
||||
}
|
||||
builder.append(']');
|
||||
return builder.toString();
|
||||
}
|
||||
if (value instanceof String) {
|
||||
return JSONObject.quote((String) value);
|
||||
}
|
||||
if (value instanceof Number || value instanceof Boolean) {
|
||||
return String.valueOf(value);
|
||||
}
|
||||
return JSONObject.quote(String.valueOf(value));
|
||||
}
|
||||
}
|
||||
|
||||
@@ -102,7 +102,7 @@ public class MainActivity extends AppCompatActivity {
|
||||
private boolean conversationQuickActionsVisible = false;
|
||||
private boolean conversationAutoRefreshArmed = false;
|
||||
private boolean conversationAutoRefreshEnabled = false;
|
||||
private long lastRealtimeRefreshAt = 0L;
|
||||
private final java.util.HashMap<String, Long> recentRealtimeEventTimestamps = new java.util.HashMap<>();
|
||||
private final Set<String> selectedConversationProjectIds = new LinkedHashSet<>();
|
||||
private @Nullable RootPagerAdapter rootPagerAdapter;
|
||||
private boolean syncingRootPagerSelection = false;
|
||||
@@ -372,30 +372,73 @@ public class MainActivity extends AppCompatActivity {
|
||||
}
|
||||
boolean shouldRefresh = false;
|
||||
if ("conversations".equals(activeTab)) {
|
||||
shouldRefresh =
|
||||
"conversation.updated".equals(event.eventName) ||
|
||||
"project.messages.updated".equals(event.eventName) ||
|
||||
"master_agent.task.updated".equals(event.eventName) ||
|
||||
"conversation.context_indicator.updated".equals(event.eventName);
|
||||
shouldRefresh = shouldRefreshConversationsTab(event);
|
||||
} else if ("devices".equals(activeTab)) {
|
||||
shouldRefresh =
|
||||
"devices.updated".equals(event.eventName) ||
|
||||
"devices.skills.updated".equals(event.eventName) ||
|
||||
"conversation.updated".equals(event.eventName);
|
||||
shouldRefresh = shouldRefreshDevicesTab(event);
|
||||
} else if ("me".equals(activeTab)) {
|
||||
shouldRefresh = "ota.updated".equals(event.eventName) || "app.logs.updated".equals(event.eventName);
|
||||
}
|
||||
if (!shouldRefresh) {
|
||||
return;
|
||||
}
|
||||
long now = System.currentTimeMillis();
|
||||
if (now - lastRealtimeRefreshAt < REALTIME_REFRESH_THROTTLE_MS) {
|
||||
String eventFingerprint = BossRealtimeClient.buildEventFingerprint(event);
|
||||
if (eventFingerprint.isEmpty()) {
|
||||
return;
|
||||
}
|
||||
long now = System.currentTimeMillis();
|
||||
if (isDuplicateRealtimeEvent(eventFingerprint, now)) {
|
||||
return;
|
||||
}
|
||||
lastRealtimeRefreshAt = now;
|
||||
runOnUiThread(this::refreshCurrentTab);
|
||||
}
|
||||
|
||||
private boolean isDuplicateRealtimeEvent(String eventFingerprint, long now) {
|
||||
pruneRecentRealtimeEvents(now);
|
||||
Long previousEventAt = recentRealtimeEventTimestamps.get(eventFingerprint);
|
||||
if (previousEventAt != null && now - previousEventAt < REALTIME_REFRESH_THROTTLE_MS) {
|
||||
return true;
|
||||
}
|
||||
recentRealtimeEventTimestamps.put(eventFingerprint, now);
|
||||
return false;
|
||||
}
|
||||
|
||||
private void pruneRecentRealtimeEvents(long now) {
|
||||
java.util.Iterator<java.util.Map.Entry<String, Long>> iterator = recentRealtimeEventTimestamps.entrySet().iterator();
|
||||
while (iterator.hasNext()) {
|
||||
java.util.Map.Entry<String, Long> entry = iterator.next();
|
||||
if (now - entry.getValue() >= REALTIME_REFRESH_THROTTLE_MS) {
|
||||
iterator.remove();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private boolean shouldRefreshConversationsTab(BossRealtimeEvent event) {
|
||||
if (!hasProjectId(event)) {
|
||||
return false;
|
||||
}
|
||||
return "conversation.updated".equals(event.eventName)
|
||||
|| "project.messages.updated".equals(event.eventName)
|
||||
|| "master_agent.task.updated".equals(event.eventName)
|
||||
|| "conversation.context_indicator.updated".equals(event.eventName);
|
||||
}
|
||||
|
||||
private boolean shouldRefreshDevicesTab(BossRealtimeEvent event) {
|
||||
if (!hasDeviceId(event)) {
|
||||
return false;
|
||||
}
|
||||
return "devices.updated".equals(event.eventName)
|
||||
|| "devices.skills.updated".equals(event.eventName)
|
||||
|| "conversation.updated".equals(event.eventName);
|
||||
}
|
||||
|
||||
private boolean hasProjectId(BossRealtimeEvent event) {
|
||||
return event != null && !event.payload.optString("projectId", "").trim().isEmpty();
|
||||
}
|
||||
|
||||
private boolean hasDeviceId(BossRealtimeEvent event) {
|
||||
return event != null && !event.payload.optString("deviceId", "").trim().isEmpty();
|
||||
}
|
||||
|
||||
private void refreshAllData(@Nullable JSONObject initialSession) {
|
||||
startRefreshing(true);
|
||||
topSubtitle.setText("");
|
||||
|
||||
@@ -84,7 +84,7 @@ public class ProjectDetailActivity extends BossScreenActivity {
|
||||
private ActivityResultLauncher<String> filePickerLauncher;
|
||||
private final ExecutorService replyWaitExecutor = Executors.newSingleThreadExecutor();
|
||||
private @Nullable BossRealtimeClient realtimeClient;
|
||||
private long lastRealtimeReloadAt;
|
||||
private final java.util.HashMap<String, Long> recentRealtimeEventTimestamps = new java.util.HashMap<>();
|
||||
|
||||
static final class ChromeBindings {
|
||||
final boolean multiSelecting;
|
||||
@@ -276,24 +276,51 @@ public class ProjectDetailActivity extends BossScreenActivity {
|
||||
if (event == null || event.eventName.isEmpty() || projectId == null || projectId.isEmpty()) {
|
||||
return;
|
||||
}
|
||||
boolean shouldReload = false;
|
||||
String payloadProjectId = event.payload.optString("projectId", "");
|
||||
if ("project.messages.updated".equals(event.eventName) || "conversation.updated".equals(event.eventName)) {
|
||||
shouldReload = projectId.equals(payloadProjectId);
|
||||
} else if ("master_agent.task.updated".equals(event.eventName)) {
|
||||
shouldReload = "master-agent".equals(projectId);
|
||||
}
|
||||
boolean shouldReload = shouldReloadForRealtimeEvent(event);
|
||||
if (!shouldReload) {
|
||||
return;
|
||||
}
|
||||
long now = System.currentTimeMillis();
|
||||
if (now - lastRealtimeReloadAt < REALTIME_RELOAD_THROTTLE_MS) {
|
||||
String eventFingerprint = BossRealtimeClient.buildEventFingerprint(event);
|
||||
if (eventFingerprint.isEmpty()) {
|
||||
return;
|
||||
}
|
||||
long now = System.currentTimeMillis();
|
||||
if (isDuplicateRealtimeEvent(eventFingerprint, now)) {
|
||||
return;
|
||||
}
|
||||
lastRealtimeReloadAt = now;
|
||||
runOnUiThread(this::triggerRealtimeReload);
|
||||
}
|
||||
|
||||
private boolean isDuplicateRealtimeEvent(String eventFingerprint, long now) {
|
||||
pruneRecentRealtimeEvents(now);
|
||||
Long previousEventAt = recentRealtimeEventTimestamps.get(eventFingerprint);
|
||||
if (previousEventAt != null && now - previousEventAt < REALTIME_RELOAD_THROTTLE_MS) {
|
||||
return true;
|
||||
}
|
||||
recentRealtimeEventTimestamps.put(eventFingerprint, now);
|
||||
return false;
|
||||
}
|
||||
|
||||
private void pruneRecentRealtimeEvents(long now) {
|
||||
java.util.Iterator<java.util.Map.Entry<String, Long>> iterator = recentRealtimeEventTimestamps.entrySet().iterator();
|
||||
while (iterator.hasNext()) {
|
||||
java.util.Map.Entry<String, Long> entry = iterator.next();
|
||||
if (now - entry.getValue() >= REALTIME_RELOAD_THROTTLE_MS) {
|
||||
iterator.remove();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private boolean shouldReloadForRealtimeEvent(BossRealtimeEvent event) {
|
||||
String payloadProjectId = event.payload.optString("projectId", "").trim();
|
||||
if (payloadProjectId.isEmpty() || !payloadProjectId.equals(projectId)) {
|
||||
return false;
|
||||
}
|
||||
return "project.messages.updated".equals(event.eventName)
|
||||
|| "conversation.updated".equals(event.eventName)
|
||||
|| "master_agent.task.updated".equals(event.eventName);
|
||||
}
|
||||
|
||||
void triggerRealtimeReload() {
|
||||
reload();
|
||||
}
|
||||
|
||||
@@ -27,4 +27,14 @@ public class BossRealtimeClientTest {
|
||||
public void parseEventBlockReturnsNullForKeepaliveComment() {
|
||||
assertNull(BossRealtimeClient.parseEventBlock(": keepalive\n\n"));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void parseEventBlockIgnoresHeartbeatControlEvents() {
|
||||
assertNull(BossRealtimeClient.parseEventBlock("event: heartbeat\n\n"));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void parseEventBlockReturnsNullForEmptyEventPayloads() {
|
||||
assertNull(BossRealtimeClient.parseEventBlock("event: conversation.updated\n\n"));
|
||||
}
|
||||
}
|
||||
|
||||
@@ -7,6 +7,7 @@ import org.junit.Test;
|
||||
import org.junit.runner.RunWith;
|
||||
import org.robolectric.Robolectric;
|
||||
import org.robolectric.RobolectricTestRunner;
|
||||
import org.robolectric.Shadows;
|
||||
import org.robolectric.annotation.Config;
|
||||
import org.robolectric.util.ReflectionHelpers;
|
||||
|
||||
@@ -25,6 +26,7 @@ public class MainActivityRealtimeTest {
|
||||
new BossRealtimeEvent("conversation.updated", new JSONObject().put("projectId", "project-1"))
|
||||
)
|
||||
);
|
||||
Shadows.shadowOf(activity.getMainLooper()).idle();
|
||||
|
||||
assertEquals(1, activity.refreshCount);
|
||||
}
|
||||
@@ -41,10 +43,79 @@ public class MainActivityRealtimeTest {
|
||||
new BossRealtimeEvent("devices.updated", new JSONObject().put("deviceId", "mac-studio"))
|
||||
)
|
||||
);
|
||||
Shadows.shadowOf(activity.getMainLooper()).idle();
|
||||
|
||||
assertEquals(0, activity.refreshCount);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void blankProjectIdConversationEventDoesNotRefreshVisibleConversationTab() throws Exception {
|
||||
TestMainActivity activity = Robolectric.buildActivity(TestMainActivity.class).setup().resume().get();
|
||||
ReflectionHelpers.callInstanceMethod(activity, "showContent");
|
||||
ReflectionHelpers.callInstanceMethod(
|
||||
activity,
|
||||
"handleRealtimeEvent",
|
||||
ReflectionHelpers.ClassParameter.from(
|
||||
BossRealtimeEvent.class,
|
||||
new BossRealtimeEvent("conversation.updated", new JSONObject())
|
||||
)
|
||||
);
|
||||
Shadows.shadowOf(activity.getMainLooper()).idle();
|
||||
|
||||
assertEquals(0, activity.refreshCount);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void contextIndicatorEventRefreshesVisibleConversationTab() throws Exception {
|
||||
TestMainActivity activity = Robolectric.buildActivity(TestMainActivity.class).setup().resume().get();
|
||||
ReflectionHelpers.callInstanceMethod(activity, "showContent");
|
||||
ReflectionHelpers.callInstanceMethod(
|
||||
activity,
|
||||
"handleRealtimeEvent",
|
||||
ReflectionHelpers.ClassParameter.from(
|
||||
BossRealtimeEvent.class,
|
||||
new BossRealtimeEvent(
|
||||
"conversation.context_indicator.updated",
|
||||
new JSONObject().put("projectId", "project-1")
|
||||
)
|
||||
)
|
||||
);
|
||||
Shadows.shadowOf(activity.getMainLooper()).idle();
|
||||
|
||||
assertEquals(1, activity.refreshCount);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void distinctConversationEventsBackToBackBothRefreshVisibleConversationTab() throws Exception {
|
||||
TestMainActivity activity = Robolectric.buildActivity(TestMainActivity.class).setup().resume().get();
|
||||
ReflectionHelpers.callInstanceMethod(activity, "showContent");
|
||||
ReflectionHelpers.callInstanceMethod(
|
||||
activity,
|
||||
"handleRealtimeEvent",
|
||||
ReflectionHelpers.ClassParameter.from(
|
||||
BossRealtimeEvent.class,
|
||||
new BossRealtimeEvent(
|
||||
"conversation.updated",
|
||||
new JSONObject().put("projectId", "project-1").put("at", "2026-04-05T10:00:00.000Z")
|
||||
)
|
||||
)
|
||||
);
|
||||
ReflectionHelpers.callInstanceMethod(
|
||||
activity,
|
||||
"handleRealtimeEvent",
|
||||
ReflectionHelpers.ClassParameter.from(
|
||||
BossRealtimeEvent.class,
|
||||
new BossRealtimeEvent(
|
||||
"project.messages.updated",
|
||||
new JSONObject().put("projectId", "project-1").put("at", "2026-04-05T10:00:00.500Z")
|
||||
)
|
||||
)
|
||||
);
|
||||
Shadows.shadowOf(activity.getMainLooper()).idle();
|
||||
|
||||
assertEquals(2, activity.refreshCount);
|
||||
}
|
||||
|
||||
public static class TestMainActivity extends MainActivity {
|
||||
int refreshCount;
|
||||
|
||||
|
||||
@@ -9,6 +9,7 @@ import org.junit.Test;
|
||||
import org.junit.runner.RunWith;
|
||||
import org.robolectric.Robolectric;
|
||||
import org.robolectric.RobolectricTestRunner;
|
||||
import org.robolectric.Shadows;
|
||||
import org.robolectric.annotation.Config;
|
||||
import org.robolectric.util.ReflectionHelpers;
|
||||
|
||||
@@ -34,6 +35,7 @@ public class ProjectDetailActivityRealtimeTest {
|
||||
new BossRealtimeEvent("project.messages.updated", new JSONObject().put("projectId", "project-1"))
|
||||
)
|
||||
);
|
||||
Shadows.shadowOf(activity.getMainLooper()).idle();
|
||||
|
||||
assertEquals(1, activity.reloadCount);
|
||||
}
|
||||
@@ -57,10 +59,111 @@ public class ProjectDetailActivityRealtimeTest {
|
||||
new BossRealtimeEvent("project.messages.updated", new JSONObject().put("projectId", "project-2"))
|
||||
)
|
||||
);
|
||||
Shadows.shadowOf(activity.getMainLooper()).idle();
|
||||
|
||||
assertEquals(0, activity.reloadCount);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void masterAgentTaskEventDoesNotRefreshForDifferentProjectId() throws Exception {
|
||||
Intent intent = new Intent()
|
||||
.putExtra(ProjectDetailActivity.EXTRA_PROJECT_ID, "master-agent")
|
||||
.putExtra(ProjectDetailActivity.EXTRA_PROJECT_NAME, "主 Agent");
|
||||
TestRealtimeProjectDetailActivity activity = Robolectric
|
||||
.buildActivity(TestRealtimeProjectDetailActivity.class, intent)
|
||||
.setup()
|
||||
.resume()
|
||||
.get();
|
||||
|
||||
ReflectionHelpers.callInstanceMethod(
|
||||
activity,
|
||||
"handleRealtimeEvent",
|
||||
ReflectionHelpers.ClassParameter.from(
|
||||
BossRealtimeEvent.class,
|
||||
new BossRealtimeEvent("master_agent.task.updated", new JSONObject().put("projectId", "project-2"))
|
||||
)
|
||||
);
|
||||
Shadows.shadowOf(activity.getMainLooper()).idle();
|
||||
|
||||
assertEquals(0, activity.reloadCount);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void distinctRealtimeEventsBackToBackStillReloadMatchingProject() throws Exception {
|
||||
Intent intent = new Intent()
|
||||
.putExtra(ProjectDetailActivity.EXTRA_PROJECT_ID, "project-1")
|
||||
.putExtra(ProjectDetailActivity.EXTRA_PROJECT_NAME, "北区试产线");
|
||||
TestRealtimeProjectDetailActivity activity = Robolectric
|
||||
.buildActivity(TestRealtimeProjectDetailActivity.class, intent)
|
||||
.setup()
|
||||
.resume()
|
||||
.get();
|
||||
|
||||
ReflectionHelpers.callInstanceMethod(
|
||||
activity,
|
||||
"handleRealtimeEvent",
|
||||
ReflectionHelpers.ClassParameter.from(
|
||||
BossRealtimeEvent.class,
|
||||
new BossRealtimeEvent(
|
||||
"conversation.updated",
|
||||
new JSONObject().put("projectId", "project-1").put("at", "2026-04-05T10:00:00.000Z")
|
||||
)
|
||||
)
|
||||
);
|
||||
ReflectionHelpers.callInstanceMethod(
|
||||
activity,
|
||||
"handleRealtimeEvent",
|
||||
ReflectionHelpers.ClassParameter.from(
|
||||
BossRealtimeEvent.class,
|
||||
new BossRealtimeEvent(
|
||||
"project.messages.updated",
|
||||
new JSONObject().put("projectId", "project-1").put("at", "2026-04-05T10:00:00.500Z")
|
||||
)
|
||||
)
|
||||
);
|
||||
Shadows.shadowOf(activity.getMainLooper()).idle();
|
||||
|
||||
assertEquals(2, activity.reloadCount);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void duplicateRealtimeEventsWithDifferentAtAreDeduped() throws Exception {
|
||||
Intent intent = new Intent()
|
||||
.putExtra(ProjectDetailActivity.EXTRA_PROJECT_ID, "project-1")
|
||||
.putExtra(ProjectDetailActivity.EXTRA_PROJECT_NAME, "北区试产线");
|
||||
TestRealtimeProjectDetailActivity activity = Robolectric
|
||||
.buildActivity(TestRealtimeProjectDetailActivity.class, intent)
|
||||
.setup()
|
||||
.resume()
|
||||
.get();
|
||||
|
||||
ReflectionHelpers.callInstanceMethod(
|
||||
activity,
|
||||
"handleRealtimeEvent",
|
||||
ReflectionHelpers.ClassParameter.from(
|
||||
BossRealtimeEvent.class,
|
||||
new BossRealtimeEvent(
|
||||
"project.messages.updated",
|
||||
new JSONObject().put("projectId", "project-1").put("at", "2026-04-05T10:00:00.000Z")
|
||||
)
|
||||
)
|
||||
);
|
||||
ReflectionHelpers.callInstanceMethod(
|
||||
activity,
|
||||
"handleRealtimeEvent",
|
||||
ReflectionHelpers.ClassParameter.from(
|
||||
BossRealtimeEvent.class,
|
||||
new BossRealtimeEvent(
|
||||
"project.messages.updated",
|
||||
new JSONObject().put("projectId", "project-1").put("at", "2026-04-05T10:00:00.500Z")
|
||||
)
|
||||
)
|
||||
);
|
||||
Shadows.shadowOf(activity.getMainLooper()).idle();
|
||||
|
||||
assertEquals(1, activity.reloadCount);
|
||||
}
|
||||
|
||||
public static class TestRealtimeProjectDetailActivity extends ProjectDetailActivity {
|
||||
int reloadCount;
|
||||
|
||||
|
||||
Reference in New Issue
Block a user