ReadRequest: Basic stuff works across multi-hop reads
This commit is contained in:
parent
487124c9dc
commit
dd47aa9a69
6 changed files with 78 additions and 48 deletions
|
@ -37,7 +37,7 @@ use crate::{
|
|||
InteractionConsumer, Transaction,
|
||||
},
|
||||
secure_channel::pake::PaseMgr,
|
||||
tlv::{TLVArray, TLVWriter, TagType, ToTLV},
|
||||
tlv::{self, FromTLV, TLVArray, TLVWriter, TagType, ToTLV},
|
||||
transport::{
|
||||
proto_demux::ResponseRequired,
|
||||
session::{Session, SessionMode},
|
||||
|
@ -259,13 +259,17 @@ impl InteractionConsumer for DataModel {
|
|||
|
||||
fn consume_read_attr(
|
||||
&self,
|
||||
req: &ReadReq,
|
||||
rx_buf: &[u8],
|
||||
trans: &mut Transaction,
|
||||
tw: &mut TLVWriter,
|
||||
) -> Result<(), Error> {
|
||||
let resume = self.handle_read_attr_array(req, trans, tw)?;
|
||||
if let Some(resume) = resume {
|
||||
let mut resume_from = None;
|
||||
let root = tlv::get_root_node(rx_buf)?;
|
||||
let req = ReadReq::from_tlv(&root)?;
|
||||
self.handle_read_attr_array(&req, trans, tw, &mut resume_from)?;
|
||||
if resume_from.is_some() {
|
||||
// This is a multi-hop read transaction, remember this read request
|
||||
let resume = read::ResumeReadReq::new(rx_buf, &resume_from)?;
|
||||
if !trans.exch.is_data_none() {
|
||||
error!("Exchange data already set, and multi-hop read");
|
||||
return Err(Error::InvalidState);
|
||||
|
@ -316,15 +320,16 @@ impl InteractionConsumer for DataModel {
|
|||
trans: &mut Transaction,
|
||||
tw: &mut TLVWriter,
|
||||
) -> Result<(OpCode, ResponseRequired), Error> {
|
||||
if let Some(resume) = trans.exch.take_data_boxed::<ResumeReq>() {
|
||||
match *resume {
|
||||
ResumeReq::Read(_) => Ok((OpCode::Reserved, ResponseRequired::No)),
|
||||
if let Some(mut resume) = trans.exch.take_data_boxed::<ResumeReq>() {
|
||||
let result = match *resume {
|
||||
ResumeReq::Read(ref mut read) => self.handle_resume_read(read, trans, tw)?,
|
||||
|
||||
ResumeReq::Subscribe(mut ctx) => {
|
||||
let result = self.handle_subscription_confirm(trans, tw, &mut ctx)?;
|
||||
self.handle_subscription_confirm(trans, tw, &mut ctx)?
|
||||
}
|
||||
};
|
||||
trans.exch.set_data_boxed(resume);
|
||||
Ok(result)
|
||||
}
|
||||
}
|
||||
} else {
|
||||
// Nothing to do for now
|
||||
trans.complete();
|
||||
|
|
|
@ -16,8 +16,10 @@
|
|||
*/
|
||||
|
||||
use crate::data_model::{core::DataModel, objects::*};
|
||||
use crate::interaction_model::core::OpCode;
|
||||
use crate::tlv::FromTLV;
|
||||
use crate::transport::packet::Packet;
|
||||
use crate::utils::writebuf::WriteBuf;
|
||||
use crate::transport::proto_demux::ResponseRequired;
|
||||
use crate::{
|
||||
acl::{AccessReq, Accessor},
|
||||
error::*,
|
||||
|
@ -25,12 +27,12 @@ use crate::{
|
|||
core::IMStatusCode,
|
||||
messages::{
|
||||
ib::{self, DataVersionFilter},
|
||||
msg::{self, ReadReq, ReportDataTag::MoreChunkedMsgs},
|
||||
msg::{self, ReadReq, ReportDataTag::MoreChunkedMsgs, ReportDataTag::SupressResponse},
|
||||
GenericPath,
|
||||
},
|
||||
Transaction,
|
||||
},
|
||||
tlv::{TLVArray, TLVWriter, TagType, ToTLV},
|
||||
tlv::{self, TLVArray, TLVWriter, TagType, ToTLV},
|
||||
};
|
||||
|
||||
/// Encoder for generating a response to a read request
|
||||
|
@ -112,7 +114,21 @@ pub struct ResumeReadReq {
|
|||
/// will start encoding from this attribute onwards.
|
||||
/// Note that given wildcard reads, one PendingPath in the member above can generated
|
||||
/// multiple encode paths. Hence this has to be maintained separately.
|
||||
resume_encode: Option<GenericPath>,
|
||||
resume_from: Option<GenericPath>,
|
||||
}
|
||||
impl ResumeReadReq {
|
||||
pub fn new(rx_buf: &[u8], resume_from: &Option<GenericPath>) -> Result<Self, Error> {
|
||||
let mut packet = Packet::new_rx()?;
|
||||
let dst = packet.as_borrow_slice();
|
||||
|
||||
let src_len = rx_buf.len();
|
||||
dst[..src_len].copy_from_slice(rx_buf);
|
||||
packet.get_parsebuf()?.set_len(src_len);
|
||||
Ok(ResumeReadReq {
|
||||
pending_req: Some(packet),
|
||||
resume_from: *resume_from,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
impl DataModel {
|
||||
|
@ -195,9 +211,8 @@ impl DataModel {
|
|||
read_req: &ReadReq,
|
||||
trans: &mut Transaction,
|
||||
tw: &mut TLVWriter,
|
||||
) -> Result<Option<ResumeReadReq>, Error> {
|
||||
let mut resume_read_req: ResumeReadReq = Default::default();
|
||||
|
||||
resume_from: &mut Option<GenericPath>,
|
||||
) -> Result<(), Error> {
|
||||
let mut attr_encoder = AttrReadEncoder::new(tw);
|
||||
if let Some(filters) = &read_req.dataver_filters {
|
||||
attr_encoder.set_data_ver_filters(filters);
|
||||
|
@ -221,7 +236,7 @@ impl DataModel {
|
|||
&accessor,
|
||||
&mut attr_encoder,
|
||||
&mut attr_details,
|
||||
&mut resume_read_req.resume_encode,
|
||||
resume_from,
|
||||
);
|
||||
}
|
||||
tw.end_container()?;
|
||||
|
@ -229,25 +244,35 @@ impl DataModel {
|
|||
// If there was an error, indicate chunking. The resume_read_req would have been
|
||||
// already populated from in the loop above.
|
||||
tw.bool(TagType::Context(MoreChunkedMsgs as u8), true)?;
|
||||
// Retain the entire request, because we need the data-filters, and subsequent attr-reads, if any
|
||||
// when we resume this read in the next hop
|
||||
resume_read_req.pending_req = Some(copy_read_req_to_packet(read_req)?);
|
||||
return Ok(Some(resume_read_req));
|
||||
return Ok(());
|
||||
}
|
||||
}
|
||||
Ok(None)
|
||||
// A None resume_from indicates no chunking
|
||||
*resume_from = None;
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
fn copy_read_req_to_packet(read_req: &ReadReq) -> Result<Packet<'static>, Error> {
|
||||
let mut packet = Packet::new_rx()?;
|
||||
let backup = packet.as_borrow_slice();
|
||||
let backup_len = backup.len();
|
||||
let mut wb = WriteBuf::new(backup, backup_len);
|
||||
let mut tw = TLVWriter::new(&mut wb);
|
||||
// TODO: This is unnecessarily wasteful, could directly copy &[u8] if accessible
|
||||
read_req.to_tlv(&mut tw, TagType::Anonymous)?;
|
||||
let data_len = wb.as_borrow_slice().len();
|
||||
packet.get_parsebuf()?.set_len(data_len);
|
||||
Ok(packet)
|
||||
pub fn handle_resume_read(
|
||||
&self,
|
||||
resume_read_req: &mut ResumeReadReq,
|
||||
trans: &mut Transaction,
|
||||
tw: &mut TLVWriter,
|
||||
) -> Result<(OpCode, ResponseRequired), Error> {
|
||||
if let Some(packet) = resume_read_req.pending_req.as_mut() {
|
||||
let rx_buf = packet.get_parsebuf()?.as_borrow_slice();
|
||||
let root = tlv::get_root_node(rx_buf)?;
|
||||
let req = ReadReq::from_tlv(&root)?;
|
||||
|
||||
tw.start_struct(TagType::Anonymous)?;
|
||||
self.handle_read_attr_array(&req, trans, tw, &mut resume_read_req.resume_from)?;
|
||||
|
||||
if resume_read_req.resume_from.is_none() {
|
||||
tw.bool(TagType::Context(SupressResponse as u8), true)?;
|
||||
// Mark transaction complete, if not chunked
|
||||
trans.complete();
|
||||
}
|
||||
tw.end_container()?;
|
||||
}
|
||||
Ok((OpCode::ReportData, ResponseRequired::Yes))
|
||||
}
|
||||
}
|
||||
|
|
|
@ -50,7 +50,8 @@ impl DataModel {
|
|||
TagType::Context(msg::ReportDataTag::SubscriptionId as u8),
|
||||
ctx.id,
|
||||
)?;
|
||||
self.handle_read_attr_array(&read_req, trans, tw)?;
|
||||
let mut resume_from = None;
|
||||
self.handle_read_attr_array(&read_req, trans, tw, &mut resume_from)?;
|
||||
tw.end_container()?;
|
||||
|
||||
Ok(ctx)
|
||||
|
|
|
@ -23,7 +23,7 @@ use crate::{
|
|||
|
||||
use self::{
|
||||
core::OpCode,
|
||||
messages::msg::{InvReq, ReadReq, StatusResp, SubscribeReq, WriteReq},
|
||||
messages::msg::{InvReq, StatusResp, SubscribeReq, WriteReq},
|
||||
};
|
||||
|
||||
#[derive(PartialEq)]
|
||||
|
@ -47,7 +47,9 @@ pub trait InteractionConsumer {
|
|||
|
||||
fn consume_read_attr(
|
||||
&self,
|
||||
req: &ReadReq,
|
||||
// TODO: This handling is different from the other APIs here, identify
|
||||
// consistent options for this trait
|
||||
req: &[u8],
|
||||
trans: &mut Transaction,
|
||||
tw: &mut TLVWriter,
|
||||
) -> Result<(), Error>;
|
||||
|
|
|
@ -18,13 +18,13 @@
|
|||
use crate::{
|
||||
error::Error,
|
||||
interaction_model::core::OpCode,
|
||||
tlv::{get_root_node_struct, FromTLV, TLVWriter, TagType},
|
||||
tlv::{TLVWriter, TagType},
|
||||
transport::{packet::Packet, proto_demux::ResponseRequired},
|
||||
utils::writebuf::WriteBuf,
|
||||
wb_shrink, wb_unshrink,
|
||||
};
|
||||
|
||||
use super::{messages::msg::ReadReq, InteractionModel, Transaction};
|
||||
use super::{InteractionModel, Transaction};
|
||||
|
||||
impl InteractionModel {
|
||||
pub fn handle_read_req(
|
||||
|
@ -40,13 +40,11 @@ impl InteractionModel {
|
|||
let proto_tx_wb = proto_tx.get_writebuf()?;
|
||||
let mut child_wb = wb_shrink!(proto_tx_wb, RESERVE_SIZE);
|
||||
let mut tw = TLVWriter::new(&mut child_wb);
|
||||
let root = get_root_node_struct(rx_buf)?;
|
||||
let read_req = ReadReq::from_tlv(&root)?;
|
||||
|
||||
tw.start_struct(TagType::Anonymous)?;
|
||||
self.consumer.consume_read_attr(&read_req, trans, &mut tw)?;
|
||||
self.consumer.consume_read_attr(rx_buf, trans, &mut tw)?;
|
||||
|
||||
// Now that we have everything, start using the proto_tx_wb, by unshrinking it
|
||||
//Now that we have everything, start using the proto_tx_wb, by unshrinking it
|
||||
wb_unshrink!(proto_tx_wb, child_wb);
|
||||
let mut tw = TLVWriter::new(proto_tx_wb);
|
||||
tw.end_container()?;
|
||||
|
|
|
@ -19,7 +19,6 @@ use boxslab::Slab;
|
|||
use matter::error::Error;
|
||||
use matter::interaction_model::core::OpCode;
|
||||
use matter::interaction_model::messages::msg::InvReq;
|
||||
use matter::interaction_model::messages::msg::ReadReq;
|
||||
use matter::interaction_model::messages::msg::WriteReq;
|
||||
use matter::interaction_model::InteractionConsumer;
|
||||
use matter::interaction_model::InteractionModel;
|
||||
|
@ -94,7 +93,7 @@ impl InteractionConsumer for DataModel {
|
|||
|
||||
fn consume_read_attr(
|
||||
&self,
|
||||
_req: &ReadReq,
|
||||
_req: &[u8],
|
||||
_trans: &mut Transaction,
|
||||
_tlvwriter: &mut TLVWriter,
|
||||
) -> Result<(), Error> {
|
||||
|
|
Loading…
Add table
Reference in a new issue