Skip to content

Track individual sessions (local and remote id) for each operation #125

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
gaykitty wants to merge 1 commit into
base: main
Choose a base branch
from
Open

Track individual sessions (local and remote id) for each operation #125

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
101 changes: 56 additions & 45 deletions adb_client/src/device/adb_message_device.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,18 +11,18 @@ use super::{ADBTransportMessage, MessageCommand, models::MessageSubcommand};
#[derive(Debug)]
pub struct ADBMessageDevice<T: ADBMessageTransport> {
transport: T,
local_id: Option<u32>,
remote_id: Option<u32>,
}

#[derive(Debug, Clone, Copy)]
pub struct ADBSession {
pub local_id: u32,
pub remote_id: u32,
}

impl<T: ADBMessageTransport> ADBMessageDevice<T> {
/// Instantiate a new [`ADBMessageTransport`]
pub fn new(transport: T) -> Self {
Self {
transport,
local_id: None,
remote_id: None,
}
Self { transport }
}

pub(crate) fn get_transport(&mut self) -> &T {
Expand All @@ -34,12 +34,15 @@ impl<T: ADBMessageTransport> ADBMessageDevice<T> {
}

/// Receive a message and acknowledge it by replying with an `OKAY` command
pub(crate) fn recv_and_reply_okay(&mut self) -> Result<ADBTransportMessage> {
pub(crate) fn recv_and_reply_okay(
&mut self,
session: ADBSession,
) -> Result<ADBTransportMessage> {
let message = self.transport.read_message()?;
self.transport.write_message(ADBTransportMessage::new(
MessageCommand::Okay,
self.get_local_id()?,
self.get_remote_id()?,
session.local_id,
session.remote_id,
&[],
))?;
Ok(message)
Expand All @@ -60,11 +63,12 @@ impl<T: ADBMessageTransport> ADBMessageDevice<T> {

pub(crate) fn recv_file<W: std::io::Write>(
&mut self,
session: ADBSession,
mut output: W,
) -> std::result::Result<(), RustADBError> {
let mut len: Option<u64> = None;
loop {
let payload = self.recv_and_reply_okay()?.into_payload();
let payload = self.recv_and_reply_okay(session)?.into_payload();
let mut rdr = Cursor::new(&payload);
while rdr.position() != payload.len() as u64 {
match len.take() {
Expand Down Expand Up @@ -97,8 +101,7 @@ impl<T: ADBMessageTransport> ADBMessageDevice<T> {

pub(crate) fn push_file<R: std::io::Read>(
&mut self,
local_id: u32,
remote_id: u32,
session: ADBSession,
mut reader: R,
) -> std::result::Result<(), RustADBError> {
let mut buffer = [0; BUFFER_SIZE];
Expand All @@ -111,8 +114,8 @@ impl<T: ADBMessageTransport> ADBMessageDevice<T> {

let message = ADBTransportMessage::new(
MessageCommand::Write,
local_id,
remote_id,
session.local_id,
session.remote_id,
&serialized_message,
);

Expand All @@ -131,8 +134,8 @@ impl<T: ADBMessageTransport> ADBMessageDevice<T> {

let message = ADBTransportMessage::new(
MessageCommand::Write,
local_id,
remote_id,
session.local_id,
session.remote_id,
&serialized_message,
);

Expand All @@ -159,8 +162,8 @@ impl<T: ADBMessageTransport> ADBMessageDevice<T> {

let message = ADBTransportMessage::new(
MessageCommand::Write,
local_id,
remote_id,
session.local_id,
session.remote_id,
&serialized_message,
);

Expand All @@ -173,24 +176,27 @@ impl<T: ADBMessageTransport> ADBMessageDevice<T> {
}
}

pub(crate) fn begin_synchronization(&mut self) -> Result<()> {
self.open_session(b"sync:\0")?;
Ok(())
pub(crate) fn begin_synchronization(&mut self) -> Result<ADBSession> {
self.open_session(b"sync:\0")
}

pub(crate) fn stat_with_explicit_ids(&mut self, remote_path: &str) -> Result<AdbStatResponse> {
pub(crate) fn stat_with_explicit_ids(
&mut self,
session: ADBSession,
remote_path: &str,
) -> Result<AdbStatResponse> {
let stat_buffer = MessageSubcommand::Stat.with_arg(remote_path.len() as u32);
let message = ADBTransportMessage::new(
MessageCommand::Write,
self.get_local_id()?,
self.get_remote_id()?,
session.local_id,
session.remote_id,
&bincode::serialize(&stat_buffer).map_err(|_e| RustADBError::ConversionError)?,
);
self.send_and_expect_okay(message)?;
self.send_and_expect_okay(ADBTransportMessage::new(
MessageCommand::Write,
self.get_local_id()?,
self.get_remote_id()?,
session.local_id,
session.remote_id,
remote_path.as_bytes(),
))?;
let response = self.transport.read_message()?;
Expand All @@ -200,46 +206,51 @@ impl<T: ADBMessageTransport> ADBMessageDevice<T> {
.map_err(|_e| RustADBError::ConversionError)
}

pub(crate) fn end_transaction(&mut self) -> Result<()> {
pub(crate) fn end_transaction(&mut self, session: ADBSession) -> Result<()> {
let quit_buffer = MessageSubcommand::Quit.with_arg(0u32);
self.send_and_expect_okay(ADBTransportMessage::new(
MessageCommand::Write,
self.get_local_id()?,
self.get_remote_id()?,
session.local_id,
session.remote_id,
&bincode::serialize(&quit_buffer).map_err(|_e| RustADBError::ConversionError)?,
))?;
let _discard_close = self.transport.read_message()?;
Ok(())
}

pub(crate) fn open_session(&mut self, data: &[u8]) -> Result<ADBTransportMessage> {
pub(crate) fn open_session(&mut self, data: &[u8]) -> Result<ADBSession> {
let mut rng = rand::rng();
let local_id: u32 = rng.random();

let message = ADBTransportMessage::new(
MessageCommand::Open,
rng.random(), // Our 'local-id'
local_id, // Our 'local-id'
0,
data,
);
self.get_transport_mut().write_message(message)?;

let response = self.get_transport_mut().read_message()?;

self.local_id = Some(response.header().arg1());
self.remote_id = Some(response.header().arg0());
if response.header().command() != MessageCommand::Okay {
return Err(RustADBError::ADBRequestFailed(format!(
"Open session failed: got {} in respone instead of OKAY",
response.header().command()
)));
}

Ok(response)
}
if response.header().arg1() != local_id {
return Err(RustADBError::ADBRequestFailed(format!(
"Open session failed: respones used {} for our local_id instead of {local_id}",
response.header().arg1()
)));
}

pub(crate) fn get_local_id(&self) -> Result<u32> {
self.local_id.ok_or(RustADBError::ADBRequestFailed(
"connection not opened, no local_id".into(),
))
}
let session = ADBSession {
local_id,
remote_id: response.header().arg0(),
};

pub(crate) fn get_remote_id(&self) -> Result<u32> {
self.remote_id.ok_or(RustADBError::ADBRequestFailed(
"connection not opened, no remote_id".into(),
))
Ok(session)
}
}
8 changes: 4 additions & 4 deletions adb_client/src/device/commands/framebuffer.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,9 +11,9 @@ use crate::{

impl<T: ADBMessageTransport> ADBMessageDevice<T> {
pub(crate) fn framebuffer_inner(&mut self) -> Result<ImageBuffer<Rgba<u8>, Vec<u8>>> {
self.open_session(b"framebuffer:\0")?;
let session = self.open_session(b"framebuffer:\0")?;

let response = self.recv_and_reply_okay()?;
let response = self.recv_and_reply_okay(session)?;

let mut payload_cursor = Cursor::new(response.payload());

Expand All @@ -36,7 +36,7 @@ impl<T: ADBMessageTransport> ADBMessageDevice<T> {
break;
}

let response = self.recv_and_reply_okay()?;
let response = self.recv_and_reply_okay(session)?;

framebuffer_data.extend_from_slice(&response.into_payload());

Expand Down Expand Up @@ -69,7 +69,7 @@ impl<T: ADBMessageTransport> ADBMessageDevice<T> {
break;
}

let response = self.recv_and_reply_okay()?;
let response = self.recv_and_reply_okay(session)?;

framebuffer_data.extend_from_slice(&response.into_payload());

Expand Down
5 changes: 3 additions & 2 deletions adb_client/src/device/commands/install.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,11 +14,12 @@ impl<T: ADBMessageTransport> ADBMessageDevice<T> {

let file_size = apk_file.metadata()?.len();

self.open_session(format!("exec:cmd package 'install' -S {}\0", file_size).as_bytes())?;
let session =
self.open_session(format!("exec:cmd package 'install' -S {}\0", file_size).as_bytes())?;

let transport = self.get_transport().clone();

let mut writer = MessageWriter::new(transport, self.get_local_id()?, self.get_remote_id()?);
let mut writer = MessageWriter::new(transport, session.local_id, session.remote_id);

std::io::copy(&mut apk_file, &mut writer)?;

Expand Down
26 changes: 14 additions & 12 deletions adb_client/src/device/commands/pull.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,22 +10,24 @@ use crate::{

impl<T: ADBMessageTransport> ADBMessageDevice<T> {
pub(crate) fn pull<A: AsRef<str>, W: Write>(&mut self, source: A, output: W) -> Result<()> {
self.begin_synchronization()?;
let session = self.begin_synchronization()?;
let source = source.as_ref();

let adb_stat_response = self.stat_with_explicit_ids(source)?;
let adb_stat_response = self.stat_with_explicit_ids(session, source)?;

if adb_stat_response.file_perm == 0 {
return Err(RustADBError::UnknownResponseType(
"mode is 0: source file does not exist".to_string(),
));
}

let local_id = self.get_local_id()?;
let remote_id = self.get_remote_id()?;

self.get_transport_mut().write_message_with_timeout(
ADBTransportMessage::new(MessageCommand::Okay, local_id, remote_id, &[]),
ADBTransportMessage::new(
MessageCommand::Okay,
session.local_id,
session.remote_id,
&[],
),
std::time::Duration::from_secs(4),
)?;

Expand All @@ -34,19 +36,19 @@ impl<T: ADBMessageTransport> ADBMessageDevice<T> {
bincode::serialize(&recv_buffer).map_err(|_e| RustADBError::ConversionError)?;
self.send_and_expect_okay(ADBTransportMessage::new(
MessageCommand::Write,
self.get_local_id()?,
self.get_remote_id()?,
session.local_id,
session.remote_id,
&recv_buffer,
))?;
self.send_and_expect_okay(ADBTransportMessage::new(
MessageCommand::Write,
self.get_local_id()?,
self.get_remote_id()?,
session.local_id,
session.remote_id,
source.as_bytes(),
))?;

self.recv_file(output)?;
self.end_transaction()?;
self.recv_file(session, output)?;
self.end_transaction(session)?;
Ok(())
}
}
11 changes: 5 additions & 6 deletions adb_client/src/device/commands/push.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,7 @@ use crate::{

impl<T: ADBMessageTransport> ADBMessageDevice<T> {
pub(crate) fn push<R: Read, A: AsRef<str>>(&mut self, stream: R, path: A) -> Result<()> {
self.begin_synchronization()?;
let session = self.begin_synchronization()?;

let path_header = format!("{},0777", path.as_ref());

Expand All @@ -21,14 +21,13 @@ impl<T: ADBMessageTransport> ADBMessageDevice<T> {

self.send_and_expect_okay(ADBTransportMessage::new(
MessageCommand::Write,
self.get_local_id()?,
self.get_remote_id()?,
session.local_id,
session.remote_id,
&send_buffer,
))?;

self.push_file(self.get_local_id()?, self.get_remote_id()?, stream)?;

self.end_transaction()?;
self.push_file(session, stream)?;
self.end_transaction(session)?;

Ok(())
}
Expand Down
25 changes: 10 additions & 15 deletions adb_client/src/device/commands/shell.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,14 +10,7 @@ use crate::{
impl<T: ADBMessageTransport> ADBMessageDevice<T> {
/// Runs 'command' in a shell on the device, and write its output and error streams into output.
pub(crate) fn shell_command(&mut self, command: &[&str], output: &mut dyn Write) -> Result<()> {
let response = self.open_session(format!("shell:{}\0", command.join(" "),).as_bytes())?;

if response.header().command() != MessageCommand::Okay {
return Err(RustADBError::ADBRequestFailed(format!(
"wrong command {}",
response.header().command()
)));
}
let session = self.open_session(format!("shell:{}\0", command.join(" "),).as_bytes())?;
Copy link
Owner

@cocool97 cocool97 Jul 6, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

session variable is not used afterwards


loop {
let response = self.get_transport_mut().read_message()?;
Expand All @@ -38,21 +31,22 @@ impl<T: ADBMessageTransport> ADBMessageDevice<T> {
mut reader: &mut dyn Read,
mut writer: Box<(dyn Write + Send)>,
) -> Result<()> {
self.open_session(b"shell:\0")?;
let session = self.open_session(b"shell:\0")?;

let mut transport = self.get_transport().clone();

let local_id = self.get_local_id()?;
let remote_id = self.get_remote_id()?;

// Reading thread, reads response from adbd
std::thread::spawn(move || -> Result<()> {
loop {
let message = transport.read_message()?;

// Acknowledge for more data
let response =
ADBTransportMessage::new(MessageCommand::Okay, local_id, remote_id, &[]);
let response = ADBTransportMessage::new(
MessageCommand::Okay,
session.local_id,
session.remote_id,
&[],
);
transport.write_message(response)?;

match message.header().command() {
Expand All @@ -67,7 +61,8 @@ impl<T: ADBMessageTransport> ADBMessageDevice<T> {
});

let transport = self.get_transport().clone();
let mut shell_writer = ShellMessageWriter::new(transport, local_id, remote_id);
let mut shell_writer =
ShellMessageWriter::new(transport, session.local_id, session.remote_id);

// Read from given reader (that could be stdin e.g), and write content to device adbd
if let Err(e) = std::io::copy(&mut reader, &mut shell_writer) {
Expand Down
Loading