use std::{
collections::VecDeque,
time::{Duration, Instant},
};
use crate::{
command::EmapiCommand,
connection::EmapiConnection,
constants::{
CHILD_REPORT_BLUETOOTH_CONNECTION, CHILD_REPORT_FLOW_CONTROL, CHILD_REPORT_PRINTER_STATUS,
CHILD_REPORT_PRINT_RESULT, CHILD_REPORT_UPGRADE_STATUS, CHILD_WIFI_CONFIG_STATUS,
PARENT_REPORT, PARENT_WIFI, TYPE_PASSTHROUGH_REQUEST, TYPE_PASSTHROUGH_RESPONSE, TYPE_REQUEST,
TYPE_RESPONSE,
},
error::{EmapiError, Result},
frame_codec::FrameCodec,
frame_parser::FrameParser,
models::EmapiReport,
tlv::Tlv,
};
pub struct EmapiSession<C> {
connection: C,
timeout: Duration,
max_retries: usize,
parser: FrameParser,
pending_commands: VecDeque<EmapiCommand>,
reports: Vec<EmapiReport>,
flow_busy: bool,
}
impl<C> EmapiSession<C>
where
C: EmapiConnection,
{
pub fn new(connection: C) -> Self {
Self {
connection,
timeout: Duration::from_secs(2),
max_retries: 3,
parser: FrameParser::new(),
pending_commands: VecDeque::new(),
reports: Vec::new(),
flow_busy: false,
}
}
pub fn with_timeout(mut self, timeout: Duration) -> Self {
self.timeout = timeout;
self
}
pub fn set_timeout(&mut self, timeout: Duration) {
self.timeout = timeout;
}
pub fn with_max_retries(mut self, max_retries: usize) -> Result<Self> {
self.max_retries = max_retries;
Ok(self)
}
pub fn set_max_retries(&mut self, max_retries: usize) {
self.max_retries = max_retries;
}
pub fn reports(&self) -> &[EmapiReport] {
&self.reports
}
pub fn read_report(&mut self) -> Result<EmapiReport> {
let start = Instant::now();
let mut pending = VecDeque::new();
while let Some(command) = self.pending_commands.pop_front() {
let report_index = self.reports.len();
if self.handle_report(command.clone())? {
pending.extend(self.pending_commands.drain(..));
self.pending_commands = pending;
return Ok(self.reports[report_index].clone());
}
pending.push_back(command);
}
self.pending_commands = pending;
loop {
let command = self.read_next_command_from_connection(start)?;
let report_index = self.reports.len();
if self.handle_report(command.clone())? {
return Ok(self.reports[report_index].clone());
}
self.pending_commands.push_back(command);
}
}
pub fn into_inner(self) -> C {
self.connection
}
pub fn send_and_wait<F>(&mut self, command: EmapiCommand, expected: F) -> Result<EmapiCommand>
where
F: Fn(&EmapiCommand) -> bool,
{
let mut last_error = None;
for _ in 0..=self.max_retries {
self.wait_until_flow_idle()?;
self.connection.write(&FrameCodec::encode(&command)?)?;
match self.read_until_expected(&expected) {
Ok(response) => return Ok(response),
Err(EmapiError::Timeout { message }) => {
last_error = Some(message);
}
Err(EmapiError::Protocol { message }) => {
last_error = Some(message);
}
Err(error) => return Err(error),
}
}
Err(EmapiError::Protocol {
message: format!(
"no expected ACK after {} attempt(s): {}",
self.max_retries + 1,
last_error.unwrap_or_else(|| "unknown error".to_string())
),
})
}
fn read_until_expected<F>(&mut self, expected: F) -> Result<EmapiCommand>
where
F: Fn(&EmapiCommand) -> bool,
{
let start = Instant::now();
let mut last_unexpected = None;
loop {
let response = match self.read_next_command(start) {
Ok(response) => response,
Err(EmapiError::Timeout { .. }) if last_unexpected.is_some() => {
return Err(EmapiError::Protocol {
message: last_unexpected.unwrap(),
});
}
Err(error) => return Err(error),
};
if response.is_error() {
let error_kind = if response.is_passthrough_error() {
"passthrough"
} else {
"protocol"
};
return Err(EmapiError::Protocol {
message: format!("received {error_kind} error response: {response:?}"),
});
}
if self.handle_report(response.clone())? {
continue;
}
if expected(&response) {
return Ok(response);
}
last_unexpected = Some(format!("unexpected response: {response:?}"));
}
}
fn read_next_command(&mut self, start: Instant) -> Result<EmapiCommand> {
if let Some(command) = self.pending_commands.pop_front() {
return Ok(command);
}
self.read_next_command_from_connection(start)
}
fn read_next_command_from_connection(&mut self, start: Instant) -> Result<EmapiCommand> {
loop {
let remaining = self.remaining_timeout(start)?;
let data = self.connection.read(remaining)?;
let mut commands = self.parser.add(&data)?;
if !commands.is_empty() {
let command = commands.remove(0);
self.pending_commands.extend(commands);
return Ok(command);
}
}
}
fn wait_until_flow_idle(&mut self) -> Result<()> {
if !self.flow_busy {
return Ok(());
}
let start = Instant::now();
while self.flow_busy {
let command = self.read_next_command(start)?;
if !self.handle_report(command.clone())? {
self.pending_commands.push_front(command);
return Ok(());
}
}
Ok(())
}
fn remaining_timeout(&self, start: Instant) -> Result<Duration> {
self
.timeout
.checked_sub(start.elapsed())
.filter(|remaining| !remaining.is_zero())
.ok_or_else(|| EmapiError::Timeout {
message: format!(
"timeout waiting for expected response after {:?}",
self.timeout
),
})
}
fn handle_report(&mut self, command: EmapiCommand) -> Result<bool> {
if is_normal_report(&command) {
self.write_ack(TYPE_RESPONSE, command.parent, command.child)?;
let report = parse_normal_report(command)?;
self.emit_report(report);
return Ok(true);
}
if is_wifi_passthrough_report(&command) {
self.write_ack(TYPE_PASSTHROUGH_RESPONSE, command.parent, command.child)?;
let report = parse_wifi_passthrough_report(command)?;
self.emit_report(report);
return Ok(true);
}
Ok(false)
}
fn write_ack(&mut self, command_type: u8, parent: u8, child: u8) -> Result<()> {
self
.connection
.write(&FrameCodec::encode(&EmapiCommand::new(
command_type,
parent,
child,
[],
))?)
}
fn emit_report(&mut self, report: EmapiReport) {
if let EmapiReport::FlowControl { state, .. } = &report {
self.flow_busy = *state == 1;
}
self.reports.push(report);
}
}
fn is_normal_report(command: &EmapiCommand) -> bool {
command.command_type == TYPE_REQUEST && command.parent == PARENT_REPORT
}
fn is_wifi_passthrough_report(command: &EmapiCommand) -> bool {
command.command_type == TYPE_PASSTHROUGH_REQUEST
&& command.parent == PARENT_WIFI
&& command.child == CHILD_WIFI_CONFIG_STATUS
}
fn parse_normal_report(command: EmapiCommand) -> Result<EmapiReport> {
match command.child {
CHILD_REPORT_PRINT_RESULT => {
let result = payload_byte(&command.payload)?;
Ok(EmapiReport::PrintResult { command, result })
}
CHILD_REPORT_PRINTER_STATUS => {
let tlv = Tlv::decode(&command.payload)?;
Ok(EmapiReport::PrinterStatus {
paper_status: tlv.uint8(0x01)?,
cover_status: tlv.uint8(0x02)?,
battery_state: tlv.uint8(0x03)?,
overheat: tlv.uint8(0x04)?,
nfc_paper_recognition: tlv.uint8(0x05)?,
command,
})
}
CHILD_REPORT_FLOW_CONTROL => {
let state = payload_byte(&command.payload)?;
Ok(EmapiReport::FlowControl { command, state })
}
CHILD_REPORT_UPGRADE_STATUS => {
let status = payload_byte(&command.payload)?;
Ok(EmapiReport::UpgradeStatus { command, status })
}
CHILD_REPORT_BLUETOOTH_CONNECTION => {
let state = payload_byte(&command.payload)?;
Ok(EmapiReport::BluetoothConnection { command, state })
}
_ => Ok(EmapiReport::Unknown { command }),
}
}
fn parse_wifi_passthrough_report(command: EmapiCommand) -> Result<EmapiReport> {
let tlv = Tlv::decode(&command.payload)?;
Ok(EmapiReport::WifiConfigStatus {
ssid: tlv.string(0x01)?,
state: tlv.uint8(0x02)?,
command,
})
}
fn payload_byte(payload: &[u8]) -> Result<u8> {
payload
.first()
.copied()
.ok_or_else(|| EmapiError::Protocol {
message: "missing report payload byte".to_string(),
})
}