blob: 51adf6e8ddc14336514ed98da41e3b1e464f0104 [file] [log] [blame]
/*
*
* Copyright (c) 2021 Project CHIP Authors
* All rights reserved.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
/**
* @file
* This file defines the initiator side of a CHIP Read Interaction.
*
*/
#include <app/AppConfig.h>
#include <app/InteractionModelEngine.h>
#include <app/InteractionModelHelper.h>
#include <app/ReadClient.h>
#include <app/StatusResponse.h>
#include <assert.h>
#include <lib/core/TLVTypes.h>
#include <lib/support/FibonacciUtils.h>
#include <messaging/ReliableMessageMgr.h>
#include <messaging/ReliableMessageProtocolConfig.h>
#include <platform/LockTracker.h>
#include <tracing/metric_event.h>
#include <app-common/zap-generated/cluster-objects.h>
#include <app-common/zap-generated/ids/Attributes.h>
#include <app-common/zap-generated/ids/Clusters.h>
namespace chip {
namespace app {
using Status = Protocols::InteractionModel::Status;
ReadClient::ReadClient(InteractionModelEngine * apImEngine, Messaging::ExchangeManager * apExchangeMgr, Callback & apCallback,
InteractionType aInteractionType) :
mExchange(*this),
mpCallback(apCallback), mOnConnectedCallback(HandleDeviceConnected, this),
mOnConnectionFailureCallback(HandleDeviceConnectionFailure, this)
{
assertChipStackLockedByCurrentThread();
mpExchangeMgr = apExchangeMgr;
mInteractionType = aInteractionType;
mpImEngine = apImEngine;
if (aInteractionType == InteractionType::Subscribe)
{
mpImEngine->AddReadClient(this);
}
}
void ReadClient::ClearActiveSubscriptionState()
{
mIsReporting = false;
mWaitingForFirstPrimingReport = true;
mPendingMoreChunks = false;
mMinIntervalFloorSeconds = 0;
mMaxInterval = 0;
mSubscriptionId = 0;
mIsResubscriptionScheduled = false;
MoveToState(ClientState::Idle);
}
void ReadClient::StopResubscription()
{
CancelLivenessCheckTimer();
CancelResubscribeTimer();
// Only deallocate the paths if they are not already deallocated.
if (mReadPrepareParams.mpAttributePathParamsList != nullptr || mReadPrepareParams.mpEventPathParamsList != nullptr ||
mReadPrepareParams.mpDataVersionFilterList != nullptr)
{
mpCallback.OnDeallocatePaths(std::move(mReadPrepareParams));
// Make sure we will never try to free those pointers again.
mReadPrepareParams.mpAttributePathParamsList = nullptr;
mReadPrepareParams.mAttributePathParamsListSize = 0;
mReadPrepareParams.mpEventPathParamsList = nullptr;
mReadPrepareParams.mEventPathParamsListSize = 0;
mReadPrepareParams.mpDataVersionFilterList = nullptr;
mReadPrepareParams.mDataVersionFilterListSize = 0;
}
}
ReadClient::~ReadClient()
{
assertChipStackLockedByCurrentThread();
if (IsSubscriptionType())
{
StopResubscription();
// Only remove ourselves from the engine's tracker list if we still continue to have a valid pointer to it.
// This won't be the case if the engine shut down before this destructor was called (in which case, mpImEngine
// will point to null)
//
if (mpImEngine)
{
mpImEngine->RemoveReadClient(this);
}
}
}
uint32_t ReadClient::ComputeTimeTillNextSubscription()
{
uint32_t maxWaitTimeInMsec = 0;
uint32_t waitTimeInMsec = 0;
uint32_t minWaitTimeInMsec = 0;
if (mNumRetries <= CHIP_RESUBSCRIBE_MAX_FIBONACCI_STEP_INDEX)
{
maxWaitTimeInMsec = GetFibonacciForIndex(mNumRetries) * CHIP_RESUBSCRIBE_WAIT_TIME_MULTIPLIER_MS;
}
else
{
maxWaitTimeInMsec = CHIP_RESUBSCRIBE_MAX_RETRY_WAIT_INTERVAL_MS;
}
if (maxWaitTimeInMsec != 0)
{
minWaitTimeInMsec = (CHIP_RESUBSCRIBE_MIN_WAIT_TIME_INTERVAL_PERCENT_PER_STEP * maxWaitTimeInMsec) / 100;
waitTimeInMsec = minWaitTimeInMsec + (Crypto::GetRandU32() % (maxWaitTimeInMsec - minWaitTimeInMsec));
}
if (mMinimalResubscribeDelay.count() > waitTimeInMsec)
{
waitTimeInMsec = mMinimalResubscribeDelay.count();
}
return waitTimeInMsec;
}
CHIP_ERROR ReadClient::ScheduleResubscription(uint32_t aTimeTillNextResubscriptionMs, Optional<SessionHandle> aNewSessionHandle,
bool aReestablishCASE)
{
VerifyOrReturnError(IsIdle(), CHIP_ERROR_INCORRECT_STATE);
//
// If we're establishing CASE, make sure we are not provided a new SessionHandle as well.
//
VerifyOrReturnError(!aReestablishCASE || !aNewSessionHandle.HasValue(), CHIP_ERROR_INVALID_ARGUMENT);
if (aNewSessionHandle.HasValue())
{
mReadPrepareParams.mSessionHolder.Grab(aNewSessionHandle.Value());
}
mForceCaseOnNextResub = aReestablishCASE;
if (mForceCaseOnNextResub && mReadPrepareParams.mSessionHolder)
{
// Mark our existing session defunct, so that we will try to
// re-establish it when the timer fires (unless something re-establishes
// before then).
mReadPrepareParams.mSessionHolder->AsSecureSession()->MarkAsDefunct();
}
ReturnErrorOnFailure(
InteractionModelEngine::GetInstance()->GetExchangeManager()->GetSessionManager()->SystemLayer()->StartTimer(
System::Clock::Milliseconds32(aTimeTillNextResubscriptionMs), OnResubscribeTimerCallback, this));
mIsResubscriptionScheduled = true;
return CHIP_NO_ERROR;
}
void ReadClient::Close(CHIP_ERROR aError, bool allowResubscription)
{
if (IsReadType())
{
if (aError != CHIP_NO_ERROR)
{
mpCallback.OnError(aError);
}
}
else
{
if (IsAwaitingInitialReport() || IsAwaitingSubscribeResponse())
{
MATTER_LOG_METRIC_END(Tracing::kMetricDeviceSubscriptionSetup, aError);
}
ClearActiveSubscriptionState();
if (aError != CHIP_NO_ERROR)
{
//
// We infer that re-subscription was requested by virtue of having a non-zero list of event OR attribute paths present
// in mReadPrepareParams. This would only be the case if an application called SendAutoResubscribeRequest which
// populates mReadPrepareParams with the values provided by the application.
//
if (allowResubscription &&
(mReadPrepareParams.mEventPathParamsListSize != 0 || mReadPrepareParams.mAttributePathParamsListSize != 0))
{
CHIP_ERROR originalReason = aError;
aError = mpCallback.OnResubscriptionNeeded(this, aError);
if (aError == CHIP_NO_ERROR)
{
return;
}
if (aError == CHIP_ERROR_LIT_SUBSCRIBE_INACTIVE_TIMEOUT)
{
VerifyOrDie(originalReason == CHIP_ERROR_LIT_SUBSCRIBE_INACTIVE_TIMEOUT);
ChipLogProgress(DataManagement, "ICD device is inactive mark subscription as InactiveICDSubscription");
MoveToState(ClientState::InactiveICDSubscription);
return;
}
}
//
// Either something bad happened when requesting resubscription or the application has decided to not
// continue by returning an error. Let's convey the error back up to the application
// and shut everything down.
//
mpCallback.OnError(aError);
}
StopResubscription();
}
mExchange.Release();
mpCallback.OnDone(this);
}
const char * ReadClient::GetStateStr() const
{
#if CHIP_DETAIL_LOGGING
switch (mState)
{
case ClientState::Idle:
return "Idle";
case ClientState::AwaitingInitialReport:
return "AwaitingInitialReport";
case ClientState::AwaitingSubscribeResponse:
return "AwaitingSubscribeResponse";
case ClientState::SubscriptionActive:
return "SubscriptionActive";
case ClientState::InactiveICDSubscription:
return "InactiveICDSubscription";
}
#endif // CHIP_DETAIL_LOGGING
return "N/A";
}
void ReadClient::MoveToState(const ClientState aTargetState)
{
mState = aTargetState;
ChipLogDetail(DataManagement, "%s ReadClient[%p]: Moving to [%10.10s]", __func__, this, GetStateStr());
}
CHIP_ERROR ReadClient::SendRequest(ReadPrepareParams & aReadPrepareParams)
{
if (mInteractionType == InteractionType::Read)
{
return SendReadRequest(aReadPrepareParams);
}
if (mInteractionType == InteractionType::Subscribe)
{
return SendSubscribeRequest(aReadPrepareParams);
}
return CHIP_ERROR_INVALID_ARGUMENT;
}
CHIP_ERROR ReadClient::SendReadRequest(ReadPrepareParams & aReadPrepareParams)
{
CHIP_ERROR err = CHIP_NO_ERROR;
ChipLogDetail(DataManagement, "%s ReadClient[%p]: Sending Read Request", __func__, this);
VerifyOrReturnError(ClientState::Idle == mState, err = CHIP_ERROR_INCORRECT_STATE);
Span<AttributePathParams> attributePaths(aReadPrepareParams.mpAttributePathParamsList,
aReadPrepareParams.mAttributePathParamsListSize);
Span<EventPathParams> eventPaths(aReadPrepareParams.mpEventPathParamsList, aReadPrepareParams.mEventPathParamsListSize);
Span<DataVersionFilter> dataVersionFilters(aReadPrepareParams.mpDataVersionFilterList,
aReadPrepareParams.mDataVersionFilterListSize);
System::PacketBufferHandle msgBuf;
ReadRequestMessage::Builder request;
System::PacketBufferTLVWriter writer;
InitWriterWithSpaceReserved(writer, kReservedSizeForTLVEncodingOverhead);
ReturnErrorOnFailure(request.Init(&writer));
if (!attributePaths.empty())
{
AttributePathIBs::Builder & attributePathListBuilder = request.CreateAttributeRequests();
ReturnErrorOnFailure(err = request.GetError());
ReturnErrorOnFailure(GenerateAttributePaths(attributePathListBuilder, attributePaths));
}
if (!eventPaths.empty())
{
EventPathIBs::Builder & eventPathListBuilder = request.CreateEventRequests();
ReturnErrorOnFailure(err = request.GetError());
ReturnErrorOnFailure(GenerateEventPaths(eventPathListBuilder, eventPaths));
Optional<EventNumber> eventMin;
ReturnErrorOnFailure(GetMinEventNumber(aReadPrepareParams, eventMin));
if (eventMin.HasValue())
{
EventFilterIBs::Builder & eventFilters = request.CreateEventFilters();
ReturnErrorOnFailure(err = request.GetError());
ReturnErrorOnFailure(eventFilters.GenerateEventFilter(eventMin.Value()));
}
}
ReturnErrorOnFailure(request.IsFabricFiltered(aReadPrepareParams.mIsFabricFiltered).GetError());
bool encodedDataVersionList = false;
TLV::TLVWriter backup;
request.Checkpoint(backup);
DataVersionFilterIBs::Builder & dataVersionFilterListBuilder = request.CreateDataVersionFilters();
ReturnErrorOnFailure(request.GetError());
if (!attributePaths.empty())
{
ReturnErrorOnFailure(GenerateDataVersionFilterList(dataVersionFilterListBuilder, attributePaths, dataVersionFilters,
encodedDataVersionList));
}
ReturnErrorOnFailure(dataVersionFilterListBuilder.GetWriter()->UnreserveBuffer(kReservedSizeForTLVEncodingOverhead));
if (encodedDataVersionList)
{
ReturnErrorOnFailure(dataVersionFilterListBuilder.EndOfDataVersionFilterIBs());
}
else
{
request.Rollback(backup);
}
ReturnErrorOnFailure(request.EndOfReadRequestMessage());
ReturnErrorOnFailure(writer.Finalize(&msgBuf));
VerifyOrReturnError(aReadPrepareParams.mSessionHolder, CHIP_ERROR_MISSING_SECURE_SESSION);
auto exchange = mpExchangeMgr->NewContext(aReadPrepareParams.mSessionHolder.Get().Value(), this);
VerifyOrReturnError(exchange != nullptr, err = CHIP_ERROR_NO_MEMORY);
mExchange.Grab(exchange);
if (aReadPrepareParams.mTimeout == System::Clock::kZero)
{
mExchange->UseSuggestedResponseTimeout(app::kExpectedIMProcessingTime);
}
else
{
mExchange->SetResponseTimeout(aReadPrepareParams.mTimeout);
}
ReturnErrorOnFailure(mExchange->SendMessage(Protocols::InteractionModel::MsgType::ReadRequest, std::move(msgBuf),
Messaging::SendFlags(Messaging::SendMessageFlags::kExpectResponse)));
mPeer = aReadPrepareParams.mSessionHolder->AsSecureSession()->GetPeer();
MoveToState(ClientState::AwaitingInitialReport);
return CHIP_NO_ERROR;
}
CHIP_ERROR ReadClient::GenerateEventPaths(EventPathIBs::Builder & aEventPathsBuilder, const Span<EventPathParams> & aEventPaths)
{
for (auto & event : aEventPaths)
{
VerifyOrReturnError(event.IsValidEventPath(), CHIP_ERROR_IM_MALFORMED_EVENT_PATH_IB);
EventPathIB::Builder & path = aEventPathsBuilder.CreatePath();
ReturnErrorOnFailure(aEventPathsBuilder.GetError());
ReturnErrorOnFailure(path.Encode(event));
}
return aEventPathsBuilder.EndOfEventPaths();
}
CHIP_ERROR ReadClient::GenerateAttributePaths(AttributePathIBs::Builder & aAttributePathIBsBuilder,
const Span<AttributePathParams> & aAttributePaths)
{
for (auto & attribute : aAttributePaths)
{
VerifyOrReturnError(attribute.IsValidAttributePath(), CHIP_ERROR_IM_MALFORMED_ATTRIBUTE_PATH_IB);
AttributePathIB::Builder & path = aAttributePathIBsBuilder.CreatePath();
ReturnErrorOnFailure(aAttributePathIBsBuilder.GetError());
ReturnErrorOnFailure(path.Encode(attribute));
}
return aAttributePathIBsBuilder.EndOfAttributePathIBs();
}
CHIP_ERROR ReadClient::BuildDataVersionFilterList(DataVersionFilterIBs::Builder & aDataVersionFilterIBsBuilder,
const Span<AttributePathParams> & aAttributePaths,
const Span<DataVersionFilter> & aDataVersionFilters,
bool & aEncodedDataVersionList)
{
#if CHIP_PROGRESS_LOGGING
size_t encodedFilterCount = 0;
size_t irrelevantFilterCount = 0;
size_t skippedFilterCount = 0;
#endif
for (auto & filter : aDataVersionFilters)
{
VerifyOrReturnError(filter.IsValidDataVersionFilter(), CHIP_ERROR_INVALID_ARGUMENT);
// If data version filter is for some cluster none of whose attributes are included in our paths, discard this filter.
bool intersected = false;
for (auto & path : aAttributePaths)
{
if (path.IncludesAttributesInCluster(filter))
{
intersected = true;
break;
}
}
if (!intersected)
{
#if CHIP_PROGRESS_LOGGING
++irrelevantFilterCount;
#endif
continue;
}
TLV::TLVWriter backup;
aDataVersionFilterIBsBuilder.Checkpoint(backup);
CHIP_ERROR err = EncodeDataVersionFilter(aDataVersionFilterIBsBuilder, filter);
if (err == CHIP_NO_ERROR)
{
#if CHIP_PROGRESS_LOGGING
++encodedFilterCount;
#endif
aEncodedDataVersionList = true;
}
else if (err == CHIP_ERROR_NO_MEMORY || err == CHIP_ERROR_BUFFER_TOO_SMALL)
{
// Packet is full, ignore the rest of the list
aDataVersionFilterIBsBuilder.Rollback(backup);
#if CHIP_PROGRESS_LOGGING
ssize_t nonSkippedFilterCount = &filter - aDataVersionFilters.data();
skippedFilterCount = aDataVersionFilters.size() - static_cast<size_t>(nonSkippedFilterCount);
#endif // CHIP_PROGRESS_LOGGING
break;
}
else
{
return err;
}
}
ChipLogProgress(DataManagement,
"%lu data version filters provided, %lu not relevant, %lu encoded, %lu skipped due to lack of space",
static_cast<unsigned long>(aDataVersionFilters.size()), static_cast<unsigned long>(irrelevantFilterCount),
static_cast<unsigned long>(encodedFilterCount), static_cast<unsigned long>(skippedFilterCount));
return CHIP_NO_ERROR;
}
CHIP_ERROR ReadClient::EncodeDataVersionFilter(DataVersionFilterIBs::Builder & aDataVersionFilterIBsBuilder,
DataVersionFilter const & aFilter)
{
// Caller has checked aFilter.IsValidDataVersionFilter()
DataVersionFilterIB::Builder & filterIB = aDataVersionFilterIBsBuilder.CreateDataVersionFilter();
ReturnErrorOnFailure(aDataVersionFilterIBsBuilder.GetError());
ClusterPathIB::Builder & path = filterIB.CreatePath();
ReturnErrorOnFailure(filterIB.GetError());
ReturnErrorOnFailure(path.Endpoint(aFilter.mEndpointId).Cluster(aFilter.mClusterId).EndOfClusterPathIB());
ReturnErrorOnFailure(filterIB.DataVersion(aFilter.mDataVersion.Value()).EndOfDataVersionFilterIB());
return CHIP_NO_ERROR;
}
CHIP_ERROR ReadClient::GenerateDataVersionFilterList(DataVersionFilterIBs::Builder & aDataVersionFilterIBsBuilder,
const Span<AttributePathParams> & aAttributePaths,
const Span<DataVersionFilter> & aDataVersionFilters,
bool & aEncodedDataVersionList)
{
// Give the callback a chance first, otherwise use the list we have, if any.
ReturnErrorOnFailure(
mpCallback.OnUpdateDataVersionFilterList(aDataVersionFilterIBsBuilder, aAttributePaths, aEncodedDataVersionList));
if (!aEncodedDataVersionList)
{
ReturnErrorOnFailure(BuildDataVersionFilterList(aDataVersionFilterIBsBuilder, aAttributePaths, aDataVersionFilters,
aEncodedDataVersionList));
}
return CHIP_NO_ERROR;
}
void ReadClient::OnActiveModeNotification()
{
// This function just tries to complete the deferred resubscription logic in `OnLivenessTimeoutCallback`.
VerifyOrDie(mpImEngine->InActiveReadClientList(this));
// If we are not in InactiveICDSubscription state, that means the liveness timeout has not been reached. Simply do nothing.
VerifyOrReturn(IsInactiveICDSubscription());
// When we reach here, the subscription definitely exceeded the liveness timeout. Just continue the unfinished resubscription
// logic in `OnLivenessTimeoutCallback`.
TriggerResubscriptionForLivenessTimeout(CHIP_ERROR_TIMEOUT);
}
void ReadClient::OnPeerTypeChange(PeerType aType)
{
VerifyOrDie(mpImEngine->InActiveReadClientList(this));
mIsPeerLIT = (aType == PeerType::kLITICD);
ChipLogProgress(DataManagement, "Peer is now %s LIT ICD.", mIsPeerLIT ? "a" : "not a");
// If the peer is no longer LIT, try to wake up the subscription and do resubscribe when necessary.
if (!mIsPeerLIT)
{
OnActiveModeNotification();
}
}
CHIP_ERROR ReadClient::OnMessageReceived(Messaging::ExchangeContext * apExchangeContext, const PayloadHeader & aPayloadHeader,
System::PacketBufferHandle && aPayload)
{
CHIP_ERROR err = CHIP_NO_ERROR;
Status status = Status::InvalidAction;
VerifyOrExit(!IsIdle() && !IsInactiveICDSubscription(), err = CHIP_ERROR_INCORRECT_STATE);
if (aPayloadHeader.HasMessageType(Protocols::InteractionModel::MsgType::ReportData))
{
err = ProcessReportData(std::move(aPayload), ReportType::kContinuingTransaction);
}
else if (aPayloadHeader.HasMessageType(Protocols::InteractionModel::MsgType::SubscribeResponse))
{
ChipLogProgress(DataManagement, "SubscribeResponse is received");
VerifyOrExit(apExchangeContext == mExchange.Get(), err = CHIP_ERROR_INCORRECT_STATE);
err = ProcessSubscribeResponse(std::move(aPayload));
MATTER_LOG_METRIC_END(Tracing::kMetricDeviceSubscriptionSetup, err);
}
else if (aPayloadHeader.HasMessageType(Protocols::InteractionModel::MsgType::StatusResponse))
{
VerifyOrExit(apExchangeContext == mExchange.Get(), err = CHIP_ERROR_INCORRECT_STATE);
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 (err != CHIP_NO_ERROR)
{
if (err == CHIP_ERROR_INVALID_SUBSCRIPTION)
{
status = Status::InvalidSubscription;
}
StatusResponse::Send(status, apExchangeContext, false /*aExpectResponse*/);
}
if ((!IsSubscriptionType() && !mPendingMoreChunks) || err != CHIP_NO_ERROR)
{
Close(err);
}
return err;
}
void ReadClient::OnUnsolicitedReportData(Messaging::ExchangeContext * apExchangeContext, System::PacketBufferHandle && aPayload)
{
Status status = Status::Success;
mExchange.Grab(apExchangeContext);
//
// Let's update the session we're tracking in our SessionHolder to that associated with the message that was just received.
// This CAN be different from the one we were tracking before, since the server is permitted to send exchanges on any valid
// session to us, of which there could be multiple.
//
// Since receipt of a message is proof of a working session on the peer, it's always best to update to that if possible
// to maximize our chances of success later.
//
mReadPrepareParams.mSessionHolder.Grab(mExchange->GetSessionHandle());
CHIP_ERROR err = ProcessReportData(std::move(aPayload), ReportType::kUnsolicited);
if (err != CHIP_NO_ERROR)
{
if (err == CHIP_ERROR_INVALID_SUBSCRIPTION)
{
status = Status::InvalidSubscription;
}
else
{
status = Status::InvalidAction;
}
StatusResponse::Send(status, mExchange.Get(), false /*aExpectResponse*/);
Close(err);
}
}
CHIP_ERROR ReadClient::ProcessReportData(System::PacketBufferHandle && aPayload, ReportType aReportType)
{
CHIP_ERROR err = CHIP_NO_ERROR;
ReportDataMessage::Parser report;
bool suppressResponse = true;
SubscriptionId subscriptionId = 0;
EventReportIBs::Parser eventReportIBs;
AttributeReportIBs::Parser attributeReportIBs;
System::PacketBufferTLVReader reader;
reader.Init(std::move(aPayload));
err = report.Init(reader);
SuccessOrExit(err);
#if CHIP_CONFIG_IM_PRETTY_PRINT
if (aReportType != ReportType::kUnsolicited)
{
report.PrettyPrint();
}
#endif
err = report.GetSuppressResponse(&suppressResponse);
if (CHIP_END_OF_TLV == err)
{
suppressResponse = false;
err = CHIP_NO_ERROR;
}
SuccessOrExit(err);
err = report.GetSubscriptionId(&subscriptionId);
if (CHIP_NO_ERROR == err)
{
VerifyOrExit(IsSubscriptionType(), err = CHIP_ERROR_INVALID_ARGUMENT);
if (mWaitingForFirstPrimingReport)
{
mSubscriptionId = subscriptionId;
}
else if (!IsMatchingSubscriptionId(subscriptionId))
{
err = CHIP_ERROR_INVALID_SUBSCRIPTION;
}
}
else if (CHIP_END_OF_TLV == err)
{
if (IsSubscriptionType())
{
err = CHIP_ERROR_INVALID_ARGUMENT;
}
else
{
err = CHIP_NO_ERROR;
}
}
SuccessOrExit(err);
err = report.GetMoreChunkedMessages(&mPendingMoreChunks);
if (CHIP_END_OF_TLV == err)
{
mPendingMoreChunks = false;
err = CHIP_NO_ERROR;
}
SuccessOrExit(err);
err = report.GetEventReports(&eventReportIBs);
if (err == CHIP_END_OF_TLV)
{
err = CHIP_NO_ERROR;
}
else if (err == CHIP_NO_ERROR)
{
chip::TLV::TLVReader EventReportsReader;
eventReportIBs.GetReader(&EventReportsReader);
err = ProcessEventReportIBs(EventReportsReader);
}
SuccessOrExit(err);
err = report.GetAttributeReportIBs(&attributeReportIBs);
if (err == CHIP_END_OF_TLV)
{
err = CHIP_NO_ERROR;
}
else if (err == CHIP_NO_ERROR)
{
TLV::TLVReader attributeReportIBsReader;
attributeReportIBs.GetReader(&attributeReportIBsReader);
err = ProcessAttributeReportIBs(attributeReportIBsReader);
}
SuccessOrExit(err);
if (mIsReporting && !mPendingMoreChunks)
{
mpCallback.OnReportEnd();
mIsReporting = false;
}
SuccessOrExit(err = report.ExitContainer());
exit:
if (IsSubscriptionType())
{
if (IsAwaitingInitialReport())
{
MoveToState(ClientState::AwaitingSubscribeResponse);
}
else if (IsSubscriptionActive() && err == CHIP_NO_ERROR)
{
//
// Only refresh the liveness check timer if we've successfully established
// a subscription and have a valid value for mMaxInterval which the function
// relies on.
//
mpCallback.NotifySubscriptionStillActive(*this);
err = RefreshLivenessCheckTimer();
}
}
if (!suppressResponse && err == CHIP_NO_ERROR)
{
bool noResponseExpected = IsSubscriptionActive() && !mPendingMoreChunks;
err = StatusResponse::Send(Status::Success, mExchange.Get(), !noResponseExpected);
}
mWaitingForFirstPrimingReport = false;
return err;
}
void ReadClient::OnResponseTimeout(Messaging::ExchangeContext * apExchangeContext)
{
ChipLogError(DataManagement, "Time out! failed to receive report data from Exchange: " ChipLogFormatExchange,
ChipLogValueExchange(apExchangeContext));
Close(CHIP_ERROR_TIMEOUT);
}
CHIP_ERROR ReadClient::ReadICDOperatingModeFromAttributeDataIB(TLV::TLVReader && aReader, PeerType & aType)
{
Clusters::IcdManagement::Attributes::OperatingMode::TypeInfo::DecodableType operatingMode;
CHIP_ERROR err = DataModel::Decode(aReader, operatingMode);
ReturnErrorOnFailure(err);
switch (operatingMode)
{
case Clusters::IcdManagement::OperatingModeEnum::kSit:
aType = PeerType::kNormal;
break;
case Clusters::IcdManagement::OperatingModeEnum::kLit:
aType = PeerType::kLITICD;
break;
default:
err = CHIP_ERROR_INVALID_ARGUMENT;
break;
}
return err;
}
CHIP_ERROR ReadClient::ProcessAttributePath(AttributePathIB::Parser & aAttributePathParser,
ConcreteDataAttributePath & aAttributePath)
{
// The ReportData must contain a concrete attribute path. Don't validate ID
// ranges here, so we can tell apart "malformed data" and "out of range
// IDs".
CHIP_ERROR err = CHIP_NO_ERROR;
// The ReportData must contain a concrete attribute path
err = aAttributePathParser.GetConcreteAttributePath(aAttributePath, AttributePathIB::ValidateIdRanges::kNo);
VerifyOrReturnError(err == CHIP_NO_ERROR, CHIP_ERROR_IM_MALFORMED_ATTRIBUTE_PATH_IB);
return CHIP_NO_ERROR;
}
void ReadClient::NoteReportingData()
{
if (!mIsReporting)
{
mpCallback.OnReportBegin();
mIsReporting = true;
}
}
CHIP_ERROR ReadClient::ProcessAttributeReportIBs(TLV::TLVReader & aAttributeReportIBsReader)
{
CHIP_ERROR err = CHIP_NO_ERROR;
while (CHIP_NO_ERROR == (err = aAttributeReportIBsReader.Next()))
{
TLV::TLVReader dataReader;
AttributeReportIB::Parser report;
AttributeDataIB::Parser data;
AttributeStatusIB::Parser status;
AttributePathIB::Parser path;
ConcreteDataAttributePath attributePath;
StatusIB statusIB;
TLV::TLVReader reader = aAttributeReportIBsReader;
ReturnErrorOnFailure(report.Init(reader));
err = report.GetAttributeStatus(&status);
if (CHIP_NO_ERROR == err)
{
StatusIB::Parser errorStatus;
ReturnErrorOnFailure(status.GetPath(&path));
ReturnErrorOnFailure(ProcessAttributePath(path, attributePath));
if (!attributePath.IsValid())
{
// Don't fail the entire read or subscription when there is an
// out-of-range ID. Just skip that one AttributeReportIB.
ChipLogError(DataManagement,
"Skipping AttributeStatusIB with out-of-range IDs: (%d, " ChipLogFormatMEI ", " ChipLogFormatMEI ") ",
attributePath.mEndpointId, ChipLogValueMEI(attributePath.mClusterId),
ChipLogValueMEI(attributePath.mAttributeId));
continue;
}
ReturnErrorOnFailure(status.GetErrorStatus(&errorStatus));
ReturnErrorOnFailure(errorStatus.DecodeStatusIB(statusIB));
NoteReportingData();
mpCallback.OnAttributeData(attributePath, nullptr, statusIB);
}
else if (CHIP_END_OF_TLV == err)
{
ReturnErrorOnFailure(report.GetAttributeData(&data));
ReturnErrorOnFailure(data.GetPath(&path));
ReturnErrorOnFailure(ProcessAttributePath(path, attributePath));
if (!attributePath.IsValid())
{
// Don't fail the entire read or subscription when there is an
// out-of-range ID. Just skip that one AttributeReportIB.
ChipLogError(DataManagement,
"Skipping AttributeDataIB with out-of-range IDs: (%d, " ChipLogFormatMEI ", " ChipLogFormatMEI ") ",
attributePath.mEndpointId, ChipLogValueMEI(attributePath.mClusterId),
ChipLogValueMEI(attributePath.mAttributeId));
continue;
}
DataVersion version = 0;
ReturnErrorOnFailure(data.GetDataVersion(&version));
attributePath.mDataVersion.SetValue(version);
if (mReadPrepareParams.mpDataVersionFilterList != nullptr)
{
UpdateDataVersionFilters(attributePath);
}
ReturnErrorOnFailure(data.GetData(&dataReader));
// The element in an array may be another array -- so we should only set the list operation when we are handling the
// whole list.
if (!attributePath.IsListOperation() && dataReader.GetType() == TLV::kTLVType_Array)
{
attributePath.mListOp = ConcreteDataAttributePath::ListOperation::ReplaceAll;
}
if (attributePath.MatchesConcreteAttributePath(ConcreteAttributePath(
kRootEndpointId, Clusters::IcdManagement::Id, Clusters::IcdManagement::Attributes::OperatingMode::Id)))
{
PeerType peerType;
TLV::TLVReader operatingModeTlvReader;
operatingModeTlvReader.Init(dataReader);
if (CHIP_NO_ERROR == ReadICDOperatingModeFromAttributeDataIB(std::move(operatingModeTlvReader), peerType))
{
// It is safe to call `OnPeerTypeChange` since we are in the middle of parsing the attribute data, And
// the subscription should be active so `OnActiveModeNotification` is a no-op in this case.
InteractionModelEngine::GetInstance()->OnPeerTypeChange(mPeer, peerType);
}
else
{
ChipLogError(DataManagement, "Failed to get ICD state from attribute data with error'%" CHIP_ERROR_FORMAT "'",
err.Format());
}
}
NoteReportingData();
mpCallback.OnAttributeData(attributePath, &dataReader, statusIB);
}
}
if (CHIP_END_OF_TLV == err)
{
err = CHIP_NO_ERROR;
}
return err;
}
CHIP_ERROR ReadClient::ProcessEventReportIBs(TLV::TLVReader & aEventReportIBsReader)
{
CHIP_ERROR err = CHIP_NO_ERROR;
while (CHIP_NO_ERROR == (err = aEventReportIBsReader.Next()))
{
TLV::TLVReader dataReader;
EventReportIB::Parser report;
EventDataIB::Parser data;
EventHeader header;
StatusIB statusIB; // Default value for statusIB is success.
TLV::TLVReader reader = aEventReportIBsReader;
ReturnErrorOnFailure(report.Init(reader));
err = report.GetEventData(&data);
if (err == CHIP_NO_ERROR)
{
header.mTimestamp = mEventTimestamp;
ReturnErrorOnFailure(data.DecodeEventHeader(header));
mEventTimestamp = header.mTimestamp;
ReturnErrorOnFailure(data.GetData(&dataReader));
//
// Update the event number being tracked in mReadPrepareParams in case
// we want to send it in the next SubscribeRequest message to convey
// the event number for which we have already received an event.
//
mReadPrepareParams.mEventNumber.SetValue(header.mEventNumber + 1);
NoteReportingData();
mpCallback.OnEventData(header, &dataReader, nullptr);
}
else if (err == CHIP_END_OF_TLV)
{
EventStatusIB::Parser status;
EventPathIB::Parser pathIB;
StatusIB::Parser statusIBParser;
ReturnErrorOnFailure(report.GetEventStatus(&status));
ReturnErrorOnFailure(status.GetPath(&pathIB));
ReturnErrorOnFailure(pathIB.GetEventPath(&header.mPath));
ReturnErrorOnFailure(status.GetErrorStatus(&statusIBParser));
ReturnErrorOnFailure(statusIBParser.DecodeStatusIB(statusIB));
NoteReportingData();
mpCallback.OnEventData(header, nullptr, &statusIB);
}
}
if (CHIP_END_OF_TLV == err)
{
err = CHIP_NO_ERROR;
}
return err;
}
void ReadClient::OverrideLivenessTimeout(System::Clock::Timeout aLivenessTimeout)
{
mLivenessTimeoutOverride = aLivenessTimeout;
auto err = RefreshLivenessCheckTimer();
if (err != CHIP_NO_ERROR)
{
Close(err);
}
}
CHIP_ERROR ReadClient::RefreshLivenessCheckTimer()
{
CHIP_ERROR err = CHIP_NO_ERROR;
VerifyOrReturnError(IsSubscriptionActive(), CHIP_ERROR_INCORRECT_STATE);
CancelLivenessCheckTimer();
System::Clock::Timeout timeout;
ReturnErrorOnFailure(ComputeLivenessCheckTimerTimeout(&timeout));
// EFR32/MBED/INFINION/K32W's chrono count return long unsigned, but other platform returns unsigned
ChipLogProgress(
DataManagement,
"Refresh LivenessCheckTime for %lu milliseconds with SubscriptionId = 0x%08" PRIx32 " Peer = %02x:" ChipLogFormatX64,
static_cast<long unsigned>(timeout.count()), mSubscriptionId, GetFabricIndex(), ChipLogValueX64(GetPeerNodeId()));
err = InteractionModelEngine::GetInstance()->GetExchangeManager()->GetSessionManager()->SystemLayer()->StartTimer(
timeout, OnLivenessTimeoutCallback, this);
return err;
}
CHIP_ERROR ReadClient::ComputeLivenessCheckTimerTimeout(System::Clock::Timeout * aTimeout)
{
if (mLivenessTimeoutOverride != System::Clock::kZero)
{
*aTimeout = mLivenessTimeoutOverride;
return CHIP_NO_ERROR;
}
VerifyOrReturnError(mReadPrepareParams.mSessionHolder, CHIP_ERROR_INCORRECT_STATE);
//
// To calculate the duration we're willing to wait for a report to come to us, we take into account the maximum interval of
// the subscription AND the time it takes for the report to make it to us in the worst case.
//
// We have no way to estimate what the network latency will be, but we do know the other side will time out its ReportData
// after its computed round-trip timeout plus the processing time it gives us (app::kExpectedIMProcessingTime). Once it
// times out, assuming it sent the report at all, there's no point in us thinking we still have a subscription.
//
// We can't use ComputeRoundTripTimeout() on the session for two reasons: we want the roundtrip timeout from the point of
// view of the peer, not us, and we want to start off with the assumption the peer will likely have, which is that we are
// idle, whereas ComputeRoundTripTimeout() uses the current activity state of the peer.
//
// So recompute the round-trip timeout directly. Assume MRP, since in practice that is likely what is happening.
auto & peerMRPConfig = mReadPrepareParams.mSessionHolder->GetRemoteMRPConfig();
// Peer will assume we are idle (hence we pass kZero to GetMessageReceiptTimeout()), but will assume we treat it as active
// for the response, so to match the retransmission timeout computation for the message back to the peeer, we should treat
// it as active.
auto roundTripTimeout = mReadPrepareParams.mSessionHolder->GetMessageReceiptTimeout(System::Clock::kZero) +
kExpectedIMProcessingTime +
GetRetransmissionTimeout(peerMRPConfig.mActiveRetransTimeout, peerMRPConfig.mIdleRetransTimeout,
System::SystemClock().GetMonotonicTimestamp(), peerMRPConfig.mActiveThresholdTime);
*aTimeout = System::Clock::Seconds16(mMaxInterval) + roundTripTimeout;
return CHIP_NO_ERROR;
}
void ReadClient::CancelLivenessCheckTimer()
{
InteractionModelEngine::GetInstance()->GetExchangeManager()->GetSessionManager()->SystemLayer()->CancelTimer(
OnLivenessTimeoutCallback, this);
}
void ReadClient::CancelResubscribeTimer()
{
InteractionModelEngine::GetInstance()->GetExchangeManager()->GetSessionManager()->SystemLayer()->CancelTimer(
OnResubscribeTimerCallback, this);
mIsResubscriptionScheduled = false;
}
void ReadClient::OnLivenessTimeoutCallback(System::Layer * apSystemLayer, void * apAppState)
{
ReadClient * const _this = reinterpret_cast<ReadClient *>(apAppState);
// TODO: add a more specific error here for liveness timeout failure to distinguish between other classes of timeouts (i.e
// response timeouts).
CHIP_ERROR subscriptionTerminationCause = CHIP_ERROR_TIMEOUT;
//
// Might as well try to see if this instance exists in the tracked list in the IM.
// This might blow-up if either the client has since been free'ed (use-after-free), or if the engine has since
// been shutdown at which point the client wouldn't exist in the active read client list.
//
VerifyOrDie(_this->mpImEngine->InActiveReadClientList(_this));
ChipLogError(DataManagement,
"Subscription Liveness timeout with SubscriptionID = 0x%08" PRIx32 ", Peer = %02x:" ChipLogFormatX64,
_this->mSubscriptionId, _this->GetFabricIndex(), ChipLogValueX64(_this->GetPeerNodeId()));
if (_this->mIsPeerLIT)
{
subscriptionTerminationCause = CHIP_ERROR_LIT_SUBSCRIBE_INACTIVE_TIMEOUT;
}
_this->TriggerResubscriptionForLivenessTimeout(subscriptionTerminationCause);
}
void ReadClient::TriggerResubscriptionForLivenessTimeout(CHIP_ERROR aReason)
{
// We didn't get a message from the server on time; it's possible that it no
// longer has a useful CASE session to us. Mark defunct all sessions that
// have not seen peer activity in at least as long as our session.
const auto & holder = mReadPrepareParams.mSessionHolder;
if (holder)
{
System::Clock::Timestamp lastPeerActivity = holder->AsSecureSession()->GetLastPeerActivityTime();
mpImEngine->GetExchangeManager()->GetSessionManager()->ForEachMatchingSession(mPeer, [&lastPeerActivity](auto * session) {
if (!session->IsCASESession())
{
return;
}
if (session->GetLastPeerActivityTime() > lastPeerActivity)
{
return;
}
session->MarkAsDefunct();
});
}
Close(aReason);
}
CHIP_ERROR ReadClient::ProcessSubscribeResponse(System::PacketBufferHandle && aPayload)
{
System::PacketBufferTLVReader reader;
reader.Init(std::move(aPayload));
SubscribeResponseMessage::Parser subscribeResponse;
ReturnErrorOnFailure(subscribeResponse.Init(reader));
#if CHIP_CONFIG_IM_PRETTY_PRINT
subscribeResponse.PrettyPrint();
#endif
SubscriptionId subscriptionId = 0;
VerifyOrReturnError(subscribeResponse.GetSubscriptionId(&subscriptionId) == CHIP_NO_ERROR, CHIP_ERROR_INVALID_ARGUMENT);
VerifyOrReturnError(IsMatchingSubscriptionId(subscriptionId), CHIP_ERROR_INVALID_SUBSCRIPTION);
ReturnErrorOnFailure(subscribeResponse.GetMaxInterval(&mMaxInterval));
ChipLogProgress(DataManagement,
"Subscription established with SubscriptionID = 0x%08" PRIx32 " MinInterval = %u"
"s MaxInterval = %us Peer = %02x:" ChipLogFormatX64,
mSubscriptionId, mMinIntervalFloorSeconds, mMaxInterval, GetFabricIndex(), ChipLogValueX64(GetPeerNodeId()));
ReturnErrorOnFailure(subscribeResponse.ExitContainer());
MoveToState(ClientState::SubscriptionActive);
mpCallback.OnSubscriptionEstablished(subscriptionId);
mNumRetries = 0;
ReturnErrorOnFailure(RefreshLivenessCheckTimer());
return CHIP_NO_ERROR;
}
CHIP_ERROR ReadClient::SendAutoResubscribeRequest(ReadPrepareParams && aReadPrepareParams)
{
// Make sure we don't use minimal resubscribe delays from previous attempts
// for this one.
mMinimalResubscribeDelay = System::Clock::kZero;
mReadPrepareParams = std::move(aReadPrepareParams);
CHIP_ERROR err = SendSubscribeRequest(mReadPrepareParams);
if (err != CHIP_NO_ERROR)
{
StopResubscription();
}
return err;
}
CHIP_ERROR ReadClient::SendAutoResubscribeRequest(const ScopedNodeId & aPublisherId, ReadPrepareParams && aReadPrepareParams)
{
mPeer = aPublisherId;
mReadPrepareParams = std::move(aReadPrepareParams);
CHIP_ERROR err = EstablishSessionToPeer();
if (err != CHIP_NO_ERROR)
{
// Make sure we call our callback's OnDeallocatePaths.
StopResubscription();
}
return err;
}
CHIP_ERROR ReadClient::SendSubscribeRequest(const ReadPrepareParams & aReadPrepareParams)
{
VerifyOrReturnError(aReadPrepareParams.mMinIntervalFloorSeconds <= aReadPrepareParams.mMaxIntervalCeilingSeconds,
CHIP_ERROR_INVALID_ARGUMENT);
auto err = SendSubscribeRequestImpl(aReadPrepareParams);
if (CHIP_NO_ERROR != err)
{
MATTER_LOG_METRIC_END(Tracing::kMetricDeviceSubscriptionSetup, err);
}
return err;
}
CHIP_ERROR ReadClient::SendSubscribeRequestImpl(const ReadPrepareParams & aReadPrepareParams)
{
MATTER_LOG_METRIC_BEGIN(Tracing::kMetricDeviceSubscriptionSetup);
VerifyOrReturnError(ClientState::Idle == mState, CHIP_ERROR_INCORRECT_STATE);
if (&aReadPrepareParams != &mReadPrepareParams)
{
mReadPrepareParams.mSessionHolder = aReadPrepareParams.mSessionHolder;
}
mIsPeerLIT = aReadPrepareParams.mIsPeerLIT;
mMinIntervalFloorSeconds = aReadPrepareParams.mMinIntervalFloorSeconds;
// Todo: Remove the below, Update span in ReadPrepareParams
Span<AttributePathParams> attributePaths(aReadPrepareParams.mpAttributePathParamsList,
aReadPrepareParams.mAttributePathParamsListSize);
Span<EventPathParams> eventPaths(aReadPrepareParams.mpEventPathParamsList, aReadPrepareParams.mEventPathParamsListSize);
Span<DataVersionFilter> dataVersionFilters(aReadPrepareParams.mpDataVersionFilterList,
aReadPrepareParams.mDataVersionFilterListSize);
System::PacketBufferHandle msgBuf;
System::PacketBufferTLVWriter writer;
SubscribeRequestMessage::Builder request;
InitWriterWithSpaceReserved(writer, kReservedSizeForTLVEncodingOverhead);
ReturnErrorOnFailure(request.Init(&writer));
request.KeepSubscriptions(aReadPrepareParams.mKeepSubscriptions)
.MinIntervalFloorSeconds(aReadPrepareParams.mMinIntervalFloorSeconds)
.MaxIntervalCeilingSeconds(aReadPrepareParams.mMaxIntervalCeilingSeconds);
if (!attributePaths.empty())
{
AttributePathIBs::Builder & attributePathListBuilder = request.CreateAttributeRequests();
ReturnErrorOnFailure(attributePathListBuilder.GetError());
ReturnErrorOnFailure(GenerateAttributePaths(attributePathListBuilder, attributePaths));
}
if (!eventPaths.empty())
{
EventPathIBs::Builder & eventPathListBuilder = request.CreateEventRequests();
ReturnErrorOnFailure(eventPathListBuilder.GetError());
ReturnErrorOnFailure(GenerateEventPaths(eventPathListBuilder, eventPaths));
Optional<EventNumber> eventMin;
ReturnErrorOnFailure(GetMinEventNumber(aReadPrepareParams, eventMin));
if (eventMin.HasValue())
{
EventFilterIBs::Builder & eventFilters = request.CreateEventFilters();
ReturnErrorOnFailure(request.GetError());
ReturnErrorOnFailure(eventFilters.GenerateEventFilter(eventMin.Value()));
}
}
ReturnErrorOnFailure(request.IsFabricFiltered(aReadPrepareParams.mIsFabricFiltered).GetError());
bool encodedDataVersionList = false;
TLV::TLVWriter backup;
request.Checkpoint(backup);
DataVersionFilterIBs::Builder & dataVersionFilterListBuilder = request.CreateDataVersionFilters();
ReturnErrorOnFailure(request.GetError());
if (!attributePaths.empty())
{
ReturnErrorOnFailure(GenerateDataVersionFilterList(dataVersionFilterListBuilder, attributePaths, dataVersionFilters,
encodedDataVersionList));
}
ReturnErrorOnFailure(dataVersionFilterListBuilder.GetWriter()->UnreserveBuffer(kReservedSizeForTLVEncodingOverhead));
if (encodedDataVersionList)
{
ReturnErrorOnFailure(dataVersionFilterListBuilder.EndOfDataVersionFilterIBs());
}
else
{
request.Rollback(backup);
}
ReturnErrorOnFailure(request.EndOfSubscribeRequestMessage());
ReturnErrorOnFailure(writer.Finalize(&msgBuf));
VerifyOrReturnError(aReadPrepareParams.mSessionHolder, CHIP_ERROR_MISSING_SECURE_SESSION);
auto exchange = mpExchangeMgr->NewContext(aReadPrepareParams.mSessionHolder.Get().Value(), this);
if (exchange == nullptr)
{
if (aReadPrepareParams.mSessionHolder->AsSecureSession()->IsActiveSession())
{
return CHIP_ERROR_NO_MEMORY;
}
// Trying to subscribe with a defunct session somehow.
return CHIP_ERROR_INCORRECT_STATE;
}
mExchange.Grab(exchange);
if (aReadPrepareParams.mTimeout == System::Clock::kZero)
{
mExchange->UseSuggestedResponseTimeout(app::kExpectedIMProcessingTime);
}
else
{
mExchange->SetResponseTimeout(aReadPrepareParams.mTimeout);
}
ReturnErrorOnFailure(mExchange->SendMessage(Protocols::InteractionModel::MsgType::SubscribeRequest, std::move(msgBuf),
Messaging::SendFlags(Messaging::SendMessageFlags::kExpectResponse)));
mPeer = aReadPrepareParams.mSessionHolder->AsSecureSession()->GetPeer();
MoveToState(ClientState::AwaitingInitialReport);
return CHIP_NO_ERROR;
}
CHIP_ERROR ReadClient::DefaultResubscribePolicy(CHIP_ERROR aTerminationCause)
{
if (aTerminationCause == CHIP_ERROR_LIT_SUBSCRIBE_INACTIVE_TIMEOUT)
{
ChipLogProgress(DataManagement, "ICD device is inactive, skipping scheduling resubscribe within DefaultResubscribePolicy");
return CHIP_ERROR_LIT_SUBSCRIBE_INACTIVE_TIMEOUT;
}
VerifyOrReturnError(IsIdle(), CHIP_ERROR_INCORRECT_STATE);
auto timeTillNextResubscription = ComputeTimeTillNextSubscription();
ChipLogProgress(DataManagement,
"Will try to resubscribe to %02x:" ChipLogFormatX64 " at retry index %" PRIu32 " after %" PRIu32
"ms due to error %" CHIP_ERROR_FORMAT,
GetFabricIndex(), ChipLogValueX64(GetPeerNodeId()), mNumRetries, timeTillNextResubscription,
aTerminationCause.Format());
return ScheduleResubscription(timeTillNextResubscription, NullOptional, aTerminationCause == CHIP_ERROR_TIMEOUT);
}
void ReadClient::HandleDeviceConnected(void * context, Messaging::ExchangeManager & exchangeMgr,
const SessionHandle & sessionHandle)
{
ReadClient * const _this = static_cast<ReadClient *>(context);
VerifyOrDie(_this != nullptr);
ChipLogProgress(DataManagement, "HandleDeviceConnected");
_this->mReadPrepareParams.mSessionHolder.Grab(sessionHandle);
_this->mpExchangeMgr = &exchangeMgr;
_this->mpCallback.OnCASESessionEstablished(sessionHandle, _this->mReadPrepareParams);
auto err = _this->SendSubscribeRequest(_this->mReadPrepareParams);
if (err != CHIP_NO_ERROR)
{
_this->Close(err);
}
}
void ReadClient::HandleDeviceConnectionFailure(void * context, const OperationalSessionSetup::ConnnectionFailureInfo & failureInfo)
{
ReadClient * const _this = static_cast<ReadClient *>(context);
VerifyOrDie(_this != nullptr);
ChipLogError(DataManagement, "Failed to establish CASE for re-subscription with error '%" CHIP_ERROR_FORMAT "'",
failureInfo.error.Format());
#if CHIP_CONFIG_ENABLE_BUSY_HANDLING_FOR_OPERATIONAL_SESSION_SETUP
#if CHIP_DETAIL_LOGGING
if (failureInfo.requestedBusyDelay.HasValue())
{
ChipLogDetail(DataManagement, "Will delay resubscription by %u ms due to BUSY response",
failureInfo.requestedBusyDelay.Value().count());
}
#endif // CHIP_DETAIL_LOGGING
_this->mMinimalResubscribeDelay = failureInfo.requestedBusyDelay.ValueOr(System::Clock::kZero);
#else
_this->mMinimalResubscribeDelay = System::Clock::kZero;
#endif // CHIP_CONFIG_ENABLE_BUSY_HANDLING_FOR_OPERATIONAL_SESSION_SETUP
_this->Close(failureInfo.error);
}
void ReadClient::OnResubscribeTimerCallback(System::Layer * /* If this starts being used, fix callers that pass nullptr */,
void * apAppState)
{
ReadClient * const _this = static_cast<ReadClient *>(apAppState);
VerifyOrDie(_this != nullptr);
_this->mIsResubscriptionScheduled = false;
CHIP_ERROR err;
ChipLogProgress(DataManagement, "OnResubscribeTimerCallback: ForceCASE = %d", _this->mForceCaseOnNextResub);
_this->mNumRetries++;
bool allowResubscribeOnError = true;
if (!_this->mReadPrepareParams.mSessionHolder ||
!_this->mReadPrepareParams.mSessionHolder->AsSecureSession()->IsActiveSession())
{
// We don't have an active CASE session. We need to go ahead and set
// one up, if we can.
if (_this->EstablishSessionToPeer() == CHIP_NO_ERROR)
{
return;
}
if (_this->mForceCaseOnNextResub)
{
// Caller asked us to force CASE but we have no way to do CASE.
// Just stop trying.
allowResubscribeOnError = false;
}
// No way to send our subscribe request.
err = CHIP_ERROR_INCORRECT_STATE;
ExitNow();
}
err = _this->SendSubscribeRequest(_this->mReadPrepareParams);
exit:
if (err != CHIP_NO_ERROR)
{
//
// Call Close (which should trigger re-subscription again) EXCEPT if we got here because we didn't have a valid
// CASESessionManager pointer when mForceCaseOnNextResub was true.
//
// In that case, don't permit re-subscription to occur.
//
_this->Close(err, allowResubscribeOnError);
}
}
void ReadClient::UpdateDataVersionFilters(const ConcreteDataAttributePath & aPath)
{
for (size_t index = 0; index < mReadPrepareParams.mDataVersionFilterListSize; index++)
{
if (mReadPrepareParams.mpDataVersionFilterList[index].mEndpointId == aPath.mEndpointId &&
mReadPrepareParams.mpDataVersionFilterList[index].mClusterId == aPath.mClusterId)
{
// Now we know the current version for this cluster is aPath.mDataVersion.
mReadPrepareParams.mpDataVersionFilterList[index].mDataVersion = aPath.mDataVersion;
}
}
}
CHIP_ERROR ReadClient::GetMinEventNumber(const ReadPrepareParams & aReadPrepareParams, Optional<EventNumber> & aEventMin)
{
if (aReadPrepareParams.mEventNumber.HasValue())
{
aEventMin = aReadPrepareParams.mEventNumber;
}
else
{
ReturnErrorOnFailure(mpCallback.GetHighestReceivedEventNumber(aEventMin));
if (aEventMin.HasValue())
{
// We want to start with the first event _after_ the last one we received.
aEventMin.SetValue(aEventMin.Value() + 1);
}
}
return CHIP_NO_ERROR;
}
void ReadClient::TriggerResubscribeIfScheduled(const char * reason)
{
if (!mIsResubscriptionScheduled)
{
return;
}
ChipLogDetail(DataManagement, "ReadClient[%p] triggering resubscribe, reason: %s", this, reason);
CancelResubscribeTimer();
OnResubscribeTimerCallback(nullptr, this);
}
Optional<System::Clock::Timeout> ReadClient::GetSubscriptionTimeout()
{
if (!IsSubscriptionType() || !IsSubscriptionActive())
{
return NullOptional;
}
System::Clock::Timeout timeout;
CHIP_ERROR err = ComputeLivenessCheckTimerTimeout(&timeout);
if (err != CHIP_NO_ERROR)
{
return NullOptional;
}
return MakeOptional(timeout);
}
CHIP_ERROR ReadClient::EstablishSessionToPeer()
{
ChipLogProgress(DataManagement, "Trying to establish a CASE session for subscription");
auto * caseSessionManager = InteractionModelEngine::GetInstance()->GetCASESessionManager();
VerifyOrReturnError(caseSessionManager != nullptr, CHIP_ERROR_INCORRECT_STATE);
caseSessionManager->FindOrEstablishSession(mPeer, &mOnConnectedCallback, &mOnConnectionFailureCallback);
return CHIP_NO_ERROR;
}
} // namespace app
} // namespace chip