This is an automated email from the ASF dual-hosted git repository.
Aias00 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hertzbeat.git
The following commit(s) were added to refs/heads/master by this push:
new 50ee6b1178 [fix] require an account to subscribe to the alert and
manager streams (#4272)
50ee6b1178 is described below
commit 50ee6b11780ad936bb01b4b1ee02344f5e34da3b
Author: Duansg <[email protected]>
AuthorDate: Mon Aug 17 14:33:55 2026 +0800
[fix] require an account to subscribe to the alert and manager streams
(#4272)
Co-authored-by: Claude Opus 5 (1M context) <[email protected]>
Co-authored-by: aias00 <[email protected]>
---
.../hertzbeat/alert/config/AlertSseManager.java | 66 ++--
.../alert/config/AlertSseManagerTest.java | 90 +++--
.../common/support/SseEmitterRegistry.java | 198 ++++++++++
.../common/support/SseEmitterRegistryTest.java | 242 +++++++++++++
.../manager/config/ManagerSseManager.java | 80 ++---
.../manager/config/ManagerSseManagerTest.java | 101 ++++++
hertzbeat-startup/src/main/resources/sureness.yml | 6 +-
.../startup/security/SurenessSseRuleTest.java | 103 ++++++
.../hertzbeat-mysql-iotdb/conf/sureness.yml | 6 +-
.../hertzbeat-mysql-tdengine/conf/sureness.yml | 6 +-
.../conf/sureness.yml | 6 +-
.../conf/sureness.yml | 6 +-
.../conf/sureness.yml | 6 +-
script/sureness.yml | 6 +-
.../core/interceptor/default.interceptor.spec.ts | 124 +++++++
.../app/core/interceptor/default.interceptor.ts | 6 +-
.../app/layout/basic/widgets/notify.component.ts | 142 ++++----
.../alert/alert-center/alert-center.component.ts | 38 +-
web-app/src/app/service/auth.service.spec.ts | 12 +
web-app/src/app/service/auth.service.ts | 8 +-
.../src/app/service/authorized-sse.service.spec.ts | 400 +++++++++++++++++++++
web-app/src/app/service/authorized-sse.service.ts | 343 ++++++++++++++++++
22 files changed, 1772 insertions(+), 223 deletions(-)
diff --git
a/hertzbeat-alerter/src/main/java/org/apache/hertzbeat/alert/config/AlertSseManager.java
b/hertzbeat-alerter/src/main/java/org/apache/hertzbeat/alert/config/AlertSseManager.java
index e3d245621e..a48e277e69 100644
---
a/hertzbeat-alerter/src/main/java/org/apache/hertzbeat/alert/config/AlertSseManager.java
+++
b/hertzbeat-alerter/src/main/java/org/apache/hertzbeat/alert/config/AlertSseManager.java
@@ -19,62 +19,50 @@
package org.apache.hertzbeat.alert.config;
-import lombok.extern.slf4j.Slf4j;
+import org.apache.hertzbeat.common.support.SseEmitterRegistry;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;
-import
org.springframework.web.servlet.mvc.method.annotation.ResponseBodyEmitter;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
-import java.io.IOException;
-import java.util.Map;
-import java.util.Optional;
-import java.util.concurrent.ConcurrentHashMap;
-
/**
- * SSE manager for alert
+ * SSE manager for alert.
+ *
+ * <p>Note: the lifecycle of a subscription - its timeout, the ceiling on how
many may be held
+ * and the cleanup of the ones that went away - belongs to {@link
SseEmitterRegistry}; what is
+ * alert specific is only the event these subscribers are waiting for.
*/
-@Slf4j
@Component
public class AlertSseManager {
- private final Map<Long, SseEmitter> emitters = new ConcurrentHashMap<>();
+ private static final String ALERT_EVENT = "ALERT_EVENT";
+
+ private final SseEmitterRegistry registry = new
SseEmitterRegistry("alert");
+
+ /**
+ * Registers a subscription for the given client.
+ *
+ * @param clientId Identifier of the subscriber, unique per subscription
+ * @return The emitter the controller returns to spring
+ */
public SseEmitter createEmitter(Long clientId) {
- SseEmitter emitter = new SseEmitter(Long.MAX_VALUE);
- emitter.onCompletion(() -> removeEmitter(clientId));
- emitter.onTimeout(() -> removeEmitter(clientId));
- emitter.onError((ex) -> removeEmitter(clientId));
- emitters.put(clientId, emitter);
- return emitter;
+ return registry.createEmitter(clientId);
}
+ /**
+ * Delivers one alert to every live subscriber.
+ *
+ * @param data Serialised alert payload
+ */
@Async
public void broadcast(String data) {
- emitters.forEach((clientId, emitter) -> {
- try {
- emitter.send(SseEmitter.event()
- .id(String.valueOf(System.currentTimeMillis()))
- .name("ALERT_EVENT")
- .data(data));
- } catch (IOException | IllegalStateException e) {
- tryCompleteAndClean(clientId, emitter);
- } catch (Exception exception) {
- log.error("Failed to broadcast alert data to client: {}",
exception.getMessage());
- tryCompleteAndClean(clientId, emitter);
- }
- });
+ registry.broadcast(ALERT_EVENT, data);
}
- private void tryCompleteAndClean(Long clientId, SseEmitter emitter) {
- try {
-
Optional.ofNullable(emitter).ifPresent(ResponseBodyEmitter::complete);
- } catch (Throwable e) {
- log.debug("Failed to complete emitter for client {}: {}",
clientId, e.getMessage());
- }
- // execute clear
- removeEmitter(clientId);
+ void setMaxEmitters(int maxEmitters) {
+ registry.setMaxEmitters(maxEmitters);
}
- private void removeEmitter(Long clientId) {
- emitters.remove(clientId);
+ int subscriptionCount() {
+ return registry.subscriptionCount();
}
}
diff --git
a/hertzbeat-alerter/src/test/java/org/apache/hertzbeat/alert/config/AlertSseManagerTest.java
b/hertzbeat-alerter/src/test/java/org/apache/hertzbeat/alert/config/AlertSseManagerTest.java
index 40e88ddef9..ecd308042c 100644
---
a/hertzbeat-alerter/src/test/java/org/apache/hertzbeat/alert/config/AlertSseManagerTest.java
+++
b/hertzbeat-alerter/src/test/java/org/apache/hertzbeat/alert/config/AlertSseManagerTest.java
@@ -17,25 +17,30 @@
package org.apache.hertzbeat.alert.config;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
-import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
-
-import java.lang.reflect.Field;
-import java.util.HashMap;
-import java.util.Map;
-
-import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
-import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.Mockito.doThrow;
+import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import java.lang.reflect.Field;
+import java.util.Map;
+import java.util.stream.Collectors;
+import org.apache.hertzbeat.common.support.SseEmitterRegistry;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+import org.springframework.http.HttpStatus;
+import org.springframework.web.server.ResponseStatusException;
+import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
/**
- * alert sse manager test
+ * Test case for {@link AlertSseManager}.
+ *
+ * <p>Note: how a subscription is bounded and cleaned up is covered by
+ * {@code SseEmitterRegistryTest}; what is left here is what makes this stream
the alert one.
*/
-public class AlertSseManagerTest {
+class AlertSseManagerTest {
private AlertSseManager alertSseManager;
@@ -44,26 +49,51 @@ public class AlertSseManagerTest {
alertSseManager = new AlertSseManager();
}
+ /**
+ * The ui subscribes by event name, so an alert delivered under any other
name reaches
+ * nobody even though the connection is up.
+ */
@Test
- void testCompleteThrowsException() throws Exception {
- SseEmitter emitter = alertSseManager.createEmitter(1L);
- assertNotNull(emitter);
+ void testAlertsAreDeliveredUnderTheAlertEventName() throws Exception {
+ alertSseManager.createEmitter(1L);
+ final SseEmitter subscriber = mock(SseEmitter.class);
+ emitters().put(1L, subscriber);
- Map<Long, SseEmitter> emitters = new HashMap<>();
- SseEmitter spyEmitter = mock(SseEmitter.class);
-
- doThrow(new IllegalStateException("Simulated output stream
error")).when(spyEmitter).send(any(SseEmitter.SseEventBuilder.class));
- doThrow(new RuntimeException("Complete
failed")).when(spyEmitter).complete();
-
- emitters.put(1L, spyEmitter);
+ alertSseManager.broadcast("{\"id\":1}");
- Field emittersField =
AlertSseManager.class.getDeclaredField("emitters");
- emittersField.setAccessible(true);
- emittersField.set(alertSseManager, emitters);
+ final ArgumentCaptor<SseEmitter.SseEventBuilder> event =
+ ArgumentCaptor.forClass(SseEmitter.SseEventBuilder.class);
+ verify(subscriber).send(event.capture());
+ final String rendered = event.getValue().build().stream()
+ .map(part -> String.valueOf(part.getData()))
+ .collect(Collectors.joining());
+ assertTrue(rendered.contains("event:ALERT_EVENT"), "alerts must be
delivered as ALERT_EVENT, was " + rendered);
+ assertTrue(rendered.contains("{\"id\":1}"), "the alert payload must be
delivered as is, was " + rendered);
+ }
- assertThrows(RuntimeException.class, () ->
alertSseManager.broadcast("{\"id\":1,\"content\":\"Test alert\"}"));
- Map<Long, SseEmitter> currentEmitters = (Map<Long, SseEmitter>)
emittersField.get(alertSseManager);
- assertFalse(currentEmitters.containsKey(1L), "Emitter should still
exist because complete() threw exception");
+ /**
+ * The manager has to hand its subscriptions to a registry rather than
hold them itself,
+ * otherwise none of the bounds that registry enforces apply to this
stream.
+ */
+ @Test
+ void testSubscriptionsAreBoundedByTheRegistry() {
+ alertSseManager.setMaxEmitters(1);
+
+ assertNotNull(alertSseManager.createEmitter(1L));
+ final ResponseStatusException thrown =
+ assertThrows(ResponseStatusException.class, () ->
alertSseManager.createEmitter(2L));
+
+ assertEquals(HttpStatus.SERVICE_UNAVAILABLE, thrown.getStatusCode());
+ assertEquals(1, alertSseManager.subscriptionCount());
}
-}
\ No newline at end of file
+ @SuppressWarnings("unchecked")
+ private Map<Long, SseEmitter> emitters() throws Exception {
+ final Field registryField =
AlertSseManager.class.getDeclaredField("registry");
+ registryField.setAccessible(true);
+ final Object registry = registryField.get(alertSseManager);
+ final Field emittersField =
SseEmitterRegistry.class.getDeclaredField("emitters");
+ emittersField.setAccessible(true);
+ return (Map<Long, SseEmitter>) emittersField.get(registry);
+ }
+}
diff --git
a/hertzbeat-common-spring/src/main/java/org/apache/hertzbeat/common/support/SseEmitterRegistry.java
b/hertzbeat-common-spring/src/main/java/org/apache/hertzbeat/common/support/SseEmitterRegistry.java
new file mode 100644
index 0000000000..9fe636e687
--- /dev/null
+++
b/hertzbeat-common-spring/src/main/java/org/apache/hertzbeat/common/support/SseEmitterRegistry.java
@@ -0,0 +1,198 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hertzbeat.common.support;
+
+import java.io.IOException;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicInteger;
+import lombok.Setter;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.http.HttpStatus;
+import org.springframework.web.server.ResponseStatusException;
+import
org.springframework.web.servlet.mvc.method.annotation.ResponseBodyEmitter;
+import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
+
+/**
+ * Holds the live server sent event subscriptions of one stream and owns their
lifecycle.
+ *
+ * <p>What: hands an emitter to each subscriber, keeps the live ones,
broadcasts an event to
+ * all of them and drops the ones that have gone away.
+ *
+ * <p>How: hold one instance per stream and name it after that stream, so
refusals and
+ * broadcast failures stay tellable apart in the log. The instance is not a
bean: a stream
+ * owns its registry the way it owns its subscribers.
+ *
+ * <p>Note: an open subscription occupies a request thread for as long as it
lives, so both
+ * how long one may live and how many may exist at once are bounded here. Each
stream used to
+ * carry its own copy of this lifecycle and the copies drifted - one released
the slot of a
+ * subscription that died on an error and the other did not, so on that stream
a failed
+ * subscriber held its slot until the process restarted.
+ */
+@Slf4j
+public class SseEmitterRegistry {
+
+ /**
+ * How long a subscription may stay open before the client has to
reconnect.
+ *
+ * <p>An unbounded timeout means a subscription never expires on its own,
so a client that
+ * goes away without closing cleanly holds its request thread until the
container notices.
+ * A finite timeout bounds that. It is only safe because the ui reconnects
when the stream
+ * ends: it reads through `fetch` rather than `EventSource`, so it has to
reconnect itself,
+ * and it does.
+ */
+ private static final long EMITTER_TIMEOUT_MILLIS = 30 * 60 * 1000L;
+
+ /** Name of the stream these subscriptions belong to, used to make log
lines tellable apart. */
+ private final String streamName;
+
+ /**
+ * Cap on concurrently held subscriptions. Each one occupies a request
thread, so without
+ * a ceiling enough parallel subscriptions exhaust the container's thread
pool and take
+ * the whole application down with them.
+ */
+ @Setter
+ private int maxEmitters = 1000;
+
+ private final Map<Long, SseEmitter> emitters = new ConcurrentHashMap<>();
+
+ /**
+ * Slots taken by held subscriptions, reserved before the subscription is
registered.
+ *
+ * <p>Reading `emitters.size()` and then putting into it would let every
request of a
+ * parallel burst pass the same check and register anyway, which is
exactly the burst the
+ * ceiling exists to survive. Claiming a slot by compare-and-set makes the
ceiling hold no
+ * matter how many requests arrive at once. The counter can briefly run
ahead of the map,
+ * between the claim and the put, which only ever refuses one subscription
too early.
+ */
+ private final AtomicInteger heldSlots = new AtomicInteger();
+
+ /**
+ * @param streamName Name of the stream, used in log lines only
+ */
+ public SseEmitterRegistry(String streamName) {
+ this.streamName = streamName;
+ }
+
+ /**
+ * Registers a subscription for the given client.
+ *
+ * <p>When: from the request thread serving a subscribe request. The
returned emitter is
+ * what the controller hands back to spring, which keeps the request open
around it.
+ *
+ * @param clientId Identifier of the subscriber, unique per subscription,
not null
+ * @return The emitter the caller returns from its controller method
+ * @throws ResponseStatusException With {@code 503} when the stream
already holds as many
+ * subscriptions as it may
+ * @throws NullPointerException When {@code clientId} is null, which is
checked before a
+ * slot is taken so that a caller that gets this wrong cannot
spend the ceiling
+ */
+ public SseEmitter createEmitter(Long clientId) {
+ Objects.requireNonNull(clientId, "clientId of an sse subscription must
not be null");
+ claimSlot();
+ boolean registered = false;
+ try {
+ final SseEmitter emitter = new SseEmitter(EMITTER_TIMEOUT_MILLIS);
+ emitter.onCompletion(() -> removeEmitter(clientId));
+ emitter.onTimeout(() -> removeEmitter(clientId));
+ // Without this a subscription that dies on an error keeps its
slot: a stream that
+ // broadcasts rarely notices nothing, and the slot is held until
the process restarts
+ emitter.onError(ex -> removeEmitter(clientId));
+ // The id is a fresh snowflake per subscription, so this never
replaces a live emitter
+ emitters.put(clientId, emitter);
+ registered = true;
+ return emitter;
+ } finally {
+ if (!registered) {
+ // The slot is only ever given back by removing the
subscription from the map,
+ // so one that never got in there would hold its slot until
the process restarts
+ heldSlots.decrementAndGet();
+ }
+ }
+ }
+
+ /**
+ * Sends one event to every live subscription and drops the ones the send
fails for.
+ *
+ * <p>Note: a failing send is how a client that went away is noticed, so
the failure is
+ * expected rather than exceptional and only unforeseen ones are logged at
error level.
+ *
+ * @param eventName Name of the sse event, which is what the ui subscribes
by
+ * @param data Payload to deliver, already serialised
+ */
+ public void broadcast(String eventName, String data) {
+ emitters.forEach((clientId, emitter) -> {
+ try {
+ emitter.send(SseEmitter.event()
+ .id(String.valueOf(System.currentTimeMillis()))
+ .name(eventName)
+ .data(data));
+ } catch (IOException | IllegalStateException e) {
+ tryCompleteAndClean(clientId, emitter);
+ } catch (Exception e) {
+ log.error("Failed to broadcast {} data to client: {}",
streamName, e.getMessage());
+ tryCompleteAndClean(clientId, emitter);
+ }
+ });
+ }
+
+ /**
+ * @return How many subscriptions this stream currently holds
+ */
+ public int subscriptionCount() {
+ return heldSlots.get();
+ }
+
+ private void claimSlot() {
+ while (true) {
+ final int held = heldSlots.get();
+ if (held >= maxEmitters) {
+ log.warn("Refused {} subscription, already holding {} of at
most {}", streamName, held, maxEmitters);
+ throw new
ResponseStatusException(HttpStatus.SERVICE_UNAVAILABLE,
+ "Too many " + streamName + " subscriptions");
+ }
+ if (heldSlots.compareAndSet(held, held + 1)) {
+ return;
+ }
+ }
+ }
+
+ private void tryCompleteAndClean(Long clientId, SseEmitter emitter) {
+ try {
+
Optional.ofNullable(emitter).ifPresent(ResponseBodyEmitter::complete);
+ } catch (Throwable e) {
+ log.debug("Failed to complete emitter for client {}: {}",
clientId, e.getMessage());
+ }
+ // Execute clear
+ removeEmitter(clientId);
+ }
+
+ /**
+ * A subscription can be dropped more than once for the same client: a
send failure cleans
+ * it up and completing it fires the completion callback on top of that.
The slot is only
+ * released for the removal that actually took the emitter out of the map,
so the count
+ * cannot drift below what is held and quietly reopen the ceiling.
+ */
+ private void removeEmitter(Long clientId) {
+ if (emitters.remove(clientId) != null) {
+ heldSlots.decrementAndGet();
+ }
+ }
+}
diff --git
a/hertzbeat-common-spring/src/test/java/org/apache/hertzbeat/common/support/SseEmitterRegistryTest.java
b/hertzbeat-common-spring/src/test/java/org/apache/hertzbeat/common/support/SseEmitterRegistryTest.java
new file mode 100644
index 0000000000..5702f9f0d4
--- /dev/null
+++
b/hertzbeat-common-spring/src/test/java/org/apache/hertzbeat/common/support/SseEmitterRegistryTest.java
@@ -0,0 +1,242 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hertzbeat.common.support;
+
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.mockConstruction;
+import static org.mockito.Mockito.verify;
+import java.lang.reflect.Field;
+import java.util.Map;
+import java.util.concurrent.CyclicBarrier;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Consumer;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+import org.mockito.MockedConstruction;
+import org.springframework.http.HttpStatus;
+import org.springframework.web.server.ResponseStatusException;
+import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
+
+/**
+ * Test case for {@link SseEmitterRegistry}.
+ *
+ * <p>Every open subscription holds a request thread for as long as it lives,
so both how long
+ * one may live and how many may exist at once have to be bounded, and every
way a
+ * subscription can end has to give its slot back.
+ */
+class SseEmitterRegistryTest {
+
+ private SseEmitterRegistry registry;
+
+ @BeforeEach
+ void setUp() {
+ registry = new SseEmitterRegistry("test");
+ }
+
+ /**
+ * An unbounded emitter never expires on its own, so a client that goes
away without
+ * closing cleanly keeps holding its request thread until the container
notices.
+ */
+ @Test
+ void testSubscriptionsAreGivenFiniteTimeout() {
+ final SseEmitter emitter = registry.createEmitter(1L);
+
+ assertNotNull(emitter.getTimeout());
+ assertTrue(emitter.getTimeout() > 0 && emitter.getTimeout() <
Long.MAX_VALUE,
+ "timeout must be finite, was " + emitter.getTimeout());
+ }
+
+ /**
+ * Each open subscription occupies a request thread, so enough of them in
parallel
+ * exhaust the container's pool and take the whole application down.
+ */
+ @Test
+ void testSubscriptionsBeyondLimitAreRefused() {
+ registry.setMaxEmitters(2);
+
+ registry.createEmitter(1L);
+ registry.createEmitter(2L);
+ final ResponseStatusException thrown =
+ assertThrows(ResponseStatusException.class, () ->
registry.createEmitter(3L));
+
+ assertEquals(HttpStatus.SERVICE_UNAVAILABLE, thrown.getStatusCode());
+ assertEquals(2, registry.subscriptionCount());
+ }
+
+ /**
+ * The cap must not become a permanent lockout: once a dead subscription
is cleaned up,
+ * its slot has to be available again.
+ */
+ @Test
+ void testDroppedSubscriptionFreesItsSlot() throws Exception {
+ registry.setMaxEmitters(1);
+ registry.createEmitter(1L);
+ assertThrows(ResponseStatusException.class, () ->
registry.createEmitter(2L));
+
+ // A client that went away makes the next send fail, which is how the
registry notices
+ final SseEmitter deadEmitter = mock(SseEmitter.class);
+ doThrow(new IllegalStateException("client
gone")).when(deadEmitter).send(any(SseEmitter.SseEventBuilder.class));
+ replaceEmitter(1L, deadEmitter);
+
+ registry.broadcast("TEST_EVENT", "{\"id\":1}");
+
+ assertEquals(0, registry.subscriptionCount());
+ assertNotNull(registry.createEmitter(2L));
+ }
+
+ /**
+ * Nothing else frees the slot of a subscription that dies on an error: a
stream that only
+ * broadcasts now and then would let a failed client hold its slot
indefinitely and the
+ * ceiling would drift into a permanent lockout.
+ */
+ @Test
+ @SuppressWarnings("unchecked")
+ void testErroredSubscriptionFreesItsSlot() {
+ registry.setMaxEmitters(1);
+
+ try (MockedConstruction<SseEmitter> construction =
mockConstruction(SseEmitter.class)) {
+ registry.createEmitter(1L);
+ final SseEmitter created = construction.constructed().get(0);
+ final ArgumentCaptor<Consumer<Throwable>> onError =
ArgumentCaptor.forClass(Consumer.class);
+ verify(created).onError(onError.capture());
+
+ onError.getValue().accept(new IllegalStateException("client
gone"));
+
+ assertEquals(0, registry.subscriptionCount());
+ }
+ }
+
+ /**
+ * Completing a dropped subscription is a courtesy, so a container that
refuses it must
+ * not keep the subscription registered and its slot taken.
+ */
+ @Test
+ void testSubscriptionIsDroppedEvenWhenCompletingItFails() throws Exception
{
+ registry.createEmitter(1L);
+ final SseEmitter unusableEmitter = mock(SseEmitter.class);
+ doThrow(new IllegalStateException("client gone"))
+
.when(unusableEmitter).send(any(SseEmitter.SseEventBuilder.class));
+ doThrow(new RuntimeException("complete
failed")).when(unusableEmitter).complete();
+ replaceEmitter(1L, unusableEmitter);
+
+ assertDoesNotThrow(() -> registry.broadcast("TEST_EVENT",
"{\"id\":1}"));
+
+ assertFalse(emitters().containsKey(1L), "a subscription that cannot be
completed still has to be dropped");
+ assertEquals(0, registry.subscriptionCount());
+ }
+
+ /**
+ * A slot is taken before the subscription exists, so anything that goes
wrong between the
+ * two has to give it back: nothing ends up in the map to be removed
later, and the slot
+ * would be held until the process restarts.
+ */
+ @Test
+ void testSlotIsGivenBackWhenRegisteringFails() {
+ registry.setMaxEmitters(1);
+
+ try (MockedConstruction<SseEmitter> failing =
mockConstruction(SseEmitter.class, (mock, context) -> {
+ throw new IllegalStateException("cannot open a subscription");
+ })) {
+ // The failure is raised through mockito, so what reaches the
caller is whatever it
+ // wraps the initializer's exception in; only that it fails
matters here
+ assertThrows(RuntimeException.class, () ->
registry.createEmitter(1L));
+ }
+
+ assertEquals(0, registry.subscriptionCount());
+ assertNotNull(registry.createEmitter(2L), "a failed registration must
not spend the ceiling");
+ }
+
+ /**
+ * A caller that has no id for its subscriber cannot be registered at all,
so it must be
+ * turned away before it takes a slot rather than after.
+ */
+ @Test
+ void testSubscriptionWithoutClientIdTakesNoSlot() {
+ assertThrows(NullPointerException.class, () ->
registry.createEmitter(null));
+
+ assertEquals(0, registry.subscriptionCount());
+ }
+
+ /**
+ * The ceiling has to hold for subscriptions that arrive together, which
is the only case
+ * that matters: a burst is what exhausts the thread pool, and a burst is
exactly what a
+ * check of the current count followed by a separate registration lets
straight through.
+ */
+ @Test
+ void testCeilingHoldsWhenSubscriptionsArriveTogether() throws Exception {
+ final int maxEmitters = 10;
+ final int racers = 200;
+ registry.setMaxEmitters(maxEmitters);
+ final CyclicBarrier allReady = new CyclicBarrier(racers);
+ final AtomicInteger accepted = new AtomicInteger();
+ final ExecutorService pool = Executors.newFixedThreadPool(racers);
+
+ try {
+ for (int i = 0; i < racers; i++) {
+ final long clientId = i;
+ pool.submit(() -> {
+ try {
+ // Every thread is released at the same instant, so
they all reach the
+ // ceiling check together
+ allReady.await(10, TimeUnit.SECONDS);
+ registry.createEmitter(clientId);
+ accepted.incrementAndGet();
+ } catch (ResponseStatusException refused) {
+ // Expected for everyone who arrives after the ceiling
is reached
+ } catch (Exception e) {
+ Thread.currentThread().interrupt();
+ }
+ });
+ }
+ pool.shutdown();
+ assertTrue(pool.awaitTermination(30, TimeUnit.SECONDS),
"subscriptions did not settle");
+ } finally {
+ pool.shutdownNow();
+ }
+
+ assertEquals(maxEmitters, accepted.get(), "more subscriptions were
accepted than the ceiling allows");
+ assertEquals(maxEmitters, registry.subscriptionCount());
+ }
+
+ /**
+ * Stands in for a client that went away: the registry only learns about
it when a send
+ * fails, which needs an emitter that fails on demand.
+ */
+ private void replaceEmitter(Long clientId, SseEmitter emitter) throws
Exception {
+ emitters().put(clientId, emitter);
+ }
+
+ @SuppressWarnings("unchecked")
+ private Map<Long, SseEmitter> emitters() throws Exception {
+ final Field emittersField =
SseEmitterRegistry.class.getDeclaredField("emitters");
+ emittersField.setAccessible(true);
+ return (Map<Long, SseEmitter>) emittersField.get(registry);
+ }
+}
diff --git
a/hertzbeat-manager/src/main/java/org/apache/hertzbeat/manager/config/ManagerSseManager.java
b/hertzbeat-manager/src/main/java/org/apache/hertzbeat/manager/config/ManagerSseManager.java
index 1fd681566c..7434f28f72 100644
---
a/hertzbeat-manager/src/main/java/org/apache/hertzbeat/manager/config/ManagerSseManager.java
+++
b/hertzbeat-manager/src/main/java/org/apache/hertzbeat/manager/config/ManagerSseManager.java
@@ -19,80 +19,68 @@
package org.apache.hertzbeat.manager.config;
-import lombok.extern.slf4j.Slf4j;
import org.apache.hertzbeat.common.constants.ManagerEventTypeEnum;
import org.apache.hertzbeat.common.entity.dto.ImportTaskMessage;
import org.apache.hertzbeat.common.entity.dto.ManagerMessage;
+import org.apache.hertzbeat.common.support.SseEmitterRegistry;
import org.apache.hertzbeat.common.util.JsonUtil;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;
-import
org.springframework.web.servlet.mvc.method.annotation.ResponseBodyEmitter;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
-import java.io.IOException;
-import java.util.Map;
-import java.util.Optional;
-import java.util.concurrent.ConcurrentHashMap;
-
/**
- * Manager SSE
+ * Manager SSE.
+ *
+ * <p>Note: the lifecycle of a subscription - its timeout, the ceiling on how
many may be held
+ * and the cleanup of the ones that went away - belongs to {@link
SseEmitterRegistry}; what is
+ * manager specific is only the events these subscribers are waiting for.
*/
-@Slf4j
@Component
public class ManagerSseManager {
- private final Map<Long, SseEmitter> emitters = new ConcurrentHashMap<>();
+ private final SseEmitterRegistry registry = new
SseEmitterRegistry("manager");
+
+ /**
+ * Registers a subscription for the given client.
+ *
+ * @param clientId Identifier of the subscriber, unique per subscription
+ * @return The emitter the controller returns to spring
+ */
public SseEmitter createEmitter(Long clientId) {
- SseEmitter emitter = new SseEmitter(Long.MAX_VALUE);
- emitter.onCompletion(() -> removeEmitter(clientId));
- emitter.onTimeout(() -> removeEmitter(clientId));
- emitters.put(clientId, emitter);
- return emitter;
+ return registry.createEmitter(clientId);
}
+ /**
+ * Delivers one manager event to every live subscriber.
+ *
+ * @param eventName Name of the sse event, which is what the ui subscribes
by
+ * @param data Serialised payload
+ */
@Async
public void broadcast(String eventName, String data) {
- emitters.forEach((clientId, emitter) -> {
- try {
- emitter.send(SseEmitter.event()
- .id(String.valueOf(System.currentTimeMillis()))
- .name(eventName)
- .data(data));
- } catch (IOException | IllegalStateException e) {
- tryCompleteAndClean(clientId, emitter);
- } catch (Exception exception) {
- log.error("Failed to broadcast manager message data to client:
{}", exception.getMessage());
- tryCompleteAndClean(clientId, emitter);
- }
- });
+ registry.broadcast(eventName, data);
}
- private void tryCompleteAndClean(Long clientId, SseEmitter emitter) {
- try {
-
Optional.ofNullable(emitter).ifPresent(ResponseBodyEmitter::complete);
- } catch (Throwable e) {
- log.debug("Failed to complete emitter for client {}: {}",
clientId, e.getMessage());
- }
- // execute clear
- removeEmitter(clientId);
+ public void broadcastImportTaskInProgress(String taskName, Integer
progress) {
+ final ManagerMessage managerMessage =
ImportTaskMessage.createInProgressMessage(taskName, progress);
+ broadcast(ManagerEventTypeEnum.IMPORT_TASK_EVENT.getValue(),
JsonUtil.toJson(managerMessage));
}
- public void broadcastImportTaskInProgress(String taskName, Integer
progress){
- ManagerMessage managerMessage =
ImportTaskMessage.createInProgressMessage(taskName, progress);
+ public void broadcastImportTaskSuccess(String taskName) {
+ final ManagerMessage managerMessage =
ImportTaskMessage.createCompletedMessage(taskName);
broadcast(ManagerEventTypeEnum.IMPORT_TASK_EVENT.getValue(),
JsonUtil.toJson(managerMessage));
}
- public void broadcastImportTaskSuccess(String taskName){
- ManagerMessage managerMessage =
ImportTaskMessage.createCompletedMessage(taskName);
+ public void broadcastImportTaskFail(String taskName, String errMsg) {
+ final ManagerMessage managerMessage =
ImportTaskMessage.createFailedMessage(taskName, errMsg);
broadcast(ManagerEventTypeEnum.IMPORT_TASK_EVENT.getValue(),
JsonUtil.toJson(managerMessage));
}
- public void broadcastImportTaskFail(String taskName, String errMsg){
- ManagerMessage managerMessage =
ImportTaskMessage.createFailedMessage(taskName, errMsg);
- broadcast(ManagerEventTypeEnum.IMPORT_TASK_EVENT.getValue(),
JsonUtil.toJson(managerMessage));
+ void setMaxEmitters(int maxEmitters) {
+ registry.setMaxEmitters(maxEmitters);
}
- private void removeEmitter(Long clientId) {
- emitters.remove(clientId);
+ int subscriptionCount() {
+ return registry.subscriptionCount();
}
-}
\ No newline at end of file
+}
diff --git
a/hertzbeat-manager/src/test/java/org/apache/hertzbeat/manager/config/ManagerSseManagerTest.java
b/hertzbeat-manager/src/test/java/org/apache/hertzbeat/manager/config/ManagerSseManagerTest.java
new file mode 100644
index 0000000000..498e4739f2
--- /dev/null
+++
b/hertzbeat-manager/src/test/java/org/apache/hertzbeat/manager/config/ManagerSseManagerTest.java
@@ -0,0 +1,101 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hertzbeat.manager.config;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import java.lang.reflect.Field;
+import java.util.Map;
+import java.util.stream.Collectors;
+import org.apache.hertzbeat.common.constants.ManagerEventTypeEnum;
+import org.apache.hertzbeat.common.support.SseEmitterRegistry;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+import org.springframework.http.HttpStatus;
+import org.springframework.web.server.ResponseStatusException;
+import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
+
+/**
+ * Test case for {@link ManagerSseManager}.
+ *
+ * <p>Note: how a subscription is bounded and cleaned up is covered by
+ * {@code SseEmitterRegistryTest}; what is left here is what makes this stream
the manager one.
+ */
+class ManagerSseManagerTest {
+
+ private ManagerSseManager managerSseManager;
+
+ @BeforeEach
+ void setUp() {
+ managerSseManager = new ManagerSseManager();
+ }
+
+ /**
+ * The ui subscribes by event name, so import progress delivered under any
other name
+ * reaches nobody even though the connection is up.
+ */
+ @Test
+ void testImportProgressIsDeliveredUnderTheImportTaskEventName() throws
Exception {
+ managerSseManager.createEmitter(1L);
+ final SseEmitter subscriber = mock(SseEmitter.class);
+ emitters().put(1L, subscriber);
+
+ managerSseManager.broadcastImportTaskSuccess("my-task");
+
+ final ArgumentCaptor<SseEmitter.SseEventBuilder> event =
+ ArgumentCaptor.forClass(SseEmitter.SseEventBuilder.class);
+ verify(subscriber).send(event.capture());
+ final String rendered = event.getValue().build().stream()
+ .map(part -> String.valueOf(part.getData()))
+ .collect(Collectors.joining());
+ assertTrue(rendered.contains("event:" +
ManagerEventTypeEnum.IMPORT_TASK_EVENT.getValue()),
+ "import progress must be delivered as IMPORT_TASK_EVENT, was "
+ rendered);
+ assertTrue(rendered.contains("my-task"), "the task name must reach the
subscriber, was " + rendered);
+ }
+
+ /**
+ * The manager has to hand its subscriptions to a registry rather than
hold them itself,
+ * otherwise none of the bounds that registry enforces apply to this
stream.
+ */
+ @Test
+ void testSubscriptionsAreBoundedByTheRegistry() {
+ managerSseManager.setMaxEmitters(1);
+
+ assertNotNull(managerSseManager.createEmitter(1L));
+ final ResponseStatusException thrown =
+ assertThrows(ResponseStatusException.class, () ->
managerSseManager.createEmitter(2L));
+
+ assertEquals(HttpStatus.SERVICE_UNAVAILABLE, thrown.getStatusCode());
+ assertEquals(1, managerSseManager.subscriptionCount());
+ }
+
+ @SuppressWarnings("unchecked")
+ private Map<Long, SseEmitter> emitters() throws Exception {
+ final Field registryField =
ManagerSseManager.class.getDeclaredField("registry");
+ registryField.setAccessible(true);
+ final Object registry = registryField.get(managerSseManager);
+ final Field emittersField =
SseEmitterRegistry.class.getDeclaredField("emitters");
+ emittersField.setAccessible(true);
+ return (Map<Long, SseEmitter>) emittersField.get(registry);
+ }
+}
diff --git a/hertzbeat-startup/src/main/resources/sureness.yml
b/hertzbeat-startup/src/main/resources/sureness.yml
index 5aa2f8a26e..9485a76b54 100644
--- a/hertzbeat-startup/src/main/resources/sureness.yml
+++ b/hertzbeat-startup/src/main/resources/sureness.yml
@@ -129,19 +129,21 @@ resourceRole:
- /swagger-resources/**===get===[admin]
# spring boot actuator exposes jvm, http and datasource internals for
scraping
- /actuator/**===get===[admin]
+ # the alert stream carries full alert payloads and the manager stream
carries import
+ # progress; both are scoped like the log stream above rather than left
anonymous
+ - /api/alert/sse/**===get===[admin,user,guest]
+ - /api/manager/sse/**===get===[admin,user,guest]
# config the resource restful api that need bypass auth protection
# rule: api===method
# eg: /api/v1/source3===get means /api/v1/source3===get can be access by
anyone, no need auth.
excludedResource:
- - /api/alert/sse/**===*
- /api/account/auth/**===*
- /api/i18n/**===get
- /api/apps/hierarchy===get
- /api/observability/capability===get
- /api/push/**===*
- /api/status/page/public/**===*
- - /api/manager/sse/**===*
# web ui resource
- /===get
- /assets/**===get
diff --git
a/hertzbeat-startup/src/test/java/org/apache/hertzbeat/startup/security/SurenessSseRuleTest.java
b/hertzbeat-startup/src/test/java/org/apache/hertzbeat/startup/security/SurenessSseRuleTest.java
new file mode 100644
index 0000000000..f8dba6e5e1
--- /dev/null
+++
b/hertzbeat-startup/src/test/java/org/apache/hertzbeat/startup/security/SurenessSseRuleTest.java
@@ -0,0 +1,103 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hertzbeat.startup.security;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import com.usthe.sureness.matcher.util.TirePathTree;
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.yaml.snakeyaml.Yaml;
+
+/**
+ * Guards the rbac rules covering the server sent event streams.
+ *
+ * <p>`/api/alert/sse/**` and `/api/manager/sse/**` used to sit in
`excludedResource`.
+ * Sureness evaluates the exclusion tree before any credential check, so an
anonymous
+ * `curl -N` stayed subscribed and received every alert the deployment raised
- internal
+ * hostnames, addresses, metric values and alert content - because
+ * `AlertNoticeDispatch` broadcasts each alert to every subscriber with no per
subscriber
+ * filtering. `/api/logs/sse/**` was already scoped this way; these now match
it.
+ */
+class SurenessSseRuleTest {
+
+ private static final String SEPARATOR = "===";
+
+ private static final String EXCLUDE_ROLE = SEPARATOR + "[exclude]";
+
+ private static TirePathTree roleTree;
+
+ private static TirePathTree excludeTree;
+
+ @BeforeAll
+ @SuppressWarnings("unchecked")
+ static void loadSurenessConfig() throws IOException {
+ List<String> resourceRole;
+ List<String> excludedResource;
+ try (InputStream in =
SurenessSseRuleTest.class.getResourceAsStream("/sureness.yml")) {
+ assertNotNull(in, "sureness.yml must be on the classpath");
+ Map<String, Object> document = new Yaml().load(in);
+ resourceRole = (List<String>) document.get("resourceRole");
+ excludedResource = (List<String>) document.get("excludedResource");
+ }
+ assertNotNull(resourceRole, "resourceRole must be present");
+ assertNotNull(excludedResource, "excludedResource must be present");
+ roleTree = new TirePathTree();
+ roleTree.buildTree(new LinkedHashSet<>(resourceRole));
+ excludeTree = new TirePathTree();
+ excludeTree.buildTree(excludedResource.stream()
+ .map(rule -> rule + EXCLUDE_ROLE)
+ .collect(Collectors.toCollection(LinkedHashSet::new)));
+ }
+
+ @Test
+ void subscribingToAlertsRequiresAnAccount() {
+ assertEquals("[admin,user,guest]",
+ roleTree.searchPathFilterRoles("/api/alert/sse/subscribe" +
SEPARATOR + "get"));
+ }
+
+ @Test
+ void subscribingToManagerEventsRequiresAnAccount() {
+ assertEquals("[admin,user,guest]",
+ roleTree.searchPathFilterRoles("/api/manager/sse/subscribe" +
SEPARATOR + "get"));
+ }
+
+ /**
+ * The rule above only takes effect if the path stops matching an
exclusion: sureness
+ * returns from `checkIn` as soon as `isExcludedResource` matches, before
any credential
+ * is looked at.
+ */
+ @Test
+ void theStreamsAreNoLongerAnonymous() {
+
assertNull(excludeTree.searchPathFilterRoles("/api/alert/sse/subscribe" +
SEPARATOR + "get"));
+
assertNull(excludeTree.searchPathFilterRoles("/api/manager/sse/subscribe" +
SEPARATOR + "get"));
+ }
+
+ @Test
+ void theLogStreamScopingIsUnchanged() {
+ assertEquals("[admin,user,guest]",
+ roleTree.searchPathFilterRoles("/api/logs/sse/subscribe" +
SEPARATOR + "get"));
+ }
+}
diff --git a/script/docker-compose/hertzbeat-mysql-iotdb/conf/sureness.yml
b/script/docker-compose/hertzbeat-mysql-iotdb/conf/sureness.yml
index 1815529be7..4fb6e2f1f5 100644
--- a/script/docker-compose/hertzbeat-mysql-iotdb/conf/sureness.yml
+++ b/script/docker-compose/hertzbeat-mysql-iotdb/conf/sureness.yml
@@ -118,19 +118,21 @@ resourceRole:
- /swagger-resources/**===get===[admin]
# spring boot actuator exposes jvm, http and datasource internals for
scraping
- /actuator/**===get===[admin]
+ # the alert stream carries full alert payloads and the manager stream
carries import
+ # progress; both are scoped like the log stream above rather than left
anonymous
+ - /api/alert/sse/**===get===[admin,user,guest]
+ - /api/manager/sse/**===get===[admin,user,guest]
# config the resource restful api that need bypass auth protection
# rule: api===method
# eg: /api/v1/source3===get means /api/v1/source3===get can be access by
anyone, no need auth.
excludedResource:
- - /api/alert/sse/**===*
- /api/account/auth/**===*
- /api/i18n/**===get
- /api/apps/hierarchy===get
- /api/observability/capability===get
- /api/push/**===*
- /api/status/page/public/**===*
- - /api/manager/sse/**===*
# web ui resource
- /===get
- /assets/**===get
diff --git a/script/docker-compose/hertzbeat-mysql-tdengine/conf/sureness.yml
b/script/docker-compose/hertzbeat-mysql-tdengine/conf/sureness.yml
index 1815529be7..4fb6e2f1f5 100644
--- a/script/docker-compose/hertzbeat-mysql-tdengine/conf/sureness.yml
+++ b/script/docker-compose/hertzbeat-mysql-tdengine/conf/sureness.yml
@@ -118,19 +118,21 @@ resourceRole:
- /swagger-resources/**===get===[admin]
# spring boot actuator exposes jvm, http and datasource internals for
scraping
- /actuator/**===get===[admin]
+ # the alert stream carries full alert payloads and the manager stream
carries import
+ # progress; both are scoped like the log stream above rather than left
anonymous
+ - /api/alert/sse/**===get===[admin,user,guest]
+ - /api/manager/sse/**===get===[admin,user,guest]
# config the resource restful api that need bypass auth protection
# rule: api===method
# eg: /api/v1/source3===get means /api/v1/source3===get can be access by
anyone, no need auth.
excludedResource:
- - /api/alert/sse/**===*
- /api/account/auth/**===*
- /api/i18n/**===get
- /api/apps/hierarchy===get
- /api/observability/capability===get
- /api/push/**===*
- /api/status/page/public/**===*
- - /api/manager/sse/**===*
# web ui resource
- /===get
- /assets/**===get
diff --git
a/script/docker-compose/hertzbeat-mysql-victoria-metrics/conf/sureness.yml
b/script/docker-compose/hertzbeat-mysql-victoria-metrics/conf/sureness.yml
index 1815529be7..4fb6e2f1f5 100644
--- a/script/docker-compose/hertzbeat-mysql-victoria-metrics/conf/sureness.yml
+++ b/script/docker-compose/hertzbeat-mysql-victoria-metrics/conf/sureness.yml
@@ -118,19 +118,21 @@ resourceRole:
- /swagger-resources/**===get===[admin]
# spring boot actuator exposes jvm, http and datasource internals for
scraping
- /actuator/**===get===[admin]
+ # the alert stream carries full alert payloads and the manager stream
carries import
+ # progress; both are scoped like the log stream above rather than left
anonymous
+ - /api/alert/sse/**===get===[admin,user,guest]
+ - /api/manager/sse/**===get===[admin,user,guest]
# config the resource restful api that need bypass auth protection
# rule: api===method
# eg: /api/v1/source3===get means /api/v1/source3===get can be access by
anyone, no need auth.
excludedResource:
- - /api/alert/sse/**===*
- /api/account/auth/**===*
- /api/i18n/**===get
- /api/apps/hierarchy===get
- /api/observability/capability===get
- /api/push/**===*
- /api/status/page/public/**===*
- - /api/manager/sse/**===*
# web ui resource
- /===get
- /assets/**===get
diff --git
a/script/docker-compose/hertzbeat-postgresql-greptimedb/conf/sureness.yml
b/script/docker-compose/hertzbeat-postgresql-greptimedb/conf/sureness.yml
index 6281d0927d..5f170b6469 100644
--- a/script/docker-compose/hertzbeat-postgresql-greptimedb/conf/sureness.yml
+++ b/script/docker-compose/hertzbeat-postgresql-greptimedb/conf/sureness.yml
@@ -122,19 +122,21 @@ resourceRole:
- /swagger-resources/**===get===[admin]
# spring boot actuator exposes jvm, http and datasource internals for
scraping
- /actuator/**===get===[admin]
+ # the alert stream carries full alert payloads and the manager stream
carries import
+ # progress; both are scoped like the log stream above rather than left
anonymous
+ - /api/alert/sse/**===get===[admin,user,guest]
+ - /api/manager/sse/**===get===[admin,user,guest]
# config the resource restful api that need bypass auth protection
# rule: api===method
# eg: /api/v1/source3===get means /api/v1/source3===get can be access by
anyone, no need auth.
excludedResource:
- - /api/alert/sse/**===*
- /api/account/auth/**===*
- /api/i18n/**===get
- /api/apps/hierarchy===get
- /api/observability/capability===get
- /api/push/**===*
- /api/status/page/public/**===*
- - /api/manager/sse/**===*
# web ui resource
- /===get
- /assets/**===get
diff --git
a/script/docker-compose/hertzbeat-postgresql-victoria-metrics/conf/sureness.yml
b/script/docker-compose/hertzbeat-postgresql-victoria-metrics/conf/sureness.yml
index 1815529be7..4fb6e2f1f5 100644
---
a/script/docker-compose/hertzbeat-postgresql-victoria-metrics/conf/sureness.yml
+++
b/script/docker-compose/hertzbeat-postgresql-victoria-metrics/conf/sureness.yml
@@ -118,19 +118,21 @@ resourceRole:
- /swagger-resources/**===get===[admin]
# spring boot actuator exposes jvm, http and datasource internals for
scraping
- /actuator/**===get===[admin]
+ # the alert stream carries full alert payloads and the manager stream
carries import
+ # progress; both are scoped like the log stream above rather than left
anonymous
+ - /api/alert/sse/**===get===[admin,user,guest]
+ - /api/manager/sse/**===get===[admin,user,guest]
# config the resource restful api that need bypass auth protection
# rule: api===method
# eg: /api/v1/source3===get means /api/v1/source3===get can be access by
anyone, no need auth.
excludedResource:
- - /api/alert/sse/**===*
- /api/account/auth/**===*
- /api/i18n/**===get
- /api/apps/hierarchy===get
- /api/observability/capability===get
- /api/push/**===*
- /api/status/page/public/**===*
- - /api/manager/sse/**===*
# web ui resource
- /===get
- /assets/**===get
diff --git a/script/sureness.yml b/script/sureness.yml
index 6281d0927d..5f170b6469 100644
--- a/script/sureness.yml
+++ b/script/sureness.yml
@@ -122,19 +122,21 @@ resourceRole:
- /swagger-resources/**===get===[admin]
# spring boot actuator exposes jvm, http and datasource internals for
scraping
- /actuator/**===get===[admin]
+ # the alert stream carries full alert payloads and the manager stream
carries import
+ # progress; both are scoped like the log stream above rather than left
anonymous
+ - /api/alert/sse/**===get===[admin,user,guest]
+ - /api/manager/sse/**===get===[admin,user,guest]
# config the resource restful api that need bypass auth protection
# rule: api===method
# eg: /api/v1/source3===get means /api/v1/source3===get can be access by
anyone, no need auth.
excludedResource:
- - /api/alert/sse/**===*
- /api/account/auth/**===*
- /api/i18n/**===get
- /api/apps/hierarchy===get
- /api/observability/capability===get
- /api/push/**===*
- /api/status/page/public/**===*
- - /api/manager/sse/**===*
# web ui resource
- /===get
- /assets/**===get
diff --git a/web-app/src/app/core/interceptor/default.interceptor.spec.ts
b/web-app/src/app/core/interceptor/default.interceptor.spec.ts
new file mode 100644
index 0000000000..cc422222fe
--- /dev/null
+++ b/web-app/src/app/core/interceptor/default.interceptor.spec.ts
@@ -0,0 +1,124 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import { HttpContext, HttpErrorResponse, HttpHandler, HttpRequest,
HttpResponse } from '@angular/common/http';
+import { Injector } from '@angular/core';
+import { Router } from '@angular/router';
+import { ALAIN_I18N_TOKEN } from '@delon/theme';
+import { of, throwError } from 'rxjs';
+
+import { AuthService } from '../../service/auth.service';
+import { LocalStorageService } from '../../service/local-storage.service';
+import { DefaultInterceptor } from './default.interceptor';
+import { SILENT_HTTP_ERROR } from './http-context';
+
+describe('DefaultInterceptor', () => {
+ it('does not redirect when a silent refresh request is refused', done => {
+ const router = jasmine.createSpyObj<Router>('Router', ['navigateByUrl']);
+ const injector = {
+ get: (token: unknown) => {
+ if (token === ALAIN_I18N_TOKEN) {
+ return { currentLang: 'en-US' };
+ }
+ if (token === Router) {
+ return router;
+ }
+ throw new Error(`Unexpected dependency: ${String(token)}`);
+ }
+ } as Injector;
+ const authService = jasmine.createSpyObj<AuthService>('AuthService',
['refreshToken']);
+ const localStorageService =
jasmine.createSpyObj<LocalStorageService>('LocalStorageService', [
+ 'getAuthorizationToken',
+ 'getRefreshToken'
+ ]);
+ localStorageService.getAuthorizationToken.and.returnValue(null as any);
+ const interceptor = new DefaultInterceptor(injector, authService,
localStorageService);
+ const request = new HttpRequest(
+ 'POST',
+ '/account/auth/refresh',
+ {},
+ {
+ context: new HttpContext().set(SILENT_HTTP_ERROR, true)
+ }
+ );
+ const next = {
+ handle: () => throwError(() => new HttpErrorResponse({ status: 401, url:
request.url }))
+ } as HttpHandler;
+
+ interceptor.intercept(request, next).subscribe({
+ error: error => {
+ expect(error.status).toBe(401);
+ setTimeout(() => {
+ expect(router.navigateByUrl).not.toHaveBeenCalled();
+ done();
+ });
+ }
+ });
+ });
+
+ it('still refreshes and replays an ordinary silent request after a 401', ()
=> {
+ const injector = {
+ get: (token: unknown) => {
+ if (token === ALAIN_I18N_TOKEN) {
+ return { currentLang: 'en-US' };
+ }
+ throw new Error(`Unexpected dependency: ${String(token)}`);
+ }
+ } as Injector;
+ const authService = jasmine.createSpyObj<AuthService>('AuthService',
['refreshToken']);
+ authService.refreshToken.and.returnValue(
+ of({ code: 0, msg: '', data: { token: 'fresh-token', refreshToken:
'next-refresh-token' } } as any)
+ );
+ const localStorageService =
jasmine.createSpyObj<LocalStorageService>('LocalStorageService', [
+ 'getAuthorizationToken',
+ 'getRefreshToken',
+ 'storageAuthorizationToken',
+ 'storageRefreshToken'
+ ]);
+ localStorageService.getAuthorizationToken.and.returnValue('expired-token');
+ localStorageService.getRefreshToken.and.returnValue('a-refresh-token');
+ const interceptor = new DefaultInterceptor(injector, authService,
localStorageService);
+ const request = new HttpRequest(
+ 'POST',
+ '/api/observability/metrics',
+ {},
+ {
+ context: new HttpContext().set(SILENT_HTTP_ERROR, true)
+ }
+ );
+ const next = jasmine.createSpyObj<HttpHandler>('HttpHandler', ['handle']);
+ next.handle.and.returnValues(
+ throwError(() => new HttpErrorResponse({ status: 401, url: request.url
})),
+ of(new HttpResponse({ status: 200 }))
+ );
+
+ let response: HttpResponse<unknown> | undefined;
+ interceptor.intercept(request, next).subscribe(event => {
+ if (event instanceof HttpResponse) {
+ response = event;
+ }
+ });
+
+ expect(response?.status).toBe(200);
+ expect(authService.refreshToken).toHaveBeenCalledWith('a-refresh-token');
+
expect(localStorageService.storageAuthorizationToken).toHaveBeenCalledWith('fresh-token');
+
expect(localStorageService.storageRefreshToken).toHaveBeenCalledWith('next-refresh-token');
+ expect(next.handle).toHaveBeenCalledTimes(2);
+ });
+});
diff --git a/web-app/src/app/core/interceptor/default.interceptor.ts
b/web-app/src/app/core/interceptor/default.interceptor.ts
index aba4d01ad3..85074f38de 100644
--- a/web-app/src/app/core/interceptor/default.interceptor.ts
+++ b/web-app/src/app/core/interceptor/default.interceptor.ts
@@ -86,7 +86,9 @@ export class DefaultInterceptor implements HttpInterceptor {
private tryRefreshToken(ev: HttpResponseBase, req: HttpRequest<any>, next:
HttpHandler): Observable<any> {
// 1, redirect to login page if this request is used for refreshing token
if ([`/account/auth/refresh`].some(url => req.url.includes(url))) {
- this.toLogin();
+ if (!req.context.get(SILENT_HTTP_ERROR)) {
+ this.toLogin();
+ }
return throwError(ev);
}
// 2, if `refreshToking` is true, means that the refreshing token request
is in progress
@@ -202,7 +204,7 @@ export class DefaultInterceptor implements HttpInterceptor {
if (!silentError) {
this.checkStatus(err);
}
- return throwError(err.error);
+ return throwError(silentError ? err : err.error);
})
);
}
diff --git a/web-app/src/app/layout/basic/widgets/notify.component.ts
b/web-app/src/app/layout/basic/widgets/notify.component.ts
index 022ff9a0e7..abb4b329b0 100644
--- a/web-app/src/app/layout/basic/widgets/notify.component.ts
+++ b/web-app/src/app/layout/basic/widgets/notify.component.ts
@@ -1,15 +1,17 @@
-import { ChangeDetectionStrategy, ChangeDetectorRef, Component, Inject,
OnInit, OnDestroy } from '@angular/core';
+import { ChangeDetectionStrategy, ChangeDetectorRef, Component, Inject,
NgZone, OnInit, OnDestroy } from '@angular/core';
import { Router } from '@angular/router';
import { I18NService } from '@core';
import { ALAIN_I18N_TOKEN } from '@delon/theme';
import { NzModalService } from 'ng-zorro-antd/modal';
import { NzNotificationService } from 'ng-zorro-antd/notification';
+import { Subscription } from 'rxjs';
import { finalize } from 'rxjs/operators';
import { Mute } from '../../../pojo/Mute';
import { SingleAlert } from '../../../pojo/SingleAlert';
import { AlertSoundService } from '../../../service/alert-sound.service';
import { AlertService } from '../../../service/alert.service';
+import { AuthorizedSseService } from '../../../service/authorized-sse.service';
import { GeneralConfigService } from '../../../service/general-config.service';
@Component({
@@ -125,7 +127,8 @@ export class HeaderNotifyComponent implements OnInit,
OnDestroy {
private previousCount = 0;
// default to mute status
mute: Mute = { mute: true };
- private eventSource!: EventSource;
+ private alertStream$!: Subscription;
+ private managerStream$!: Subscription;
constructor(
private router: Router,
@Inject(ALAIN_I18N_TOKEN) private i18nSvc: I18NService,
@@ -134,7 +137,9 @@ export class HeaderNotifyComponent implements OnInit,
OnDestroy {
private alertSvc: AlertService,
private modal: NzModalService,
private cdr: ChangeDetectorRef,
- private alertSound: AlertSoundService
+ private alertSound: AlertSoundService,
+ private authorizedSseSvc: AuthorizedSseService,
+ private ngZone: NgZone
) {}
ngOnInit(): void {
@@ -166,9 +171,10 @@ export class HeaderNotifyComponent implements OnInit,
OnDestroy {
if (this.refreshInterval) {
clearInterval(this.refreshInterval);
}
- if (this.eventSource) {
- this.eventSource.close();
- }
+ // both streams are unsubscribed: they used to share one field, so the
alert connection
+ // was never closed once the manager one overwrote it
+ this.alertStream$?.unsubscribe();
+ this.managerStream$?.unsubscribe();
}
onPopoverVisibleChange(visible: boolean): void {
@@ -285,74 +291,70 @@ export class HeaderNotifyComponent implements OnInit,
OnDestroy {
}
private initAlertSSEConnection(): void {
- const sseUrl = '/api/alert/sse/subscribe';
-
- this.eventSource = new EventSource(sseUrl);
-
- this.eventSource.addEventListener('ALERT_EVENT', (evt: MessageEvent) => {
- let list: any[] = [];
- let alert: SingleAlert = JSON.parse(evt.data);
- let item = {
- id: alert.id,
- avatar: '/assets/img/notification.svg',
- // title:
`${alert.tags?.monitorName}--${this.i18nSvc.fanyi(`alert.severity.${alert.severity}`)}`,
- title: alert.content,
- datetime: new Date(alert.activeAt).toLocaleString(),
- color: 'blue',
- status: alert.status,
- type: this.i18nSvc.fanyi('dashboard.alerts.title-no')
- };
- console.log('alert:', alert);
- list.push(item);
- this.data = this.updateNoticeData(list);
- if (!this.mute.mute && !this.notifiedAlert.includes(alert.id)) {
- this.notifiedAlert.push(alert.id);
- this.alertSound.playAlertSound(this.i18nSvc.currentLang);
- const notification = new
Notification(this.i18nSvc.fanyi('alert.notify.title'), {
- body: this.i18nSvc.fanyi('alert.notify.body'),
- icon: 'assets/logo.svg'
- });
- notification.onclick = () => {
- window.focus();
- this.router.navigateByUrl(`/alert/center`);
- notification.close();
- };
- }
- this.cdr.detectChanges();
+ // read through AuthorizedSseService rather than EventSource: the stream
now requires a
+ // credential, and EventSource cannot carry an Authorization header
+ this.alertStream$ =
this.authorizedSseSvc.stream('/api/alert/sse/subscribe',
'ALERT_EVENT').subscribe({
+ next: data =>
+ this.ngZone.run(() => {
+ let list: any[] = [];
+ let alert: SingleAlert = JSON.parse(data);
+ let item = {
+ id: alert.id,
+ avatar: '/assets/img/notification.svg',
+ // title:
`${alert.tags?.monitorName}--${this.i18nSvc.fanyi(`alert.severity.${alert.severity}`)}`,
+ title: alert.content,
+ datetime: new Date(alert.activeAt).toLocaleString(),
+ color: 'blue',
+ status: alert.status,
+ type: this.i18nSvc.fanyi('dashboard.alerts.title-no')
+ };
+ console.log('alert:', alert);
+ list.push(item);
+ this.data = this.updateNoticeData(list);
+ if (!this.mute.mute && !this.notifiedAlert.includes(alert.id)) {
+ this.notifiedAlert.push(alert.id);
+ this.alertSound.playAlertSound(this.i18nSvc.currentLang);
+ const notification = new
Notification(this.i18nSvc.fanyi('alert.notify.title'), {
+ body: this.i18nSvc.fanyi('alert.notify.body'),
+ icon: 'assets/logo.svg'
+ });
+ notification.onclick = () => {
+ window.focus();
+ this.router.navigateByUrl(`/alert/center`);
+ notification.close();
+ };
+ }
+ this.cdr.detectChanges();
+ }),
+ error: error => console.error('SSE connection error:', error)
});
- this.eventSource.onerror = error => {
- console.error('SSE connection error:', error);
- this.eventSource.close();
- };
}
private initManagerSSEConnection(): void {
- const sseUrl = '/api/manager/sse/subscribe';
- this.eventSource = new EventSource(sseUrl);
- this.eventSource.addEventListener('IMPORT_TASK_EVENT', (evt: MessageEvent)
=> {
- let msg = JSON.parse(evt.data);
- if (msg.notifyLevel === 'SUCCESS') {
- this.notifySvc.success(
- this.i18nSvc.fanyi('common.notice'),
- this.i18nSvc.fanyi('common.notify.import-success-detail', {
taskName: msg.taskName })
- );
- } else if (msg.notifyLevel === 'ERROR') {
- this.notifySvc.error(
- this.i18nSvc.fanyi('common.notice'),
- this.i18nSvc.fanyi('common.notify.import-fail-detail', { taskName:
msg.taskName, errMsg: msg.errMsg })
- );
- } else if (msg.notifyLevel === 'INFO') {
- this.notifySvc.info(
- this.i18nSvc.fanyi('common.notice'),
- this.i18nSvc.fanyi('common.notify.import-progress', { taskName:
msg.taskName, progress: msg.progress })
- );
- } else {
- console.error('Parse message error, msg:', evt.data);
- }
+ this.managerStream$ =
this.authorizedSseSvc.stream('/api/manager/sse/subscribe',
'IMPORT_TASK_EVENT').subscribe({
+ next: data =>
+ this.ngZone.run(() => {
+ let msg = JSON.parse(data);
+ if (msg.notifyLevel === 'SUCCESS') {
+ this.notifySvc.success(
+ this.i18nSvc.fanyi('common.notice'),
+ this.i18nSvc.fanyi('common.notify.import-success-detail', {
taskName: msg.taskName })
+ );
+ } else if (msg.notifyLevel === 'ERROR') {
+ this.notifySvc.error(
+ this.i18nSvc.fanyi('common.notice'),
+ this.i18nSvc.fanyi('common.notify.import-fail-detail', {
taskName: msg.taskName, errMsg: msg.errMsg })
+ );
+ } else if (msg.notifyLevel === 'INFO') {
+ this.notifySvc.info(
+ this.i18nSvc.fanyi('common.notice'),
+ this.i18nSvc.fanyi('common.notify.import-progress', { taskName:
msg.taskName, progress: msg.progress })
+ );
+ } else {
+ console.error('Parse message error, msg:', data);
+ }
+ }),
+ error: error => console.error('Manager SSE connection error:', error)
});
- this.eventSource.onerror = error => {
- console.error('Manager SSE connection error:', error);
- this.eventSource.close();
- };
}
}
diff --git
a/web-app/src/app/routes/alert/alert-center/alert-center.component.ts
b/web-app/src/app/routes/alert/alert-center/alert-center.component.ts
index 2d734e89fb..6c94f3e184 100644
--- a/web-app/src/app/routes/alert/alert-center/alert-center.component.ts
+++ b/web-app/src/app/routes/alert/alert-center/alert-center.component.ts
@@ -17,14 +17,16 @@
* under the License.
*/
-import { Component, Inject, OnDestroy, OnInit } from '@angular/core';
+import { Component, Inject, NgZone, OnDestroy, OnInit } from '@angular/core';
import { I18NService } from '@core';
import { ALAIN_I18N_TOKEN } from '@delon/theme';
import { NzModalService } from 'ng-zorro-antd/modal';
import { NzNotificationService } from 'ng-zorro-antd/notification';
+import { Subscription } from 'rxjs';
import { GroupAlert } from '../../../pojo/GroupAlert';
import { AlertService } from '../../../service/alert.service';
+import { AuthorizedSseService } from '../../../service/authorized-sse.service';
interface ExtendedGroupAlert extends GroupAlert {
isNew?: boolean;
@@ -39,6 +41,8 @@ export class AlertCenterComponent implements OnInit,
OnDestroy {
private notifySvc: NzNotificationService,
private modal: NzModalService,
private alertSvc: AlertService,
+ private authorizedSseSvc: AuthorizedSseService,
+ private ngZone: NgZone,
@Inject(ALAIN_I18N_TOKEN) private i18nSvc: I18NService
) {}
@@ -50,7 +54,7 @@ export class AlertCenterComponent implements OnInit,
OnDestroy {
checkedAlertIds = new Set<number>();
filterStatus!: string;
filterContent: string | undefined;
- private eventSource!: EventSource;
+ private alertStream$!: Subscription;
ngOnInit(): void {
this.loadAlertsTable();
@@ -58,28 +62,24 @@ export class AlertCenterComponent implements OnInit,
OnDestroy {
}
ngOnDestroy(): void {
- if (this.eventSource) {
- this.eventSource.close();
- }
+ this.alertStream$?.unsubscribe();
}
// Initialize SSE subscription for real-time alerts
private initSSESubscription(): void {
- this.eventSource = new EventSource('/api/alert/sse/subscribe');
- this.eventSource.addEventListener('ALERT_EVENT', (evt: MessageEvent) => {
- try {
- const newAlert: GroupAlert = JSON.parse(evt.data);
- this.updateAlertList(newAlert);
- } catch (error) {
- console.error('Error parsing SSE data:', error);
- }
+ // read through AuthorizedSseService rather than EventSource: the stream
now requires a
+ // credential, and EventSource cannot carry an Authorization header
+ this.alertStream$ =
this.authorizedSseSvc.stream('/api/alert/sse/subscribe',
'ALERT_EVENT').subscribe({
+ next: data => {
+ try {
+ const newAlert: GroupAlert = JSON.parse(data);
+ this.ngZone.run(() => this.updateAlertList(newAlert));
+ } catch (error) {
+ console.error('Error parsing SSE data:', error);
+ }
+ },
+ error: error => console.error('SSE connection error:', error)
});
-
- // Handle SSE errors
- this.eventSource.onerror = error => {
- console.error('SSE connection error:', error);
- this.eventSource.close();
- };
}
private updateAlertList(newAlert: GroupAlert): void {
diff --git a/web-app/src/app/service/auth.service.spec.ts
b/web-app/src/app/service/auth.service.spec.ts
index 650102d387..85008b856f 100644
--- a/web-app/src/app/service/auth.service.spec.ts
+++ b/web-app/src/app/service/auth.service.spec.ts
@@ -17,9 +17,11 @@
* under the License.
*/
+import { HttpTestingController } from '@angular/common/http/testing';
import { TestBed } from '@angular/core/testing';
import { configureHttpServiceTest } from '@testing';
+import { SILENT_HTTP_ERROR } from '../core/interceptor/http-context';
import { AuthService } from './auth.service';
describe('AuthService', () => {
@@ -33,4 +35,14 @@ describe('AuthService', () => {
it('should be created', () => {
expect(service).toBeTruthy();
});
+
+ it('can make a refresh request without global HTTP error handling', () => {
+ const http = TestBed.inject(HttpTestingController);
+
+ service.refreshToken('a-refresh-token', true).subscribe();
+
+ const request = http.expectOne('/account/auth/refresh');
+ expect(request.request.context.get(SILENT_HTTP_ERROR)).toBeTrue();
+ request.flush({ code: 0, msg: '', data: {} });
+ });
});
diff --git a/web-app/src/app/service/auth.service.ts
b/web-app/src/app/service/auth.service.ts
index ccc8ab20ff..c2dbcfcc34 100644
--- a/web-app/src/app/service/auth.service.ts
+++ b/web-app/src/app/service/auth.service.ts
@@ -17,10 +17,11 @@
* under the License.
*/
-import { HttpClient } from '@angular/common/http';
+import { HttpClient, HttpContext } from '@angular/common/http';
import { Injectable } from '@angular/core';
import { Observable } from 'rxjs';
+import { SILENT_HTTP_ERROR } from '../core/interceptor/http-context';
import { Message } from '../pojo/Message';
const account_auth_refresh_uri = '/account/auth/refresh';
@@ -32,11 +33,12 @@ const account_token = '/account/token';
export class AuthService {
constructor(private http: HttpClient) {}
- public refreshToken(refreshToken: string): Observable<Message<any>> {
+ public refreshToken(refreshToken: string, silentHttpError = false):
Observable<Message<any>> {
let body = {
token: refreshToken
};
- return this.http.post<Message<any>>(`${account_auth_refresh_uri}`, body);
+ const context = new HttpContext().set(SILENT_HTTP_ERROR, silentHttpError);
+ return this.http.post<Message<any>>(`${account_auth_refresh_uri}`, body, {
context });
}
public generateToken(name?: string, expireSeconds?: number):
Observable<Message<any>> {
diff --git a/web-app/src/app/service/authorized-sse.service.spec.ts
b/web-app/src/app/service/authorized-sse.service.spec.ts
new file mode 100644
index 0000000000..88f82ad2dc
--- /dev/null
+++ b/web-app/src/app/service/authorized-sse.service.spec.ts
@@ -0,0 +1,400 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import { HttpErrorResponse } from '@angular/common/http';
+import { TestBed } from '@angular/core/testing';
+import { Subject, of, throwError } from 'rxjs';
+
+import { AuthService } from './auth.service';
+import { AuthorizedSseService } from './authorized-sse.service';
+import { LocalStorageService } from './local-storage.service';
+
+describe('AuthorizedSseService', () => {
+ let service: AuthorizedSseService;
+ let localStorageService: jasmine.SpyObj<LocalStorageService>;
+ let authService: jasmine.SpyObj<AuthService>;
+
+ /** Serves the given chunks as a readable body, the way a live sse response
arrives. */
+ function respondWith(chunks: string[], ok = true, status = 200): void {
+ const encoder = new TextEncoder();
+ let index = 0;
+ const reader = {
+ read: () => (index < chunks.length ? Promise.resolve({ value:
encoder.encode(chunks[index++]), done: false }) : new Promise(() => {})) // an
open stream never completes on its own
+ };
+ spyOn(window, 'fetch').and.returnValue(Promise.resolve({ ok, status, body:
{ getReader: () => reader } } as any));
+ }
+
+ /**
+ * Serves one response per connection attempt, so a test can let the first
one end and
+ * assert on what the service does next. A `null` entry stands for a request
that fails
+ * outright rather than returning a response.
+ */
+ function respondPerAttempt(attempts: Array<string[] | null>): void {
+ const encoder = new TextEncoder();
+ let attempt = 0;
+ spyOn(window, 'fetch').and.callFake(() => {
+ const chunks = attempts[Math.min(attempt++, attempts.length - 1)];
+ if (chunks === null) {
+ return Promise.reject(new Error('network down'));
+ }
+ let index = 0;
+ const reader = {
+ // once the chunks run out the server closes the stream, as an emitter
timeout does
+ read: () =>
+ index < chunks.length
+ ? Promise.resolve({ value: encoder.encode(chunks[index++]), done:
false })
+ : Promise.resolve({ value: undefined, done: true })
+ };
+ return Promise.resolve({ ok: true, status: 200, body: { getReader: () =>
reader } } as any);
+ });
+ }
+
+ /**
+ * Serves one outcome per connection attempt: a status code for a response
the server
+ * refuses, or a list of chunks for a stream that stays open.
+ */
+ function respondPerOutcome(outcomes: Array<number | string[]>): void {
+ const encoder = new TextEncoder();
+ let attempt = 0;
+ spyOn(window, 'fetch').and.callFake(() => {
+ const outcome = outcomes[Math.min(attempt++, outcomes.length - 1)];
+ if (typeof outcome === 'number') {
+ return Promise.resolve({ ok: false, status: outcome, body: null } as
any);
+ }
+ let index = 0;
+ const reader = {
+ read: () =>
+ index < outcome.length ? Promise.resolve({ value:
encoder.encode(outcome[index++]), done: false }) : new Promise(() => {})
+ };
+ return Promise.resolve({ ok: true, status: 200, body: { getReader: () =>
reader } } as any);
+ });
+ }
+
+ beforeEach(() => {
+ localStorageService = jasmine.createSpyObj('LocalStorageService', [
+ 'getAuthorizationToken',
+ 'getRefreshToken',
+ 'storageAuthorizationToken',
+ 'storageRefreshToken'
+ ]);
+ authService = jasmine.createSpyObj('AuthService', ['refreshToken']);
+ TestBed.configureTestingModule({
+ providers: [
+ { provide: LocalStorageService, useValue: localStorageService },
+ { provide: AuthService, useValue: authService }
+ ]
+ });
+ service = TestBed.inject(AuthorizedSseService);
+ });
+
+ it('sends the stored token so the stream can require a credential', done => {
+ localStorageService.getAuthorizationToken.and.returnValue('a-token');
+ respondWith(['event:ALERT_EVENT\ndata:{"id":1}\n\n']);
+
+ const subscription = service.stream('/api/alert/sse/subscribe',
'ALERT_EVENT').subscribe(() => {
+ const [, init] = (window.fetch as jasmine.Spy).calls.mostRecent().args;
+ expect((init.headers as Record<string,
string>)['Authorization']).toBe('Bearer a-token');
+ subscription.unsubscribe();
+ done();
+ });
+ });
+
+ it('emits the data payload of a matching event', done => {
+ localStorageService.getAuthorizationToken.and.returnValue('a-token');
+ respondWith(['event:ALERT_EVENT\ndata:{"id":1}\n\n']);
+
+ const subscription = service.stream('/api/alert/sse/subscribe',
'ALERT_EVENT').subscribe(data => {
+ expect(data).toBe('{"id":1}');
+ subscription.unsubscribe();
+ done();
+ });
+ });
+
+ it('ignores events of another name on the same stream', done => {
+ localStorageService.getAuthorizationToken.and.returnValue('a-token');
+ respondWith(['event:OTHER_EVENT\ndata:{"id":1}\n\n',
'event:ALERT_EVENT\ndata:{"id":2}\n\n']);
+
+ const subscription = service.stream('/api/alert/sse/subscribe',
'ALERT_EVENT').subscribe(data => {
+ expect(data).toBe('{"id":2}');
+ subscription.unsubscribe();
+ done();
+ });
+ });
+
+ it('reassembles an event split across chunks', done => {
+ localStorageService.getAuthorizationToken.and.returnValue('a-token');
+ respondWith(['event:ALERT_EVENT\ndata:{"id"', ':3}\n\n']);
+
+ const subscription = service.stream('/api/alert/sse/subscribe',
'ALERT_EVENT').subscribe(data => {
+ expect(data).toBe('{"id":3}');
+ subscription.unsubscribe();
+ done();
+ });
+ });
+
+ it('surfaces a rejected subscription as an error', done => {
+ localStorageService.getAuthorizationToken.and.returnValue(null as any);
+ localStorageService.getRefreshToken.and.returnValue(null);
+ respondWith([], false, 401);
+
+ service.stream('/api/alert/sse/subscribe', 'ALERT_EVENT').subscribe({
+ error: error => {
+ expect(String(error)).toContain('401');
+ done();
+ }
+ });
+ });
+
+ /**
+ * The server closes a subscription once its emitter times out, so a stream
that gave up
+ * at the first close would stop delivering alerts half an hour in.
`EventSource` used to
+ * reconnect on its own; reading through `fetch` has to do it here instead.
+ */
+ it('reconnects after the server closes the stream', done => {
+ localStorageService.getAuthorizationToken.and.returnValue('a-token');
+ respondPerAttempt([['event:ALERT_EVENT\ndata:{"id":1}\n\n'],
['event:ALERT_EVENT\ndata:{"id":2}\n\n']]);
+ service.initialRetryDelayMillis = 0;
+
+ const received: string[] = [];
+ const subscription = service.stream('/api/alert/sse/subscribe',
'ALERT_EVENT').subscribe(data => {
+ received.push(data);
+ if (received.length === 2) {
+ expect(received).toEqual(['{"id":1}', '{"id":2}']);
+ expect((window.fetch as jasmine.Spy).calls.count()).toBe(2);
+ subscription.unsubscribe();
+ done();
+ }
+ });
+ });
+
+ it('reconnects after a failed connection attempt', done => {
+ localStorageService.getAuthorizationToken.and.returnValue('a-token');
+ respondPerAttempt([null, ['event:ALERT_EVENT\ndata:{"id":9}\n\n']]);
+ service.initialRetryDelayMillis = 0;
+
+ const subscription = service.stream('/api/alert/sse/subscribe',
'ALERT_EVENT').subscribe(data => {
+ expect(data).toBe('{"id":9}');
+ expect((window.fetch as jasmine.Spy).calls.count()).toBe(2);
+ subscription.unsubscribe();
+ done();
+ });
+ });
+
+ it('re-reads the token on every attempt so a refreshed one is used', done =>
{
+ localStorageService.getAuthorizationToken.and.returnValues('stale-token',
'fresh-token');
+ respondPerAttempt([['event:ALERT_EVENT\ndata:{"id":1}\n\n'],
['event:ALERT_EVENT\ndata:{"id":2}\n\n']]);
+ service.initialRetryDelayMillis = 0;
+
+ let count = 0;
+ const subscription = service.stream('/api/alert/sse/subscribe',
'ALERT_EVENT').subscribe(() => {
+ if (++count === 2) {
+ const [, init] = (window.fetch as jasmine.Spy).calls.mostRecent().args;
+ expect((init.headers as Record<string,
string>)['Authorization']).toBe('Bearer fresh-token');
+ subscription.unsubscribe();
+ done();
+ }
+ });
+ });
+
+ it('stops reconnecting once the subscription is closed', done => {
+ localStorageService.getAuthorizationToken.and.returnValue('a-token');
+ respondPerAttempt([['event:ALERT_EVENT\ndata:{"id":1}\n\n']]);
+ service.initialRetryDelayMillis = 0;
+
+ const subscription = service.stream('/api/alert/sse/subscribe',
'ALERT_EVENT').subscribe(() => {
+ subscription.unsubscribe();
+ const attemptsAtUnsubscribe = (window.fetch as
jasmine.Spy).calls.count();
+ setTimeout(() => {
+ expect((window.fetch as
jasmine.Spy).calls.count()).toBe(attemptsAtUnsubscribe);
+ done();
+ }, 20);
+ });
+ });
+
+ /**
+ * A credential the server refuses is the one failure reconnecting cannot
fix, so it has to
+ * end the observable instead of turning into a retry loop against the
endpoint.
+ */
+ it('does not retry a rejected credential', done => {
+ localStorageService.getAuthorizationToken.and.returnValue('a-token');
+ localStorageService.getRefreshToken.and.returnValue('a-refresh-token');
+ respondWith([], false, 403);
+ service.initialRetryDelayMillis = 0;
+
+ service.stream('/api/alert/sse/subscribe', 'ALERT_EVENT').subscribe({
+ error: () => {
+ setTimeout(() => {
+ expect((window.fetch as jasmine.Spy).calls.count()).toBe(1);
+ // a refused role is not an expired token, so there is nothing to
refresh
+ expect(authService.refreshToken).not.toHaveBeenCalled();
+ done();
+ }, 20);
+ }
+ });
+ });
+
+ /**
+ * A subscription outlives its credential: the access token is good for an
hour while the
+ * server closes a subscription every half hour, so a reconnect eventually
carries an
+ * expired token. This request never reaches the interceptor that refreshes
on 401, so
+ * without refreshing here the stream would stop delivering for the rest of
the session.
+ */
+ it('refreshes an expired token and reconnects', done => {
+
localStorageService.getAuthorizationToken.and.returnValues('expired-token',
'fresh-token');
+ localStorageService.getRefreshToken.and.returnValue('a-refresh-token');
+ authService.refreshToken.and.returnValue(
+ of({ code: 0, msg: '', data: { token: 'fresh-token', refreshToken:
'next-refresh-token' } } as any)
+ );
+ respondPerOutcome([401, ['event:ALERT_EVENT\ndata:{"id":7}\n\n']]);
+ service.initialRetryDelayMillis = 0;
+
+ const subscription = service.stream('/api/alert/sse/subscribe',
'ALERT_EVENT').subscribe({
+ next: data => {
+ expect(data).toBe('{"id":7}');
+
expect(authService.refreshToken).toHaveBeenCalledWith('a-refresh-token', true);
+
expect(localStorageService.storageAuthorizationToken).toHaveBeenCalledWith('fresh-token');
+
expect(localStorageService.storageRefreshToken).toHaveBeenCalledWith('next-refresh-token');
+ const [, init] = (window.fetch as jasmine.Spy).calls.mostRecent().args;
+ expect((init.headers as Record<string,
string>)['Authorization']).toBe('Bearer fresh-token');
+ subscription.unsubscribe();
+ done();
+ },
+ error: error => done.fail(`stream ended instead of reconnecting:
${error}`)
+ });
+ });
+
+ /**
+ * Logging out invalidates the session even when a refresh request is
already in flight.
+ * A late response must not put credentials back after logout cleared local
storage.
+ */
+ it('does not restore credentials after logout during a refresh', done => {
+ const refreshResponse = new Subject<any>();
+ let storedAccessToken: string | null = 'expired-token';
+ let storedRefreshToken: string | null = 'a-refresh-token';
+ localStorageService.getAuthorizationToken.and.callFake(() =>
storedAccessToken as any);
+ localStorageService.getRefreshToken.and.callFake(() => storedRefreshToken);
+ authService.refreshToken.and.returnValue(refreshResponse);
+ respondPerOutcome([401]);
+
+ const subscription = service.stream('/api/alert/sse/subscribe',
'ALERT_EVENT').subscribe();
+ setTimeout(() => {
+ expect(authService.refreshToken).toHaveBeenCalledWith('a-refresh-token',
true);
+ storedAccessToken = null;
+ storedRefreshToken = null;
+ subscription.unsubscribe();
+ refreshResponse.next({ code: 0, msg: '', data: { token: 'late-token',
refreshToken: 'late-refresh-token' } });
+ refreshResponse.complete();
+
+ setTimeout(() => {
+
expect(localStorageService.storageAuthorizationToken).not.toHaveBeenCalled();
+ expect(localStorageService.storageRefreshToken).not.toHaveBeenCalled();
+ done();
+ });
+ });
+ });
+
+ /**
+ * A session that is genuinely over has to end the observable rather than
turn into a loop
+ * of refresh attempts against an endpoint that keeps refusing.
+ */
+ it('ends the stream when the refresh is refused', done => {
+ localStorageService.getAuthorizationToken.and.returnValue('expired-token');
+ localStorageService.getRefreshToken.and.returnValue('a-refresh-token');
+ authService.refreshToken.and.returnValue(throwError(() => new
HttpErrorResponse({ status: 401 })));
+ respondPerOutcome([401]);
+ service.initialRetryDelayMillis = 0;
+
+ service.stream('/api/alert/sse/subscribe', 'ALERT_EVENT').subscribe({
+ error: error => {
+ expect(String(error)).toContain('401');
+ setTimeout(() => {
+ expect((window.fetch as jasmine.Spy).calls.count()).toBe(1);
+ expect(authService.refreshToken).toHaveBeenCalledTimes(1);
+ done();
+ }, 20);
+ }
+ });
+ });
+
+ it('retries when the refresh endpoint is temporarily unavailable', done => {
+
localStorageService.getAuthorizationToken.and.returnValues('expired-token',
'expired-token', 'fresh-token');
+ localStorageService.getRefreshToken.and.returnValue('a-refresh-token');
+ authService.refreshToken.and.returnValues(
+ throwError(() => new HttpErrorResponse({ status: 502 })),
+ of({ code: 0, msg: '', data: { token: 'fresh-token', refreshToken:
'next-refresh-token' } } as any)
+ );
+ respondPerOutcome([401, 401, ['event:ALERT_EVENT\ndata:{"id":8}\n\n']]);
+ service.initialRetryDelayMillis = 0;
+
+ const subscription = service.stream('/api/alert/sse/subscribe',
'ALERT_EVENT').subscribe({
+ next: data => {
+ expect(data).toBe('{"id":8}');
+ expect(authService.refreshToken).toHaveBeenCalledTimes(2);
+ expect((window.fetch as jasmine.Spy).calls.count()).toBe(3);
+ subscription.unsubscribe();
+ done();
+ },
+ error: error => done.fail(`temporary refresh failure ended the stream:
${error}`)
+ });
+ });
+
+ it('allows another refresh after a non-401 reconnect response', done => {
+
localStorageService.getAuthorizationToken.and.returnValues('expired-token',
'fresh-token', 'fresh-token', 'newer-token');
+ localStorageService.getRefreshToken.and.returnValue('a-refresh-token');
+ authService.refreshToken.and.returnValues(
+ of({ code: 0, msg: '', data: { token: 'fresh-token' } } as any),
+ of({ code: 0, msg: '', data: { token: 'newer-token' } } as any)
+ );
+ respondPerOutcome([401, 503, 401,
['event:ALERT_EVENT\ndata:{"id":10}\n\n']]);
+ service.initialRetryDelayMillis = 0;
+
+ const subscription = service.stream('/api/alert/sse/subscribe',
'ALERT_EVENT').subscribe({
+ next: data => {
+ expect(data).toBe('{"id":10}');
+ expect(authService.refreshToken).toHaveBeenCalledTimes(2);
+ expect((window.fetch as jasmine.Spy).calls.count()).toBe(4);
+ subscription.unsubscribe();
+ done();
+ },
+ error: error => done.fail(`a non-401 response prevented a later refresh:
${error}`)
+ });
+ });
+
+ /**
+ * A token that the server still refuses after a refresh is the end of the
session: retrying
+ * the refresh on every attempt would loop indefinitely.
+ */
+ it('refreshes only once for a credential that stays refused', done => {
+ localStorageService.getAuthorizationToken.and.returnValue('expired-token');
+ localStorageService.getRefreshToken.and.returnValue('a-refresh-token');
+ authService.refreshToken.and.returnValue(of({ code: 0, msg: '', data: {
token: 'fresh-token' } } as any));
+ respondPerOutcome([401]);
+ service.initialRetryDelayMillis = 0;
+
+ service.stream('/api/alert/sse/subscribe', 'ALERT_EVENT').subscribe({
+ error: () => {
+ setTimeout(() => {
+ expect(authService.refreshToken).toHaveBeenCalledTimes(1);
+ expect((window.fetch as jasmine.Spy).calls.count()).toBe(2);
+ done();
+ }, 20);
+ }
+ });
+ });
+});
diff --git a/web-app/src/app/service/authorized-sse.service.ts
b/web-app/src/app/service/authorized-sse.service.ts
new file mode 100644
index 0000000000..7cf9474084
--- /dev/null
+++ b/web-app/src/app/service/authorized-sse.service.ts
@@ -0,0 +1,343 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import { HttpErrorResponse } from '@angular/common/http';
+import { Injectable, NgZone } from '@angular/core';
+import { Observable, firstValueFrom } from 'rxjs';
+
+import { AuthService } from './auth.service';
+import { LocalStorageService } from './local-storage.service';
+
+type RefreshOutcome = 'refreshed' | 'refused' | 'unavailable';
+
+/** What a stream does about an attempt the server rejected. */
+type StreamRecovery = 'reconnect' | 'retry-later' | 'end';
+
+/**
+ * Spaces out the reconnect attempts of one stream, doubling the wait after
each attempt that
+ * did not get the stream running and starting over once one did.
+ */
+class ReconnectBackoff {
+ private timer?: ReturnType<typeof setTimeout>;
+
+ private delayMillis: number;
+
+ constructor(private initialMillis: number, private maxMillis: number) {
+ this.delayMillis = initialMillis;
+ }
+
+ schedule(attempt: () => void): void {
+ const wait = this.delayMillis;
+ this.delayMillis = Math.min(this.delayMillis * 2, this.maxMillis);
+ this.timer = setTimeout(attempt, wait);
+ }
+
+ /** Called once a connection is up, so the next drop is retried without any
accumulated wait. */
+ reset(): void {
+ this.delayMillis = this.initialMillis;
+ }
+
+ cancel(): void {
+ if (this.timer) {
+ clearTimeout(this.timer);
+ }
+ }
+}
+
+/**
+ * Reads a server sent event stream with the bearer token attached.
+ *
+ * The browser's own `EventSource` cannot carry an `Authorization` header,
which is why the
+ * alert and manager streams used to be reachable without any credential at
all. Reading the
+ * stream through `fetch` instead lets the token travel with the request, so
the endpoints can
+ * be moved behind the same rbac rules as the rest of the api.
+ *
+ * The returned observable starts the request on subscribe and aborts it on
unsubscribe.
+ * Events are emitted outside the angular zone; a caller that touches
component state should
+ * re-enter the zone itself, as it would with any other stream.
+ *
+ * A dropped connection is re-established rather than surfaced as an error,
because the
+ * server closes an idle subscription on purpose once its emitter times out.
`EventSource`
+ * used to reconnect on its own, so reading through `fetch` has to carry that
behaviour over
+ * or a stream would simply stop delivering.
+ *
+ * An expired token is refreshed and the connection retried, because a
subscription outlives
+ * its credential: the access token is good for an hour while the server
closes a subscription
+ * every half hour, so a reconnect eventually lands on an expired token. This
request does not
+ * go through `HttpClient`, so the interceptor that refreshes on 401 never
sees it and the
+ * refresh has to happen here. Only a credential that is still refused after a
refresh ends the
+ * observable: reconnecting cannot change that answer, and retrying would
hammer the endpoint.
+ */
+@Injectable({ providedIn: 'root' })
+export class AuthorizedSseService {
+ /**
+ * Delay before the first reconnect attempt; doubles on each consecutive
failure.
+ * An instance field rather than a constant so a test can drive the
reconnect without
+ * waiting a real second.
+ */
+ initialRetryDelayMillis = 1000;
+
+ /** Ceiling for the doubling above, so a server that stays down is polled at
a fixed rate. */
+ maxRetryDelayMillis = 30000;
+
+ /**
+ * The refresh in flight, shared by every stream this service runs. Two
streams are open at
+ * once and their tokens expire together. Sharing avoids duplicate requests
and prevents
+ * two successful responses from racing to replace the stored token pair.
+ */
+ private refreshInFlight?: Promise<RefreshOutcome>;
+
+ constructor(private localStorageService: LocalStorageService, private
authService: AuthService, private ngZone: NgZone) {}
+
+ /**
+ * @param url stream endpoint, relative to the origin
+ * @param eventName name of the sse event to emit; other events are ignored
+ * @returns the `data` payload of every matching event, as raw text
+ */
+ stream(url: string, eventName: string): Observable<string> {
+ return new Observable<string>(subscriber => {
+ let controller: AbortController | undefined;
+ let stopped = false;
+ // one refresh per rejection, so a server that refuses the fresh token
too ends the
+ // stream instead of looping between refreshing and being refused
+ let refreshed = false;
+ const backoff = new ReconnectBackoff(this.initialRetryDelayMillis,
this.maxRetryDelayMillis);
+
+ const scheduleReconnect = (): void => {
+ if (stopped) {
+ return;
+ }
+ backoff.schedule(connect);
+ };
+
+ const connect = (): void => {
+ if (stopped) {
+ return;
+ }
+ const current = new AbortController();
+ controller = current;
+
+ this.ngZone.runOutsideAngular(() => {
+ fetch(url, { method: 'GET', headers: this.authorizedHeaders(),
signal: current.signal })
+ .then(async response => {
+ // A response from the stream endpoint proves that the preceding
refresh was
+ // accepted. Even an unrelated server error must allow a later
401 to refresh
+ // again instead of permanently ending this stream.
+ if (response.status !== 401) {
+ refreshed = false;
+ }
+ if (response.status === 401 || response.status === 403) {
+ const recovery = await
this.recoverFromRejection(response.status, refreshed);
+ // abort first, so what happens next is the only thing that
happens: the
+ // handler that follows this one schedules its own reconnect
unless the
+ // attempt was aborted
+ current.abort();
+ if (recovery === 'reconnect') {
+ refreshed = true;
+ connect();
+ } else if (recovery === 'retry-later') {
+ scheduleReconnect();
+ } else {
+ stopped = true;
+ subscriber.error(new Error(`SSE request to ${url} failed
with status ${response.status}`));
+ }
+ return;
+ }
+ const reader = this.readerOf(response, url);
+ // the stream is up, so a later drop starts its backoff from the
bottom again,
+ // and the token it is holding may be refreshed again when it in
turn expires
+ backoff.reset();
+ refreshed = false;
+
+ await this.pumpEvents(reader, current.signal, eventName, data =>
subscriber.next(data));
+ })
+ .then(() => {
+ if (!stopped && !current.signal.aborted) {
+ scheduleReconnect();
+ }
+ })
+ .catch(error => {
+ if (stopped || current.signal.aborted) {
+ return;
+ }
+ console.error(`SSE connection to ${url} interrupted,
reconnecting`, error);
+ scheduleReconnect();
+ });
+ });
+ };
+
+ connect();
+
+ return () => {
+ stopped = true;
+ backoff.cancel();
+ controller?.abort();
+ };
+ });
+ }
+
+ /**
+ * @returns The headers every attempt carries, with the stored token
attached when there is
+ * one; it is read per attempt because it may have been refreshed
since the last
+ */
+ private authorizedHeaders(): Record<string, string> {
+ const headers: Record<string, string> = {
+ Accept: 'text/event-stream',
+ 'Cache-Control': 'no-cache'
+ };
+ const token = this.localStorageService.getAuthorizationToken();
+ if (token) {
+ headers['Authorization'] = `Bearer ${token}`;
+ }
+ return headers;
+ }
+
+ /**
+ * Decides what a rejected attempt means for the stream.
+ *
+ * @param status the status the stream endpoint answered with, either 401 or
403
+ * @param alreadyRefreshed whether this stream has refreshed its token since
it last had a
+ * working connection; a second rejection in a row cannot be fixed by
refreshing again
+ * @returns `reconnect` to retry at once with the fresh token, `retry-later`
to back off and
+ * try again, `end` when the answer is settled and reconnecting
cannot change it
+ */
+ private async recoverFromRejection(status: number, alreadyRefreshed:
boolean): Promise<StreamRecovery> {
+ // a refused role is not an expired token, so there is nothing to refresh
+ if (status !== 401 || alreadyRefreshed) {
+ return 'end';
+ }
+ const outcome = await this.refreshAuthToken();
+ if (outcome === 'refreshed') {
+ return 'reconnect';
+ }
+ return outcome === 'unavailable' ? 'retry-later' : 'end';
+ }
+
+ /**
+ * @returns The reader over the response body
+ * @throws Error When the response cannot be read as a stream, which the
caller turns into a
+ * reconnect the same way it treats a connection that dropped
+ */
+ private readerOf(response: Response, url: string):
ReadableStreamDefaultReader<Uint8Array> {
+ if (!response.ok) {
+ throw new Error(`SSE request to ${url} failed with status
${response.status}`);
+ }
+ const reader = response.body?.getReader();
+ if (!reader) {
+ throw new Error(`SSE response from ${url} has no readable body`);
+ }
+ return reader;
+ }
+
+ /**
+ * Reads the response until the server closes it or the caller aborts,
handing every
+ * matching event to `emit`.
+ *
+ * <p>An event can be split across chunks, so what a read returns is
appended to a buffer
+ * and only whole frames are taken off it.
+ */
+ private async pumpEvents(
+ reader: ReadableStreamDefaultReader<Uint8Array>,
+ signal: AbortSignal,
+ eventName: string,
+ emit: (data: string) => void
+ ): Promise<void> {
+ const decoder = new TextDecoder();
+ let buffer = '';
+ while (!signal.aborted) {
+ const { value, done } = await reader.read();
+ if (done) {
+ // the server closed it, most likely an emitter timeout; the caller
reconnects
+ return;
+ }
+
+ buffer += decoder.decode(value, { stream: true });
+ const frames = buffer.split(/\r?\n\r?\n/);
+ buffer = frames.pop() ?? '';
+
+ for (const frame of frames) {
+ const payload = this.payloadOf(frame, eventName);
+ if (payload !== undefined) {
+ emit(payload);
+ }
+ }
+ }
+ }
+
+ /**
+ * @returns The `data` of the frame when it carries the event this stream
subscribes to, and
+ * `undefined` for every other frame
+ */
+ private payloadOf(frame: string, eventName: string): string | undefined {
+ let frameEvent = '';
+ const dataLines: string[] = [];
+ for (const line of frame.split(/\r?\n/)) {
+ if (line.startsWith('event:')) {
+ frameEvent = line.substring(6).trim();
+ } else if (line.startsWith('data:')) {
+ dataLines.push(line.substring(5));
+ }
+ }
+ return frameEvent === eventName && dataLines.length > 0 ?
dataLines.join('\n') : undefined;
+ }
+
+ /**
+ * Exchanges the refresh token for a new pair and stores it.
+ *
+ * @returns whether the token was refreshed, explicitly refused, or
temporarily unavailable
+ */
+ private refreshAuthToken(): Promise<RefreshOutcome> {
+ if (!this.refreshInFlight) {
+ this.refreshInFlight = this.requestFreshToken().finally(() =>
(this.refreshInFlight = undefined));
+ }
+ return this.refreshInFlight;
+ }
+
+ private async requestFreshToken(): Promise<RefreshOutcome> {
+ const refreshToken = this.localStorageService.getRefreshToken();
+ if (!refreshToken) {
+ return 'refused';
+ }
+ try {
+ const message = await
firstValueFrom(this.authService.refreshToken(refreshToken, true));
+ if (message.code !== 0 || !message.data?.token) {
+ console.warn('SSE token refresh was refused', message.msg);
+ return 'refused';
+ }
+ // The request may finish after logout, or after another refresh/login
replaced the
+ // session. Never let that stale response restore or overwrite
credentials. If another
+ // session is already present, its access token is the one the reconnect
should use.
+ if (this.localStorageService.getRefreshToken() !== refreshToken) {
+ return this.localStorageService.getAuthorizationToken() === null ?
'refused' : 'refreshed';
+ }
+ this.localStorageService.storageAuthorizationToken(message.data.token);
+ if (message.data.refreshToken) {
+
this.localStorageService.storageRefreshToken(message.data.refreshToken);
+ }
+ return 'refreshed';
+ } catch (error) {
+ if (error instanceof HttpErrorResponse && (error.status === 401 ||
error.status === 403)) {
+ console.warn('SSE token refresh was refused', error.status);
+ return 'refused';
+ }
+ console.warn('SSE token refresh is temporarily unavailable', error);
+ return 'unavailable';
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
