refactor(managed): ensure event hub disconnection on all exit paths
Wrap the client execution in a cleanup handler to guarantee that event hub subscriptions are disconnected regardless of whether the session completes successfully or returns an error.
This commit is contained in:
1 parent
0a9054a51d
commit
a773a94a45
1 file changed
+12
-3
@@ -86,7 +86,8 @@ impl ManagedClient {
|
|||||||
let cancellation = CancellationToken::new();
|
let cancellation = CancellationToken::new();
|
||||||
let worker_hub = hub.clone();
|
let worker_hub = hub.clone();
|
||||||
let worker_cancellation = cancellation.clone();
|
let worker_cancellation = cancellation.clone();
|
||||||
let worker = thread::spawn(move || run_client(client, worker_hub, worker_cancellation));
|
let worker =
|
||||||
|
thread::spawn(move || run_client_with_cleanup(client, worker_hub, worker_cancellation));
|
||||||
|
|
||||||
Ok((
|
Ok((
|
||||||
ManagedClientSession {
|
ManagedClientSession {
|
||||||
@@ -143,7 +144,6 @@ fn run_client(
|
|||||||
cancellation: CancellationToken,
|
cancellation: CancellationToken,
|
||||||
) -> Result<SessionOutcome, ClientError> {
|
) -> Result<SessionOutcome, ClientError> {
|
||||||
if client.is_run_complete() {
|
if client.is_run_complete() {
|
||||||
hub.disconnect_all();
|
|
||||||
return Ok(SessionOutcome {
|
return Ok(SessionOutcome {
|
||||||
end_reason: SessionEndReason::NoTestComplete,
|
end_reason: SessionEndReason::NoTestComplete,
|
||||||
packets_sent: 0,
|
packets_sent: 0,
|
||||||
@@ -204,7 +204,6 @@ fn run_client(
|
|||||||
let packets_sent = client.packets_sent();
|
let packets_sent = client.packets_sent();
|
||||||
let close_events = client.close(ClientTimestamp::now())?;
|
let close_events = client.close(ClientTimestamp::now())?;
|
||||||
publish_events(&hub, &mut counters, close_events);
|
publish_events(&hub, &mut counters, close_events);
|
||||||
hub.disconnect_all();
|
|
||||||
|
|
||||||
Ok(SessionOutcome {
|
Ok(SessionOutcome {
|
||||||
end_reason: if cancelled {
|
end_reason: if cancelled {
|
||||||
@@ -220,6 +219,16 @@ fn run_client(
|
|||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
|
fn run_client_with_cleanup(
|
||||||
|
client: Client,
|
||||||
|
hub: EventHub,
|
||||||
|
cancellation: CancellationToken,
|
||||||
|
) -> Result<SessionOutcome, ClientError> {
|
||||||
|
let outcome = run_client(client, hub.clone(), cancellation);
|
||||||
|
hub.disconnect_all();
|
||||||
|
outcome
|
||||||
|
}
|
||||||
|
|
||||||
fn publish_events(hub: &EventHub, counters: &mut OutcomeCounters, events: Vec<ClientEvent>) {
|
fn publish_events(hub: &EventHub, counters: &mut OutcomeCounters, events: Vec<ClientEvent>) {
|
||||||
for event in events {
|
for event in events {
|
||||||
counters.observe(&event);
|
counters.observe(&event);
|
||||||
|
|||||||
Reference in new issue
Block a user