feat(clients): CLI-15 surface ReplayGap reconnect sentinel as typed signal (4/5 clients)
The gateway emits a ReplayGap sentinel MxEvent at the head of a StreamEvents stream resumed via after_worker_sequence when the requested cursor predates the oldest retained event. Clients previously ignored it, silently mis-treating a lossy resume as continuous. Each client now surfaces the sentinel as a distinct, typed, non-terminal signal (never synthesized, never swallowed) so a consumer can detect the gap and re-snapshot; resume contract is after_worker_sequence = oldest_available_sequence - 1. - .NET: MxEventStreamItem (IsReplayGap/ReplayGap/Event) via new StreamEventItemsAsync + AsStreamItemsAsync extension. Build clean, 87 passed. - Go: EventResult.ReplayGap field + IsReplayGap(); ReplayGap type alias. build/vet/test clean. - Rust: EventItem enum (Event/ReplayGap); EventStream now yields Result<EventItem, Error>; CLI renders REPLAY_GAP line / replayGap JSON. fmt/check/test/clippy clean. - Python: ReplayGap dataclass; stream_events yields pb.MxEvent | ReplayGap. 131 passed. - Shared docs: ClientLibrariesDesign non-goals reframed (reconnect-replay protocol is consumable; auto-reconnect stays a non-goal); CrossLanguageSmokeMatrix resume-gap note. Java client is deferred to the windev batch (no local JRE); CLI-15 stays open until it lands. Claude-Session: https://claude.ai/code/session_01DMXXvNuPekkkrTEyPNxEkW
This commit is contained in:
Joseph Doherty
@@ -137,6 +137,47 @@ redaction. Per-item bulk failures are reported inside each result entry
|
||||
`invoke_raw` / `client.invoke_raw` escape hatch performs neither check and
|
||||
returns the unvalidated reply.
|
||||
|
||||
## Event Streaming And Reconnect-Replay Gaps
|
||||
|
||||
`session.events()` / `session.events_after(after_worker_sequence)` (and the
|
||||
lower-level `client.stream_events`) return an `EventStream` that yields
|
||||
`EventItem` values, not bare `MxEvent`s:
|
||||
|
||||
```rust
|
||||
use zb_mom_ww_mxgateway_client::EventItem;
|
||||
|
||||
let mut stream = session.events_after(cursor).await?;
|
||||
while let Some(item) = stream.next().await {
|
||||
match item? {
|
||||
EventItem::Event(event) => { /* apply the MXAccess change */ }
|
||||
EventItem::ReplayGap(gap) => {
|
||||
// Recent history was evicted — discard local state and re-snapshot,
|
||||
// then resume without provoking another gap:
|
||||
let resume = gap.oldest_available_sequence.saturating_sub(1);
|
||||
stream = session.events_after(resume).await?;
|
||||
}
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
Almost every item is a normal `EventItem::Event`. `EventItem::ReplayGap` is a
|
||||
faithful, typed surfacing of the gateway's reconnect-replay gap sentinel — the
|
||||
client does not synthesize it. The gateway emits the sentinel at most once, at
|
||||
the head of a stream **resumed** via `events_after` (`after_worker_sequence`)
|
||||
when the requested sequence is older than the oldest event still retained in the
|
||||
session's replay ring: events in the open interval
|
||||
`(requested_after_sequence, oldest_available_sequence)` were evicted and cannot
|
||||
be replayed. A `ReplayGap` therefore means "you missed events — discard any
|
||||
local state and re-snapshot." To resume without a second gap, reconnect with
|
||||
`events_after(gap.oldest_available_sequence - 1)`, which replays starting at the
|
||||
first still-retained event. A stream opened from the beginning
|
||||
(`session.events()` / `events_after(0)`) never produces a `ReplayGap`.
|
||||
|
||||
`EventItem` provides `as_event()`, `into_event()`, and `replay_gap()` accessors
|
||||
for callers that prefer not to `match`. The `mxgw-cli stream-events` subcommand
|
||||
renders the sentinel as a distinct `REPLAY_GAP …` line (or a `replayGap` JSON
|
||||
object under `--json` / `--jsonl`).
|
||||
|
||||
## Write Semantics And Common Pitfalls
|
||||
|
||||
These are MXAccess parity behaviors that surprise new callers. The gateway
|
||||
|
||||
@@ -28,8 +28,8 @@ use zb_mom_ww_mxgateway_client::generated::mxaccess_gateway::v1::{
|
||||
WriteSecuredBulkEntry,
|
||||
};
|
||||
use zb_mom_ww_mxgateway_client::{
|
||||
next_correlation_id, ApiKey, ClientOptions, Error, GalaxyClient, GatewayClient, MxValue,
|
||||
MxValueProjection, CLIENT_VERSION, GATEWAY_PROTOCOL_VERSION, WORKER_PROTOCOL_VERSION,
|
||||
next_correlation_id, ApiKey, ClientOptions, Error, EventItem, GalaxyClient, GatewayClient,
|
||||
MxValue, MxValueProjection, CLIENT_VERSION, GATEWAY_PROTOCOL_VERSION, WORKER_PROTOCOL_VERSION,
|
||||
};
|
||||
|
||||
const MAX_AGGREGATE_EVENTS: usize = 10_000;
|
||||
@@ -886,17 +886,43 @@ async fn dispatch(command: Command) -> Result<(), Error> {
|
||||
let mut events: Vec<Value> = Vec::new();
|
||||
let mut event_count = 0usize;
|
||||
while event_count < max_events {
|
||||
let Some(event) = stream.next().await else {
|
||||
let Some(item) = stream.next().await else {
|
||||
break;
|
||||
};
|
||||
let event = event?;
|
||||
let item = item?;
|
||||
event_count += 1;
|
||||
if jsonl {
|
||||
println!("{}", event_to_json(&event));
|
||||
} else if json {
|
||||
events.push(event_to_json(&event));
|
||||
} else {
|
||||
println!("{} {}", event.worker_sequence, event.family);
|
||||
match item {
|
||||
EventItem::Event(event) => {
|
||||
if jsonl {
|
||||
println!("{}", event_to_json(&event));
|
||||
} else if json {
|
||||
events.push(event_to_json(&event));
|
||||
} else {
|
||||
println!("{} {}", event.worker_sequence, event.family);
|
||||
}
|
||||
}
|
||||
// Reconnect-replay gap sentinel: recent history was evicted
|
||||
// before this resumed stream could replay it. Render it as a
|
||||
// distinct row so the caller can re-snapshot and resume with
|
||||
// `oldest_available_sequence - 1`.
|
||||
EventItem::ReplayGap(gap) => {
|
||||
let value = json!({
|
||||
"replayGap": {
|
||||
"requestedAfterSequence": gap.requested_after_sequence,
|
||||
"oldestAvailableSequence": gap.oldest_available_sequence,
|
||||
}
|
||||
});
|
||||
if jsonl {
|
||||
println!("{value}");
|
||||
} else if json {
|
||||
events.push(value);
|
||||
} else {
|
||||
println!(
|
||||
"REPLAY_GAP requested_after={} oldest_available={}",
|
||||
gap.requested_after_sequence, gap.oldest_available_sequence
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
if json {
|
||||
|
||||
+121
-8
@@ -18,7 +18,7 @@ use crate::generated::mxaccess_gateway::v1::mx_access_gateway_client::MxAccessGa
|
||||
use crate::generated::mxaccess_gateway::v1::{
|
||||
AcknowledgeAlarmReply, AcknowledgeAlarmRequest, ActiveAlarmSnapshot, AlarmFeedMessage,
|
||||
CloseSessionReply, CloseSessionRequest, MxCommandReply, MxCommandRequest, MxEvent,
|
||||
OpenSessionReply, OpenSessionRequest, QueryActiveAlarmsRequest, StreamAlarmsRequest,
|
||||
OpenSessionReply, OpenSessionRequest, QueryActiveAlarmsRequest, ReplayGap, StreamAlarmsRequest,
|
||||
StreamEventsRequest,
|
||||
};
|
||||
use crate::options::{build_tls_config, ClientOptions};
|
||||
@@ -28,11 +28,120 @@ use crate::session::Session;
|
||||
/// [`GatewayClient`] uses internally.
|
||||
pub type RawGatewayClient = MxAccessGatewayClient<InterceptedService<Channel, AuthInterceptor>>;
|
||||
|
||||
/// Pinned, boxed [`MxEvent`] stream returned by
|
||||
/// [`GatewayClient::stream_events`]. Errors are pre-mapped from
|
||||
/// `tonic::Status` to [`Error`]; dropping the stream cancels the call.
|
||||
/// One item yielded by the per-session event stream returned by
|
||||
/// [`GatewayClient::stream_events`].
|
||||
///
|
||||
/// Almost every item is an ordinary MXAccess event ([`EventItem::Event`]).
|
||||
/// The one exception is the reconnect-replay gap sentinel
|
||||
/// ([`EventItem::ReplayGap`]): the gateway emits it at most once, at the head
|
||||
/// of a stream that was *resumed* via
|
||||
/// [`Session::events_after`](crate::session::Session::events_after)
|
||||
/// (`StreamEventsRequest.after_worker_sequence`) when the requested sequence is
|
||||
/// older than the oldest event still retained in the session replay ring — i.e.
|
||||
/// events were evicted and cannot be replayed.
|
||||
///
|
||||
/// The client does **not** synthesize this signal: it faithfully forwards the
|
||||
/// gateway's sentinel `MxEvent` (whose `replay_gap` field is set), and only
|
||||
/// makes it a distinct, typed variant so consumers can `match` on it instead of
|
||||
/// inspecting a field on a value that otherwise looks like a normal event.
|
||||
///
|
||||
/// # Reacting to a gap
|
||||
///
|
||||
/// A [`EventItem::ReplayGap`] means "you missed events — discard any local
|
||||
/// state and re-snapshot." The events in the open interval
|
||||
/// `(requested_after_sequence, oldest_available_sequence)` are gone. To resume
|
||||
/// the stream without provoking another gap, reconnect with
|
||||
/// [`Session::events_after`](crate::session::Session::events_after) passing
|
||||
/// `oldest_available_sequence - 1`, which replays starting at the first still
|
||||
/// retained event (`oldest_available_sequence`):
|
||||
///
|
||||
/// ```no_run
|
||||
/// # use zb_mom_ww_mxgateway_client::{EventItem, Session};
|
||||
/// # use futures_util::StreamExt;
|
||||
/// # async fn run(session: Session, cursor: u64) -> Result<(), zb_mom_ww_mxgateway_client::Error> {
|
||||
/// let mut stream = session.events_after(cursor).await?;
|
||||
/// while let Some(item) = stream.next().await {
|
||||
/// match item? {
|
||||
/// EventItem::Event(event) => {
|
||||
/// let _ = event; // apply the change
|
||||
/// }
|
||||
/// EventItem::ReplayGap(gap) => {
|
||||
/// // Local state is stale — re-snapshot, then resume without a gap.
|
||||
/// let resume_cursor = gap.oldest_available_sequence.saturating_sub(1);
|
||||
/// stream = session.events_after(resume_cursor).await?;
|
||||
/// }
|
||||
/// }
|
||||
/// }
|
||||
/// # Ok(())
|
||||
/// # }
|
||||
/// ```
|
||||
// The `Event` variant is the hot path (nearly every stream item) and is the
|
||||
// large one; the rare `ReplayGap` sentinel is small. Boxing `Event` to equalize
|
||||
// the variants would add a heap allocation to every streamed event — a
|
||||
// regression versus the prior `Result<MxEvent, Error>` surface, which already
|
||||
// moved `MxEvent` by value. Keep the common path allocation-free.
|
||||
#[allow(clippy::large_enum_variant)]
|
||||
#[derive(Clone, Debug, PartialEq)]
|
||||
pub enum EventItem {
|
||||
/// A normal MXAccess event forwarded from the worker.
|
||||
Event(MxEvent),
|
||||
/// The reconnect-replay gap sentinel — recent event history was evicted
|
||||
/// before this resumed stream could replay it. See [`EventItem`] for how
|
||||
/// to react.
|
||||
ReplayGap(ReplayGap),
|
||||
}
|
||||
|
||||
impl EventItem {
|
||||
/// Classify an incoming `MxEvent` into the typed stream item.
|
||||
///
|
||||
/// A present `replay_gap` promotes the event to [`EventItem::ReplayGap`];
|
||||
/// otherwise it is an ordinary [`EventItem::Event`]. The sentinel is never
|
||||
/// dropped and never surfaced as a normal event.
|
||||
fn from_event(mut event: MxEvent) -> Self {
|
||||
match event.replay_gap.take() {
|
||||
Some(gap) => EventItem::ReplayGap(gap),
|
||||
None => EventItem::Event(event),
|
||||
}
|
||||
}
|
||||
|
||||
/// Borrow the inner [`MxEvent`] when this item is a normal event, or
|
||||
/// `None` when it is the [`EventItem::ReplayGap`] sentinel.
|
||||
#[must_use]
|
||||
pub fn as_event(&self) -> Option<&MxEvent> {
|
||||
match self {
|
||||
EventItem::Event(event) => Some(event),
|
||||
EventItem::ReplayGap(_) => None,
|
||||
}
|
||||
}
|
||||
|
||||
/// Borrow the [`ReplayGap`] when this item is the reconnect-replay gap
|
||||
/// sentinel, or `None` for a normal event.
|
||||
#[must_use]
|
||||
pub fn replay_gap(&self) -> Option<&ReplayGap> {
|
||||
match self {
|
||||
EventItem::ReplayGap(gap) => Some(gap),
|
||||
EventItem::Event(_) => None,
|
||||
}
|
||||
}
|
||||
|
||||
/// Consume the item and return the inner [`MxEvent`] when it is a normal
|
||||
/// event, or `None` for the [`EventItem::ReplayGap`] sentinel.
|
||||
#[must_use]
|
||||
pub fn into_event(self) -> Option<MxEvent> {
|
||||
match self {
|
||||
EventItem::Event(event) => Some(event),
|
||||
EventItem::ReplayGap(_) => None,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Pinned, boxed [`EventItem`] stream returned by
|
||||
/// [`GatewayClient::stream_events`]. Each item is either a normal
|
||||
/// [`EventItem::Event`] or the [`EventItem::ReplayGap`] reconnect-replay
|
||||
/// sentinel. Errors are pre-mapped from `tonic::Status` to [`Error`]; dropping
|
||||
/// the stream cancels the call.
|
||||
pub type EventStream =
|
||||
std::pin::Pin<Box<dyn futures_core::Stream<Item = Result<MxEvent, Error>> + Send + 'static>>;
|
||||
std::pin::Pin<Box<dyn futures_core::Stream<Item = Result<EventItem, Error>> + Send + 'static>>;
|
||||
|
||||
/// Pinned, boxed [`ActiveAlarmSnapshot`] stream returned by
|
||||
/// [`GatewayClient::query_active_alarms`]. Errors are pre-mapped from
|
||||
@@ -190,8 +299,12 @@ impl GatewayClient {
|
||||
|
||||
/// Open the server-streaming `StreamEvents` RPC.
|
||||
///
|
||||
/// The returned [`EventStream`] yields `MxEvent` messages as the worker
|
||||
/// produces them. Dropping the stream cancels the gRPC call cooperatively.
|
||||
/// The returned [`EventStream`] yields [`EventItem`] values as the worker
|
||||
/// produces them: ordinary MXAccess events as [`EventItem::Event`], and the
|
||||
/// gateway's reconnect-replay gap sentinel — set only on resumed streams
|
||||
/// whose requested sequence predates the retained replay history — as
|
||||
/// [`EventItem::ReplayGap`]. Dropping the stream cancels the gRPC call
|
||||
/// cooperatively.
|
||||
///
|
||||
/// # Errors
|
||||
///
|
||||
@@ -201,7 +314,7 @@ impl GatewayClient {
|
||||
let mut client = self.inner.clone();
|
||||
let response = client.stream_events(self.stream_request(request)).await?;
|
||||
let stream = futures_util::StreamExt::map(response.into_inner(), |result| {
|
||||
result.map_err(Error::from)
|
||||
result.map(EventItem::from_event).map_err(Error::from)
|
||||
});
|
||||
|
||||
Ok(Box::pin(stream))
|
||||
|
||||
@@ -24,12 +24,14 @@ pub mod version;
|
||||
#[doc(inline)]
|
||||
pub use auth::{ApiKey, AuthInterceptor};
|
||||
#[doc(inline)]
|
||||
pub use client::{AlarmFeedStream, EventStream, GatewayClient};
|
||||
pub use client::{AlarmFeedStream, EventItem, EventStream, GatewayClient};
|
||||
#[doc(inline)]
|
||||
pub use error::{CommandError, Error, MxAccessError};
|
||||
#[doc(inline)]
|
||||
pub use galaxy::{DeployEventStream, GalaxyClient};
|
||||
#[doc(inline)]
|
||||
pub use generated::mxaccess_gateway::v1::ReplayGap;
|
||||
#[doc(inline)]
|
||||
pub use options::ClientOptions;
|
||||
#[doc(inline)]
|
||||
pub use session::{next_correlation_id, Session};
|
||||
|
||||
@@ -630,6 +630,13 @@ impl Session {
|
||||
|
||||
/// Open the per-session event stream from the beginning.
|
||||
///
|
||||
/// The returned [`EventStream`] yields [`EventItem`](crate::EventItem)
|
||||
/// values — normal MXAccess events as
|
||||
/// [`EventItem::Event`](crate::EventItem::Event). A stream opened from the
|
||||
/// beginning never produces a
|
||||
/// [`EventItem::ReplayGap`](crate::EventItem::ReplayGap); that sentinel
|
||||
/// only appears on a resumed stream (see [`Session::events_after`]).
|
||||
///
|
||||
/// # Errors
|
||||
///
|
||||
/// Returns the `tonic::Status` mapped through [`Error::from`] when the
|
||||
@@ -642,6 +649,15 @@ impl Session {
|
||||
/// `worker_sequence` is greater than `after_worker_sequence`. Pass `0`
|
||||
/// to receive every buffered event.
|
||||
///
|
||||
/// If `after_worker_sequence` predates the oldest event still retained in
|
||||
/// the gateway's replay ring, the stream opens with a single
|
||||
/// [`EventItem::ReplayGap`](crate::EventItem::ReplayGap) sentinel: recent
|
||||
/// history was evicted and cannot be replayed, so the caller must discard
|
||||
/// any local state and re-snapshot. To resume without provoking another
|
||||
/// gap, call this method again with
|
||||
/// `gap.oldest_available_sequence - 1`. See
|
||||
/// [`EventItem`](crate::EventItem) for the full contract.
|
||||
///
|
||||
/// # Errors
|
||||
///
|
||||
/// Same conditions as [`Session::events`].
|
||||
|
||||
@@ -27,13 +27,13 @@ use zb_mom_ww_mxgateway_client::generated::mxaccess_gateway::v1::{
|
||||
CloseSessionReply, CloseSessionRequest, MxCommandKind, MxCommandReply, MxDataType, MxEvent,
|
||||
MxEventFamily, MxSparseArray, MxSparseElement, MxStatusCategory, MxStatusProxy, MxStatusSource,
|
||||
MxValue, OnAlarmTransitionEvent, OpenSessionReply, OpenSessionRequest, ProtocolStatus,
|
||||
ProtocolStatusCode, QueryActiveAlarmsRequest, RegisterReply, SessionState, StreamAlarmsRequest,
|
||||
StreamEventsRequest, SubscribeResult, Write2BulkEntry, WriteBulkEntry, WriteCommand,
|
||||
WriteSecured2BulkEntry, WriteSecuredBulkEntry,
|
||||
ProtocolStatusCode, QueryActiveAlarmsRequest, RegisterReply, ReplayGap, SessionState,
|
||||
StreamAlarmsRequest, StreamEventsRequest, SubscribeResult, Write2BulkEntry, WriteBulkEntry,
|
||||
WriteCommand, WriteSecured2BulkEntry, WriteSecuredBulkEntry,
|
||||
};
|
||||
use zb_mom_ww_mxgateway_client::{
|
||||
next_correlation_id, ApiKey, ClientOptions, CommandError, Error, GatewayClient, MxStatus,
|
||||
MxValue as ClientMxValue, MxValueProjection,
|
||||
next_correlation_id, ApiKey, ClientOptions, CommandError, Error, EventItem, GatewayClient,
|
||||
MxStatus, MxValue as ClientMxValue, MxValueProjection,
|
||||
};
|
||||
|
||||
#[tokio::test]
|
||||
@@ -128,8 +128,28 @@ async fn event_stream_preserves_order_and_drop_cancels_server_stream() {
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
assert_eq!(stream.next().await.unwrap().unwrap().worker_sequence, 1);
|
||||
assert_eq!(stream.next().await.unwrap().unwrap().worker_sequence, 2);
|
||||
assert_eq!(
|
||||
stream
|
||||
.next()
|
||||
.await
|
||||
.unwrap()
|
||||
.unwrap()
|
||||
.as_event()
|
||||
.unwrap()
|
||||
.worker_sequence,
|
||||
1
|
||||
);
|
||||
assert_eq!(
|
||||
stream
|
||||
.next()
|
||||
.await
|
||||
.unwrap()
|
||||
.unwrap()
|
||||
.as_event()
|
||||
.unwrap()
|
||||
.worker_sequence,
|
||||
2
|
||||
);
|
||||
|
||||
drop(stream);
|
||||
for _ in 0..20 {
|
||||
@@ -142,6 +162,55 @@ async fn event_stream_preserves_order_and_drop_cancels_server_stream() {
|
||||
assert!(state.stream_dropped.load(Ordering::SeqCst));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn replay_gap_sentinel_surfaces_as_typed_event_item() {
|
||||
let state = Arc::new(FakeState::default());
|
||||
// Script a resumed stream: the reconnect-replay gap sentinel at the head
|
||||
// (family UNSPECIFIED, no body, `replay_gap` set) followed by a normal
|
||||
// event. The client must promote the sentinel to `EventItem::ReplayGap`
|
||||
// and leave the following event as a normal `EventItem::Event`.
|
||||
*state.stream_events_script.lock().await = Some(vec![
|
||||
MxEvent {
|
||||
replay_gap: Some(ReplayGap {
|
||||
requested_after_sequence: 5,
|
||||
oldest_available_sequence: 42,
|
||||
}),
|
||||
..MxEvent::default()
|
||||
},
|
||||
event(42),
|
||||
]);
|
||||
let endpoint = spawn_fake_gateway(state.clone()).await;
|
||||
let client = GatewayClient::connect(ClientOptions::new(endpoint))
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
let mut stream = client
|
||||
.stream_events(StreamEventsRequest {
|
||||
session_id: "session-fixture".to_owned(),
|
||||
after_worker_sequence: 5,
|
||||
})
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
// First item is the typed gap sentinel, not a normal event.
|
||||
let first = stream.next().await.unwrap().unwrap();
|
||||
match &first {
|
||||
EventItem::ReplayGap(gap) => {
|
||||
assert_eq!(gap.requested_after_sequence, 5);
|
||||
assert_eq!(gap.oldest_available_sequence, 42);
|
||||
}
|
||||
EventItem::Event(_) => panic!("expected a ReplayGap sentinel, got a normal event"),
|
||||
}
|
||||
// Accessor helpers reflect the variant.
|
||||
assert!(first.as_event().is_none());
|
||||
assert_eq!(first.replay_gap().unwrap().oldest_available_sequence, 42);
|
||||
|
||||
// The normal event that follows is unaffected.
|
||||
let second = stream.next().await.unwrap().unwrap();
|
||||
assert_eq!(second.as_event().unwrap().worker_sequence, 42);
|
||||
assert!(second.replay_gap().is_none());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn acknowledge_alarm_returns_reply_with_native_status() {
|
||||
let state = Arc::new(FakeState::default());
|
||||
@@ -672,6 +741,11 @@ struct FakeState {
|
||||
/// handler to emit a synthetic ConditionRefresh -> snapshot_complete
|
||||
/// -> transition sequence.
|
||||
stream_alarms_script: Mutex<Option<Vec<AlarmFeedMessage>>>,
|
||||
/// Optional per-test override that pins the fake's `StreamEvents`
|
||||
/// handler to emit a scripted `MxEvent` sequence (e.g. a `replay_gap`
|
||||
/// sentinel followed by a normal event). When `None`, the handler falls
|
||||
/// back to the default `event(1)` / `event(2)` pair.
|
||||
stream_events_script: Mutex<Option<Vec<MxEvent>>>,
|
||||
}
|
||||
|
||||
/// Per-test override for the fake's `Invoke` handler.
|
||||
@@ -921,9 +995,12 @@ impl MxAccessGateway for FakeGateway {
|
||||
&self,
|
||||
_request: Request<StreamEventsRequest>,
|
||||
) -> Result<Response<Self::StreamEventsStream>, Status> {
|
||||
let (sender, receiver) = mpsc::channel(4);
|
||||
sender.send(Ok(event(1))).await.unwrap();
|
||||
sender.send(Ok(event(2))).await.unwrap();
|
||||
let script = self.state.stream_events_script.lock().await.take();
|
||||
let events = script.unwrap_or_else(|| vec![event(1), event(2)]);
|
||||
let (sender, receiver) = mpsc::channel(events.len().max(1));
|
||||
for event in events {
|
||||
sender.send(Ok(event)).await.unwrap();
|
||||
}
|
||||
|
||||
Ok(Response::new(DropAwareStream {
|
||||
inner: ReceiverStream::new(receiver),
|
||||
|
||||
Reference in New Issue
Block a user