Skip to content

Commit

Permalink
Add mNumMessagesToAllowBeforeDropping
Browse files Browse the repository at this point in the history 
Add mNumMessagesToAllowBeforeDropping in NetworkTestHelper so that it
can drop the number of message after configurable number of allowed
message.
  • Loading branch information
@yunhanw-google
yunhanw-google committed Jul 28, 2022
1 parent 6cd4676 commit 5bb344c
Show file tree
Hide file tree
Showing 23 changed files with 770 additions and 241 deletions.
78 changes: 35 additions & 43 deletions src/app/CommandHandler.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -39,6 +39,7 @@

namespace chip {
namespace app {
using Status = Protocols::InteractionModel::Status;

CommandHandler::CommandHandler(Callback * apCallback) : mExchangeCtx(*this), mpCallback(apCallback), mSuppressResponse(false) {}

Expand All @@ -65,11 +66,13 @@ CHIP_ERROR CommandHandler::AllocateBuffer()
return CHIP_NO_ERROR;
}

CHIP_ERROR CommandHandler::OnInvokeCommandRequest(Messaging::ExchangeContext * ec, const PayloadHeader & payloadHeader,
System::PacketBufferHandle && payload, bool isTimedInvoke)
Status CommandHandler::OnInvokeCommandRequest(Messaging::ExchangeContext * ec, const PayloadHeader & payloadHeader,
System::PacketBufferHandle && payload, bool isTimedInvoke)
{
System::PacketBufferHandle response;
VerifyOrReturnError(mState == State::Idle, CHIP_ERROR_INCORRECT_STATE);

VerifyOrReturnError(ec != nullptr, Status::Failure);
VerifyOrReturnError(mState == State::Idle, Status::Failure);

// NOTE: we already know this is an InvokeCommand Request message because we explicitly registered with the
// Exchange Manager for unsolicited InvokeCommand Requests.
Expand All @@ -81,67 +84,51 @@ CHIP_ERROR CommandHandler::OnInvokeCommandRequest(Messaging::ExchangeContext * e
// response too. Figure out at what point it's our responsibility to
// handler errors vs our caller's.
Handle workHandle(this);
mExchangeCtx->WillSendMessage();
ReturnErrorOnFailure(ProcessInvokeRequest(std::move(payload), isTimedInvoke));

return CHIP_NO_ERROR;
mExchangeCtx->WillSendMessage();
return ProcessInvokeRequest(std::move(payload), isTimedInvoke);
}

CHIP_ERROR CommandHandler::ProcessInvokeRequest(System::PacketBufferHandle && payload, bool isTimedInvoke)
Status CommandHandler::ProcessInvokeRequest(System::PacketBufferHandle && payload, bool isTimedInvoke)
{
CHIP_ERROR err = CHIP_NO_ERROR;
System::PacketBufferTLVReader reader;
TLV::TLVReader invokeRequestsReader;
InvokeRequestMessage::Parser invokeRequestMessage;
InvokeRequests::Parser invokeRequests;
reader.Init(std::move(payload));
ReturnErrorOnFailure(invokeRequestMessage.Init(reader));
VerifyOrReturnError(invokeRequestMessage.Init(reader) == CHIP_NO_ERROR, Status::InvalidAction);
#if CHIP_CONFIG_IM_ENABLE_SCHEMA_CHECK
ReturnErrorOnFailure(invokeRequestMessage.CheckSchemaValidity());
VerifyOrReturnError(invokeRequestMessage.CheckSchemaValidity() == CHIP_NO_ERROR, Status::InvalidAction);
#endif
ReturnErrorOnFailure(invokeRequestMessage.GetSuppressResponse(&mSuppressResponse));
ReturnErrorOnFailure(invokeRequestMessage.GetTimedRequest(&mTimedRequest));
ReturnErrorOnFailure(invokeRequestMessage.GetInvokeRequests(&invokeRequests));

VerifyOrReturnError(mExchangeCtx, CHIP_ERROR_INCORRECT_STATE);

if (mTimedRequest != isTimedInvoke)
{
// The message thinks it should be part of a timed interaction but it's
// not, or vice versa. Spec says to Respond with UNSUPPORTED_ACCESS.
err = StatusResponse::Send(Protocols::InteractionModel::Status::UnsupportedAccess, mExchangeCtx.Get(),
/* aExpectResponse = */ false);
if (err == CHIP_NO_ERROR)
{
mSentStatusResponse = true;
}

return err;
}

VerifyOrReturnError(invokeRequestMessage.GetSuppressResponse(&mSuppressResponse) == CHIP_NO_ERROR, Status::InvalidAction);
VerifyOrReturnError(invokeRequestMessage.GetTimedRequest(&mTimedRequest) == CHIP_NO_ERROR, Status::InvalidAction);
VerifyOrReturnError(invokeRequestMessage.GetInvokeRequests(&invokeRequests) == CHIP_NO_ERROR, Status::InvalidAction);
VerifyOrReturnError(mTimedRequest == isTimedInvoke, Status::UnsupportedAccess);
invokeRequests.GetReader(&invokeRequestsReader);

{
// We don't support handling multiple commands but the protocol is ready to support it in the future, reject all of them and
// IM Engine will send a status response.
size_t commandCount = 0;
TLV::Utilities::Count(invokeRequestsReader, commandCount, false /* recurse */);
VerifyOrReturnError(commandCount == 1, CHIP_ERROR_UNSUPPORTED_CHIP_FEATURE);
VerifyOrReturnError(commandCount == 1, Status::Failure);
}

while (CHIP_NO_ERROR == (err = invokeRequestsReader.Next()))
{
VerifyOrReturnError(TLV::AnonymousTag() == invokeRequestsReader.GetTag(), CHIP_ERROR_INVALID_TLV_TAG);
VerifyOrReturnError(TLV::AnonymousTag() == invokeRequestsReader.GetTag(), Status::InvalidAction);
CommandDataIB::Parser commandData;
ReturnErrorOnFailure(commandData.Init(invokeRequestsReader));
VerifyOrReturnError(commandData.Init(invokeRequestsReader) == CHIP_NO_ERROR, Status::InvalidAction);

if (mExchangeCtx->IsGroupExchangeContext())
{
ReturnErrorOnFailure(ProcessGroupCommandDataIB(commandData));
VerifyOrReturnError(ProcessGroupCommandDataIB(commandData) == CHIP_NO_ERROR, Status::Failure);
}
else
{
ReturnErrorOnFailure(ProcessCommandDataIB(commandData));
VerifyOrReturnError(ProcessCommandDataIB(commandData) == CHIP_NO_ERROR, Status::Failure);
}
}

Expand All @@ -150,8 +137,16 @@ CHIP_ERROR CommandHandler::ProcessInvokeRequest(System::PacketBufferHandle && pa
{
err = CHIP_NO_ERROR;
}
ReturnErrorOnFailure(err);
return invokeRequestMessage.ExitContainer();
VerifyOrReturnError(err == CHIP_NO_ERROR, Status::InvalidAction);
VerifyOrReturnError(invokeRequestMessage.ExitContainer() == CHIP_NO_ERROR, Status::InvalidAction);
return Status::Success;
}

CHIP_ERROR CommandHandler::OnMessageReceived(Messaging::ExchangeContext * apExchangeContext, const PayloadHeader & aPayloadHeader,
System::PacketBufferHandle && aPayload)
{
ChipLogDetail(DataManagement, "CommandHandler: Unexpected message type %d", aPayloadHeader.GetMessageType());
return StatusResponse::Send(Status::InvalidAction, mExchangeCtx.Get(), false /*aExpectResponse*/);
}

void CommandHandler::Close()
Expand Down Expand Up @@ -186,7 +181,7 @@ void CommandHandler::DecrementHoldOff()
return;
}

if (!mExchangeCtx->IsGroupExchangeContext() && !mSentStatusResponse)
if (!mExchangeCtx->IsGroupExchangeContext())
{
CHIP_ERROR err = SendCommandResponse();
if (err != CHIP_NO_ERROR)
Expand Down Expand Up @@ -250,7 +245,6 @@ CHIP_ERROR CommandHandler::ProcessCommandDataIB(CommandDataIB::Parser & aCommand
err = commandPath.GetEndpointId(&concretePath.mEndpointId);
SuccessOrExit(err);

using Protocols::InteractionModel::Status;
{
Status commandExists = mpCallback->CommandExists(concretePath);
if (commandExists != Status::Success)
Expand Down Expand Up @@ -284,7 +278,7 @@ CHIP_ERROR CommandHandler::ProcessCommandDataIB(CommandDataIB::Parser & aCommand
{
// TODO: when wildcard invokes are supported, discard a
// wildcard-expanded path instead of returning a status.
return AddStatus(concretePath, Protocols::InteractionModel::Status::NeedsTimedInteraction);
return AddStatus(concretePath, Status::NeedsTimedInteraction);
}

if (CommandIsFabricScoped(concretePath.mClusterId, concretePath.mCommandId))
Expand Down Expand Up @@ -399,7 +393,7 @@ CHIP_ERROR CommandHandler::ProcessGroupCommandDataIB(CommandDataIB::Parser & aCo

const ConcreteCommandPath concretePath(mapping.endpoint_id, clusterId, commandId);

if (mpCallback->CommandExists(concretePath) != Protocols::InteractionModel::Status::Success)
if (mpCallback->CommandExists(concretePath) != Status::Success)
{
ChipLogDetail(DataManagement, "No command " ChipLogFormatMEI " in Cluster " ChipLogFormatMEI " on Endpoint 0x%x",
ChipLogValueMEI(commandId), ChipLogValueMEI(clusterId), mapping.endpoint_id);
Expand Down Expand Up @@ -452,20 +446,18 @@ CHIP_ERROR CommandHandler::AddStatusInternal(const ConcreteCommandPath & aComman
return FinishStatus();
}

CHIP_ERROR CommandHandler::AddStatus(const ConcreteCommandPath & aCommandPath, const Protocols::InteractionModel::Status aStatus)
CHIP_ERROR CommandHandler::AddStatus(const ConcreteCommandPath & aCommandPath, const Status aStatus)
{
return AddStatusInternal(aCommandPath, StatusIB(aStatus));
}

CHIP_ERROR CommandHandler::AddClusterSpecificSuccess(const ConcreteCommandPath & aCommandPath, ClusterStatus aClusterStatus)
{
using Protocols::InteractionModel::Status;
return AddStatusInternal(aCommandPath, StatusIB(Status::Success, aClusterStatus));
}

CHIP_ERROR CommandHandler::AddClusterSpecificFailure(const ConcreteCommandPath & aCommandPath, ClusterStatus aClusterStatus)
{
using Protocols::InteractionModel::Status;
return AddStatusInternal(aCommandPath, StatusIB(Status::Failure, aClusterStatus));
}

Expand Down Expand Up @@ -635,7 +627,7 @@ const char * CommandHandler::GetStateStr() const
void CommandHandler::MoveToState(const State aTargetState)
{
mState = aTargetState;
ChipLogDetail(DataManagement, "ICR moving to [%10.10s]", GetStateStr());
ChipLogDetail(DataManagement, "Command handler moving to [%10.10s]", GetStateStr());
}

} // namespace app
Expand Down
17 changes: 5 additions & 12 deletions src/app/CommandHandler.h
View file
Original file line number Diff line number Diff line change
Expand Up @@ -139,15 +139,15 @@ class CommandHandler : public Messaging::ExchangeDelegate
* transaction (i.e. was preceded by a Timed Request). If we reach here,
* the timer verification has already been done.
*/
CHIP_ERROR OnInvokeCommandRequest(Messaging::ExchangeContext * ec, const PayloadHeader & payloadHeader,
System::PacketBufferHandle && payload, bool isTimedInvoke);
Protocols::InteractionModel::Status OnInvokeCommandRequest(Messaging::ExchangeContext * ec, const PayloadHeader & payloadHeader,
System::PacketBufferHandle && payload, bool isTimedInvoke);
CHIP_ERROR AddStatus(const ConcreteCommandPath & aCommandPath, const Protocols::InteractionModel::Status aStatus);

CHIP_ERROR AddClusterSpecificSuccess(const ConcreteCommandPath & aCommandPath, ClusterStatus aClusterStatus);

CHIP_ERROR AddClusterSpecificFailure(const ConcreteCommandPath & aCommandPath, ClusterStatus aClusterStatus);

CHIP_ERROR ProcessInvokeRequest(System::PacketBufferHandle && payload, bool isTimedInvoke);
Protocols::InteractionModel::Status ProcessInvokeRequest(System::PacketBufferHandle && payload, bool isTimedInvoke);
CHIP_ERROR PrepareCommand(const ConcreteCommandPath & aCommandPath, bool aStartDataStruct = true);
CHIP_ERROR FinishCommand(bool aEndDataStruct = true);
CHIP_ERROR PrepareStatus(const ConcreteCommandPath & aCommandPath);
Expand Down Expand Up @@ -250,13 +250,7 @@ class CommandHandler : public Messaging::ExchangeDelegate
friend class CommandHandler::Handle;

CHIP_ERROR OnMessageReceived(Messaging::ExchangeContext * ec, const PayloadHeader & payloadHeader,
System::PacketBufferHandle && payload) override
{
//
// We shouldn't be receiving any further messages on this exchange.
//
return CHIP_ERROR_INCORRECT_STATE;
}
System::PacketBufferHandle && payload) override;

void OnResponseTimeout(Messaging::ExchangeContext * ec) override
{
Expand Down Expand Up @@ -359,14 +353,13 @@ class CommandHandler : public Messaging::ExchangeDelegate

Messaging::ExchangeHolder mExchangeCtx;
Callback * mpCallback = nullptr;

InvokeResponseMessage::Builder mInvokeResponseBuilder;
TLV::TLVType mDataElementContainerType = TLV::kTLVType_NotSpecified;
size_t mPendingWork = 0;
bool mSuppressResponse = false;
bool mTimedRequest = false;

bool mSentStatusResponse = false;

State mState = State::Idle;
chip::System::PacketBufferTLVWriter mCommandMessageWriter;
TLV::TLVWriter mBackupWriter;
Expand Down
49 changes: 32 additions & 17 deletions src/app/CommandSender.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -120,6 +120,7 @@ CHIP_ERROR CommandSender::SendInvokeRequest()
CHIP_ERROR CommandSender::OnMessageReceived(Messaging::ExchangeContext * apExchangeContext, const PayloadHeader & aPayloadHeader,
System::PacketBufferHandle && aPayload)
{
bool suppressErrorStatusResponse = false;
if (mState == State::CommandSent)
{
MoveToState(State::ResponseReceived);
Expand All @@ -130,7 +131,18 @@ CHIP_ERROR CommandSender::OnMessageReceived(Messaging::ExchangeContext * apExcha

if (mState == State::AwaitingTimedStatus)
{
err = HandleTimedStatus(aPayloadHeader, std::move(aPayload));
if (aPayloadHeader.HasMessageType(Protocols::InteractionModel::MsgType::StatusResponse))
{
CHIP_ERROR statusError = CHIP_NO_ERROR;
SuccessOrExit(err = StatusResponse::ProcessStatusResponse(std::move(aPayload), statusError));
suppressErrorStatusResponse = true;
SuccessOrExit(err = statusError);
err = SendInvokeRequest();
}
else
{
err = CHIP_ERROR_INVALID_MESSAGE_TYPE;
}
// Skip all other processing here (which is for the response to the
// invoke request), no matter whether err is success or not.
goto exit;
Expand All @@ -143,30 +155,40 @@ CHIP_ERROR CommandSender::OnMessageReceived(Messaging::ExchangeContext * apExcha
}
else if (aPayloadHeader.HasMessageType(Protocols::InteractionModel::MsgType::StatusResponse))
{
err = StatusResponse::ProcessStatusResponse(std::move(aPayload));
SuccessOrExit(err);
CHIP_ERROR statusError = CHIP_NO_ERROR;
SuccessOrExit(err = StatusResponse::ProcessStatusResponse(std::move(aPayload), statusError));
SuccessOrExit(err = statusError);
err = CHIP_ERROR_INVALID_MESSAGE_TYPE;
}
else
{
err = CHIP_ERROR_INVALID_MESSAGE_TYPE;
}

exit:
if (mpCallback != nullptr)
ResponseMessageHandled(err, apExchangeContext, suppressErrorStatusResponse);
return err;
}

void CommandSender::ResponseMessageHandled(CHIP_ERROR aError, Messaging::ExchangeContext * apExchangeContext,
bool aSuppressErrorStatusResponse)
{
if (aError != CHIP_NO_ERROR)
{
if (err != CHIP_NO_ERROR)
if (!aSuppressErrorStatusResponse)
{
mpCallback->OnError(this, err);
StatusResponse::Send(Protocols::InteractionModel::Status::InvalidAction, apExchangeContext, false /*aExpectResponse*/);
if (mpCallback != nullptr)
{
mpCallback->OnError(this, aError);
}
}
}

if (mState != State::CommandSent)
{
Close();
}
// Else we got a response to a Timed Request and just sent the invoke.

return err;
}

CHIP_ERROR CommandSender::ProcessInvokeResponse(System::PacketBufferHandle && payload)
Expand Down Expand Up @@ -369,13 +391,6 @@ TLV::TLVWriter * CommandSender::GetCommandDataIBTLVWriter()
return mInvokeRequestBuilder.GetInvokeRequests().GetCommandData().GetWriter();
}

CHIP_ERROR CommandSender::HandleTimedStatus(const PayloadHeader & aPayloadHeader, System::PacketBufferHandle && aPayload)
{
ReturnErrorOnFailure(TimedRequest::HandleResponse(aPayloadHeader, std::move(aPayload)));

return SendInvokeRequest();
}

CHIP_ERROR CommandSender::FinishCommand(const Optional<uint16_t> & aTimedInvokeTimeoutMs)
{
ReturnErrorOnFailure(FinishCommand(/* aEndDataStruct = */ false));
Expand Down Expand Up @@ -430,7 +445,7 @@ const char * CommandSender::GetStateStr() const
void CommandSender::MoveToState(const State aTargetState)
{
mState = aTargetState;
ChipLogDetail(DataManagement, "ICR moving to [%10.10s]", GetStateStr());
ChipLogDetail(DataManagement, "Command Sender moving to [%10.10s]", GetStateStr());
}

} // namespace app
Expand Down
10 changes: 2 additions & 8 deletions src/app/CommandSender.h
View file
Original file line number Diff line number Diff line change
Expand Up @@ -259,19 +259,13 @@ class CommandSender final : public Messaging::ExchangeDelegate
CHIP_ERROR ProcessInvokeResponse(System::PacketBufferHandle && payload);
CHIP_ERROR ProcessInvokeResponseIB(InvokeResponseIB::Parser & aInvokeResponse);

// Handle a message received when we are expecting a status response to a
// Timed Request. The caller is assumed to have already checked that our
// exchange context member is the one the message came in on.
//
// If the server returned an error status, that will be returned as an error
// value of CHIP_ERROR.
CHIP_ERROR HandleTimedStatus(const PayloadHeader & aPayloadHeader, System::PacketBufferHandle && aPayload);

// Send our queued-up Invoke Request message. Assumes the exchange is ready
// and mPendingInvokeData is populated.
CHIP_ERROR SendInvokeRequest();

CHIP_ERROR Finalize(System::PacketBufferHandle & commandPacket);
void ResponseMessageHandled(CHIP_ERROR aError, Messaging::ExchangeContext * apExchangeContext,
bool aSuppressErrorStatusResponse);

Messaging::ExchangeHolder mExchangeCtx;
Callback * mpCallback = nullptr;
Expand Down
Loading

0 comments on commit 5bb344c

Please sign in to comment.