refactor(client): rename ClientState to ClientPhase and introduce ActiveSession
This commit is contained in:
2 files changed
+55
-25
No files matched your search
@@ -27,7 +27,8 @@ pub struct Client {
|
|||||||
remote: SocketAddr,
|
remote: SocketAddr,
|
||||||
requested: Params,
|
requested: Params,
|
||||||
negotiated: Option<NegotiatedParams>,
|
negotiated: Option<NegotiatedParams>,
|
||||||
state: ClientState,
|
phase: ClientPhase,
|
||||||
|
session: Option<ActiveSession>,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl Client {
|
impl Client {
|
||||||
@@ -43,16 +44,17 @@ impl Client {
|
|||||||
remote,
|
remote,
|
||||||
requested,
|
requested,
|
||||||
negotiated: None,
|
negotiated: None,
|
||||||
state: ClientState::Connected,
|
phase: ClientPhase::Connected,
|
||||||
|
session: None,
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
pub fn open(&mut self, now: ClientTimestamp) -> Result<OpenOutcome, ClientError> {
|
pub fn open(&mut self, now: ClientTimestamp) -> Result<OpenOutcome, ClientError> {
|
||||||
match self.state {
|
match self.phase {
|
||||||
ClientState::Connected => {}
|
ClientPhase::Connected => {}
|
||||||
ClientState::Open { .. } => return Err(ClientError::AlreadyOpen),
|
ClientPhase::Open { .. } => return Err(ClientError::AlreadyOpen),
|
||||||
ClientState::Closed => return Err(ClientError::AlreadyClosed),
|
ClientPhase::Closed => return Err(ClientError::AlreadyClosed),
|
||||||
ClientState::NoTestCompleted => return Err(ClientError::AlreadyCompleted),
|
ClientPhase::NoTestCompleted => return Err(ClientError::AlreadyCompleted),
|
||||||
}
|
}
|
||||||
|
|
||||||
let outcome = self.open_inner(now);
|
let outcome = self.open_inner(now);
|
||||||
@@ -61,15 +63,7 @@ impl Client {
|
|||||||
.set_read_timeout(self.config.socket_config.recv_timeout);
|
.set_read_timeout(self.config.socket_config.recv_timeout);
|
||||||
match (outcome, restore) {
|
match (outcome, restore) {
|
||||||
(Ok(outcome), Ok(())) => Ok(outcome),
|
(Ok(outcome), Ok(())) => Ok(outcome),
|
||||||
(Ok(outcome), Err(_)) => {
|
(Ok(outcome), Err(_)) => Ok(outcome),
|
||||||
// Open succeeded and state already advanced. Treating a
|
|
||||||
// read-timeout restore failure as fatal would leave the
|
|
||||||
// caller believing open failed while internal state is
|
|
||||||
// Open/NoTestCompleted. The socket remains usable; the
|
|
||||||
// worst case is a stale read timeout which Milestone 3
|
|
||||||
// send/recv methods will set per-operation anyway.
|
|
||||||
Ok(outcome)
|
|
||||||
}
|
|
||||||
(Err(err), Ok(())) => Err(err),
|
(Err(err), Ok(())) => Err(err),
|
||||||
(Err(err), Err(_)) => Err(err),
|
(Err(err), Err(_)) => Err(err),
|
||||||
}
|
}
|
||||||
@@ -105,10 +99,10 @@ impl Client {
|
|||||||
}
|
}
|
||||||
|
|
||||||
pub fn close(&mut self, now: ClientTimestamp) -> Result<Vec<ClientEvent>, ClientError> {
|
pub fn close(&mut self, now: ClientTimestamp) -> Result<Vec<ClientEvent>, ClientError> {
|
||||||
let token = match self.state {
|
let token = match self.phase {
|
||||||
ClientState::Open { token } => token,
|
ClientPhase::Open { token } => token,
|
||||||
ClientState::Closed => return Err(ClientError::AlreadyClosed),
|
ClientPhase::Closed => return Err(ClientError::AlreadyClosed),
|
||||||
ClientState::Connected | ClientState::NoTestCompleted => {
|
ClientPhase::Connected | ClientPhase::NoTestCompleted => {
|
||||||
return Err(ClientError::NotOpen)
|
return Err(ClientError::NotOpen)
|
||||||
}
|
}
|
||||||
};
|
};
|
||||||
@@ -116,7 +110,8 @@ impl Client {
|
|||||||
let packet =
|
let packet =
|
||||||
encode_close_request(&CloseRequest { token }, self.config.hmac_key.as_deref())?;
|
encode_close_request(&CloseRequest { token }, self.config.hmac_key.as_deref())?;
|
||||||
self.socket.send(&packet)?;
|
self.socket.send(&packet)?;
|
||||||
self.state = ClientState::Closed;
|
self.phase = ClientPhase::Closed;
|
||||||
|
self.session = None;
|
||||||
|
|
||||||
Ok(vec![ClientEvent::SessionClosed {
|
Ok(vec![ClientEvent::SessionClosed {
|
||||||
remote: self.remote,
|
remote: self.remote,
|
||||||
@@ -164,7 +159,28 @@ impl Client {
|
|||||||
params: reply.params.clone(),
|
params: reply.params.clone(),
|
||||||
};
|
};
|
||||||
self.negotiated = Some(negotiated.clone());
|
self.negotiated = Some(negotiated.clone());
|
||||||
self.state = ClientState::Open { token: reply.token };
|
self.phase = ClientPhase::Open { token: reply.token };
|
||||||
|
|
||||||
|
let duration_ns = negotiated.params.duration_ns;
|
||||||
|
let end_mono = if duration_ns > 0 {
|
||||||
|
Some(now.mono + Duration::from_nanos(duration_ns as u64))
|
||||||
|
} else {
|
||||||
|
None
|
||||||
|
};
|
||||||
|
|
||||||
|
self.session = Some(ActiveSession {
|
||||||
|
next_wire_seq: 0,
|
||||||
|
next_logical_seq: 0,
|
||||||
|
highest_received_seq: None,
|
||||||
|
packets_sent: 0,
|
||||||
|
start_mono: now.mono,
|
||||||
|
end_mono,
|
||||||
|
next_send_at: now.mono,
|
||||||
|
pending: PendingMap::new(self.config.max_pending_probes),
|
||||||
|
completed: CompletedSet::new(self.config.max_pending_probes),
|
||||||
|
sending_done: false,
|
||||||
|
});
|
||||||
|
|
||||||
let event = ClientEvent::SessionStarted {
|
let event = ClientEvent::SessionStarted {
|
||||||
remote: self.remote,
|
remote: self.remote,
|
||||||
token: reply.token,
|
token: reply.token,
|
||||||
@@ -194,7 +210,7 @@ impl Client {
|
|||||||
params: reply.params.clone(),
|
params: reply.params.clone(),
|
||||||
};
|
};
|
||||||
self.negotiated = Some(negotiated.clone());
|
self.negotiated = Some(negotiated.clone());
|
||||||
self.state = ClientState::NoTestCompleted;
|
self.phase = ClientPhase::NoTestCompleted;
|
||||||
let event = ClientEvent::NoTestCompleted {
|
let event = ClientEvent::NoTestCompleted {
|
||||||
remote: self.remote,
|
remote: self.remote,
|
||||||
negotiated: negotiated.clone(),
|
negotiated: negotiated.clone(),
|
||||||
@@ -393,7 +409,7 @@ mod tests {
|
|||||||
|
|
||||||
let negotiated = assert_open_started(client.open(ClientTimestamp::now()).unwrap());
|
let negotiated = assert_open_started(client.open(ClientTimestamp::now()).unwrap());
|
||||||
assert_eq!(negotiated.params, params);
|
assert_eq!(negotiated.params, params);
|
||||||
assert!(matches!(client.state, ClientState::Open { token: TOKEN }));
|
assert!(matches!(client.phase, ClientPhase::Open { token: TOKEN }));
|
||||||
server.join();
|
server.join();
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -8,13 +8,27 @@ pub struct NegotiatedParams {
|
|||||||
}
|
}
|
||||||
|
|
||||||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||||||
pub(crate) enum ClientState {
|
pub(crate) enum ClientPhase {
|
||||||
Connected,
|
Connected,
|
||||||
Open { token: u64 },
|
Open { token: u64 },
|
||||||
NoTestCompleted,
|
NoTestCompleted,
|
||||||
Closed,
|
Closed,
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[derive(Debug)]
|
||||||
|
pub(crate) struct ActiveSession {
|
||||||
|
pub next_wire_seq: u32,
|
||||||
|
pub next_logical_seq: u64,
|
||||||
|
pub highest_received_seq: Option<u32>,
|
||||||
|
pub packets_sent: u64,
|
||||||
|
pub start_mono: Instant,
|
||||||
|
pub end_mono: Option<Instant>,
|
||||||
|
pub next_send_at: Instant,
|
||||||
|
pub pending: PendingMap,
|
||||||
|
pub completed: CompletedSet,
|
||||||
|
pub sending_done: bool,
|
||||||
|
}
|
||||||
|
|
||||||
pub(crate) fn validate_negotiated_params(
|
pub(crate) fn validate_negotiated_params(
|
||||||
requested: &Params,
|
requested: &Params,
|
||||||
returned: &Params,
|
returned: &Params,
|
||||||
|
|||||||
Reference in new issue
Block a user