| /* |
| * |
| * Copyright (c) 2016-2017 Nest Labs, Inc. |
| * 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 implements subscription engine for Weave |
| * Data Management (WDM) profile. |
| * |
| */ |
| |
| #ifndef __STDC_FORMAT_MACROS |
| #define __STDC_FORMAT_MACROS |
| #endif // __STDC_FORMAT_MACROS |
| |
| #include <Weave/Profiles/WeaveProfiles.h> |
| #include <Weave/Profiles/common/CommonProfile.h> |
| #include <Weave/Profiles/data-management/Current/WdmManagedNamespace.h> |
| #include <Weave/Profiles/data-management/DataManagement.h> |
| #include <Weave/Profiles/time/WeaveTime.h> |
| #include <Weave/Support/crypto/WeaveCrypto.h> |
| #include <Weave/Support/WeaveFaultInjection.h> |
| #include <SystemLayer/SystemStats.h> |
| |
| #if WEAVE_CONFIG_ENABLE_RELIABLE_MESSAGING |
| |
| namespace nl { |
| namespace Weave { |
| namespace Profiles { |
| namespace WeaveMakeManagedNamespaceIdentifier(DataManagement, kWeaveManagedNamespaceDesignation_Current) { |
| |
| SubscriptionEngine::SubscriptionEngine() |
| { |
| } |
| |
| void SubscriptionEngine::SetEventCallback(void * const aAppState, const EventCallback aEventCallback) |
| { |
| mAppState = aAppState; |
| mEventCallback = aEventCallback; |
| } |
| |
| void SubscriptionEngine::DefaultEventHandler(EventID aEvent, const InEventParam & aInParam, OutEventParam & aOutParam) |
| { |
| IgnoreUnusedVariable(aInParam); |
| IgnoreUnusedVariable(aOutParam); |
| |
| WeaveLogDetail(DataManagement, "%s event: %d", __func__, aEvent); |
| } |
| |
| WEAVE_ERROR SubscriptionEngine::Init (nl::Weave::WeaveExchangeManager * const apExchangeMgr, void * const aAppState, const EventCallback aEventCallback) |
| { |
| WEAVE_ERROR err = WEAVE_NO_ERROR; |
| |
| mExchangeMgr = apExchangeMgr; |
| mAppState = aAppState; |
| mEventCallback = aEventCallback; |
| mLock = NULL; |
| |
| err = mExchangeMgr->RegisterUnsolicitedMessageHandler(nl::Weave::Profiles::kWeaveProfile_WDM, UnsolicitedMessageHandler, this); |
| SuccessOrExit(err); |
| |
| #if WDM_ENABLE_SUBSCRIPTION_CLIENT |
| for (size_t i = 0; i < kMaxNumCommandObjs; ++i) |
| { |
| mCommandObjs[i].Init(NULL); |
| } |
| |
| for (size_t i = 0; i < kMaxNumSubscriptionClients; ++i) |
| { |
| mClients[i].InitAsFree (); |
| } |
| |
| #endif // WDM_ENABLE_SUBSCRIPTION_CLIENT |
| |
| #if WDM_ENABLE_SUBSCRIPTION_PUBLISHER |
| err = mNotificationEngine.Init(); |
| SuccessOrExit(err); |
| |
| for (size_t i = 0; i < kMaxNumSubscriptionHandlers; ++i) |
| { |
| mHandlers[i].InitAsFree(); |
| } |
| |
| // erase everything |
| DisablePublisher (); |
| |
| #endif // WDM_ENABLE_SUBSCRIPTION_PUBLISHER |
| |
| mNumTraitInfosInPool = 0; |
| |
| exit: |
| WeaveLogFunctError(err); |
| |
| return err; |
| } |
| |
| #if WEAVE_DETAIL_LOGGING |
| void SubscriptionEngine::LogSubscriptionFreed(void) const |
| { |
| // Report number of clients and handlers that are still allocated |
| uint32_t countAllocatedClients = 0; |
| uint32_t countAllocatedHandlers = 0; |
| |
| #if WDM_ENABLE_SUBSCRIPTION_CLIENT |
| for (int i = 0; i < kMaxNumSubscriptionClients; ++i) |
| { |
| if (SubscriptionClient::kState_Free != mClients[i].mCurrentState) |
| { |
| ++countAllocatedClients; |
| } |
| } |
| #endif // #if WDM_ENABLE_SUBSCRIPTION_CLIENT |
| |
| #if WDM_ENABLE_SUBSCRIPTION_PUBLISHER |
| for (int i = 0; i < kMaxNumSubscriptionHandlers; ++i) |
| { |
| if (SubscriptionHandler::kState_Free != mHandlers[i].mCurrentState) |
| { |
| ++countAllocatedHandlers; |
| } |
| } |
| #endif // WDM_ENABLE_SUBSCRIPTION_PUBLISHER |
| |
| WeaveLogDetail(DataManagement, "Allocated clients: %" PRIu32 ". Allocated handlers: %" PRIu32 ".", |
| countAllocatedClients, countAllocatedHandlers); |
| } |
| #endif // #if WEAVE_DETAIL_LOGGING |
| |
| #if WDM_ENABLE_SUBSCRIPTION_CANCEL |
| void SubscriptionEngine::OnCancelRequest (nl::Weave::ExchangeContext *aEC, const nl::Inet::IPPacketInfo *aPktInfo, |
| const nl::Weave::WeaveMessageInfo *aMsgInfo, uint32_t aProfileId, |
| uint8_t aMsgType, PacketBuffer *aPayload) |
| { |
| WEAVE_ERROR err = WEAVE_NO_ERROR; |
| SubscriptionEngine* const pEngine = reinterpret_cast<SubscriptionEngine *>(aEC->AppState); |
| uint64_t SubscriptionId = 0; |
| bool found = false; |
| |
| { |
| nl::Weave::TLV::TLVReader reader; |
| SubscribeCancelRequest::Parser request; |
| |
| reader.Init(aPayload); |
| |
| err = reader.Next(); |
| SuccessOrExit(err); |
| |
| err = request.Init(reader); |
| SuccessOrExit(err); |
| |
| #if WEAVE_CONFIG_DATA_MANAGEMENT_ENABLE_SCHEMA_CHECK |
| err = request.CheckSchemaValidity(); |
| SuccessOrExit(err); |
| #endif // WEAVE_CONFIG_DATA_MANAGEMENT_ENABLE_SCHEMA_CHECK |
| |
| err = request.GetSubscriptionID(&SubscriptionId); |
| SuccessOrExit(err); |
| } |
| |
| #if WDM_ENABLE_SUBSCRIPTION_CLIENT |
| for (size_t i = 0; i < kMaxNumSubscriptionClients; ++i) |
| { |
| if ((SubscriptionClient::kState_SubscriptionEstablished_Idle == pEngine->mClients[i].mCurrentState) || |
| (SubscriptionClient::kState_SubscriptionEstablished_Confirming == pEngine->mClients[i].mCurrentState)) |
| { |
| if (pEngine->mClients[i].mSubscriptionId == SubscriptionId) |
| { |
| pEngine->mClients[i].CancelRequestHandler(aEC, aPktInfo, aMsgInfo, aPayload); |
| found = true; |
| break; |
| } |
| } |
| } |
| #endif // #if WDM_ENABLE_SUBSCRIPTION_CLIENT |
| |
| #if WDM_ENABLE_SUBSCRIPTION_PUBLISHER |
| for (size_t i = 0; i < kMaxNumSubscriptionHandlers; ++i) |
| { |
| if ((pEngine->mHandlers[i].mCurrentState >= SubscriptionHandler::kState_SubscriptionInfoValid_Begin) && |
| (pEngine->mHandlers[i].mCurrentState <= SubscriptionHandler::kState_SubscriptionInfoValid_End)) |
| { |
| // Note that there is no need to compare more than subscription ID, because it must already be unique on publisher side |
| if (pEngine->mHandlers[i].mSubscriptionId == SubscriptionId) |
| { |
| pEngine->mHandlers[i].CancelRequestHandler(aEC, aPktInfo, aMsgInfo, aPayload); |
| found = true; |
| break; |
| } |
| } |
| } |
| #endif // WDM_ENABLE_SUBSCRIPTION_PUBLISHER |
| |
| if (!found) |
| { |
| err = SendStatusReport(aEC, nl::Weave::Profiles::kWeaveProfile_WDM, kStatus_InvalidSubscriptionID); |
| SuccessOrExit(err); |
| } |
| |
| exit: |
| WeaveLogFunctError(err); |
| |
| // aPayload guaranteed to be non-NULL |
| PacketBuffer::Free(aPayload); |
| |
| // aEC guaranteed to be non-NULL |
| aEC->Close(); |
| } |
| #endif // WDM_ENABLE_SUBSCRIPTION_CANCEL |
| |
| #if WDM_ENABLE_SUBSCRIPTION_CLIENT |
| |
| uint16_t SubscriptionEngine::GetClientId (const SubscriptionClient * const apClient) const |
| { |
| return static_cast<uint16_t>(apClient - mClients); |
| } |
| |
| WEAVE_ERROR SubscriptionEngine::NewClient (SubscriptionClient ** const appClient, |
| Binding * const apBinding, |
| void * const apAppState, |
| SubscriptionClient::EventCallback const aEventCallback, |
| const TraitCatalogBase<TraitDataSink>* const apCatalog, |
| const uint32_t aInactivityTimeoutDuringSubscribingMsec) |
| { |
| WEAVE_ERROR err = WEAVE_ERROR_NO_MEMORY; |
| |
| WEAVE_FAULT_INJECT(FaultInjection::kFault_WDM_SubscriptionClientNew, ExitNow()); |
| |
| *appClient = NULL; |
| for (size_t i = 0; i < kMaxNumSubscriptionClients; ++i) |
| { |
| if (SubscriptionClient::kState_Free == mClients[i].mCurrentState) |
| { |
| *appClient = &mClients[i]; |
| err = (*appClient)->Init(apBinding, |
| apAppState, |
| aEventCallback, |
| apCatalog, |
| aInactivityTimeoutDuringSubscribingMsec); |
| |
| if (WEAVE_NO_ERROR != err) |
| { |
| *appClient = NULL; |
| ExitNow(); |
| } |
| SYSTEM_STATS_INCREMENT(nl::Weave::System::Stats::kWDMNext_NumSubscriptionClients); |
| break; |
| } |
| } |
| |
| exit: |
| |
| return err; |
| } |
| |
| /** |
| * Reply to a request with a StatuReport message. |
| * |
| * @param[in] aEC Pointer to the ExchangeContext on which the request was received. |
| * This function does not take ownership of this object. The ExchangeContext |
| * must be closed or aborted by the calling function according to the |
| * WEAVE_ERROR returned. |
| * @param[in] aProfileId The profile to be put in the StatusReport payload. |
| * @param[in] aStatusCode The status code to be put in the StatusReport payload; must refer to the |
| * profile passed in aProfileId, but this function does not enforce this |
| * condition. |
| * |
| * @return WEAVE_NO_ERROR in case of success. |
| * WEAVE_NO_MEMORY if no pbufs are available. |
| * Any other WEAVE_ERROR code returned by ExchangeContext::SendMessage |
| */ |
| WEAVE_ERROR SubscriptionEngine::SendStatusReport (nl::Weave::ExchangeContext *aEC, |
| uint32_t aProfileId, |
| uint16_t aStatusCode) |
| { |
| WEAVE_ERROR err = WEAVE_NO_ERROR; |
| |
| err = nl::Weave::WeaveServerBase::SendStatusReport(aEC, aProfileId, aStatusCode, WEAVE_NO_ERROR, |
| aEC->HasPeerRequestedAck() ? nl::Weave::ExchangeContext::kSendFlag_RequestAck : 0); |
| WeaveLogFunctError(err); |
| |
| return err; |
| } |
| |
| /** |
| * Unsolicited message handler for all WDM messages. |
| * This function is a @ref ExchangeContext::MessageReceiveFunct. |
| */ |
| void SubscriptionEngine::UnsolicitedMessageHandler (nl::Weave::ExchangeContext *aEC, const nl::Inet::IPPacketInfo *aPktInfo, |
| const nl::Weave::WeaveMessageInfo *aMsgInfo, uint32_t aProfileId, |
| uint8_t aMsgType, PacketBuffer *aPayload) |
| { |
| nl::Weave::ExchangeContext::MessageReceiveFunct func = OnUnknownMsgType; |
| |
| switch (aMsgType) |
| { |
| #if WDM_ENABLE_SUBSCRIPTION_CLIENT |
| case kMsgType_NotificationRequest: |
| func = OnNotificationRequest; |
| break; |
| #endif // WDM_ENABLE_SUBSCRIPTION_CLIENT |
| |
| #if WDM_ENABLE_SUBSCRIPTION_PUBLISHER |
| |
| case kMsgType_SubscribeRequest: |
| func = OnSubscribeRequest; |
| break; |
| |
| case kMsgType_SubscribeConfirmRequest: |
| func = OnSubscribeConfirmRequest; |
| break; |
| |
| case kMsgType_CustomCommandRequest: |
| func = OnCustomCommandRequest; |
| break; |
| |
| #endif // WDM_ENABLE_SUBSCRIPTION_PUBLISHER |
| |
| #if WDM_ENABLE_SUBSCRIPTION_CANCEL |
| case kMsgType_SubscribeCancelRequest: |
| func = OnCancelRequest; |
| break; |
| #endif // WDM_ENABLE_SUBSCRIPTION_CANCEL |
| |
| default: |
| break; |
| } |
| |
| func(aEC, aPktInfo, aMsgInfo, aProfileId, aMsgType, aPayload); |
| } |
| |
| /** |
| * Unsolicited message handler for unsupported WDM messages. |
| * This function is a @ref ExchangeContext::MessageReceiveFunct. |
| */ |
| void SubscriptionEngine::OnUnknownMsgType (nl::Weave::ExchangeContext *aEC, const nl::Inet::IPPacketInfo *aPktInfo, |
| const nl::Weave::WeaveMessageInfo *aMsgInfo, uint32_t aProfileId, |
| uint8_t aMsgType, PacketBuffer *aPayload) |
| { |
| WEAVE_ERROR err = WEAVE_NO_ERROR; |
| |
| PacketBuffer::Free(aPayload); |
| aPayload = NULL; |
| |
| WeaveLogDetail(DataManagement, "Msg type %" PRIu8 " not supported", aMsgType); |
| |
| err = SendStatusReport(aEC, nl::Weave::Profiles::kWeaveProfile_Common, nl::Weave::Profiles::Common::kStatus_UnsupportedMessage); |
| SuccessOrExit(err); |
| |
| aEC->Close(); |
| aEC = NULL; |
| |
| exit: |
| WeaveLogFunctError(err); |
| |
| if (NULL != aEC) |
| { |
| aEC->Abort(); |
| aEC = NULL; |
| } |
| } |
| |
| void SubscriptionEngine::OnNotificationRequest (nl::Weave::ExchangeContext *aEC, const nl::Inet::IPPacketInfo *aPktInfo, |
| const nl::Weave::WeaveMessageInfo *aMsgInfo, uint32_t aProfileId, |
| uint8_t aMsgType, PacketBuffer *aPayload) |
| { |
| WEAVE_ERROR err = WEAVE_NO_ERROR; |
| SubscriptionEngine* const pEngine = reinterpret_cast<SubscriptionEngine *>(aEC->AppState); |
| uint64_t SubscriptionId = 0; |
| |
| { |
| nl::Weave::TLV::TLVReader reader; |
| NotificationRequest::Parser notify; |
| |
| reader.Init(aPayload); |
| |
| err = reader.Next(); |
| SuccessOrExit(err); |
| |
| err = notify.Init(reader); |
| SuccessOrExit(err); |
| |
| // Note that it is okay to bail out, without any response, if the message doesn't even have a subscription ID in it |
| err = notify.GetSubscriptionID(&SubscriptionId); |
| SuccessOrExit(err); |
| |
| WEAVE_FAULT_INJECT(FaultInjection::kFault_WDM_BadSubscriptionId, SubscriptionId += 1); |
| } |
| |
| for (size_t i = 0; i < kMaxNumSubscriptionClients; ++i) |
| { |
| if ((SubscriptionClient::kState_SubscriptionEstablished_Idle == pEngine->mClients[i].mCurrentState) || |
| (SubscriptionClient::kState_SubscriptionEstablished_Confirming == pEngine->mClients[i].mCurrentState)) |
| { |
| if (pEngine->mClients[i].mBinding->IsAuthenticMessageFromPeer(aMsgInfo) && |
| pEngine->mClients[i].mSubscriptionId == SubscriptionId) |
| { |
| pEngine->mClients[i].NotificationRequestHandler(aEC, aPktInfo, aMsgInfo, aPayload); |
| aPayload = NULL; |
| aEC = NULL; |
| ExitNow(); |
| } |
| } |
| } |
| |
| WeaveLogDetail(DataManagement, "%s: couldn't find matching client. Subscription ID: 0x%" PRIX64, __func__, SubscriptionId); |
| |
| err = SendStatusReport(aEC, nl::Weave::Profiles::kWeaveProfile_WDM, kStatus_InvalidSubscriptionID); |
| SuccessOrExit(err); |
| |
| exit: |
| WeaveLogFunctError(err); |
| |
| if (NULL != aPayload) |
| { |
| PacketBuffer::Free(aPayload); |
| aPayload = NULL; |
| } |
| |
| if (NULL != aEC) |
| { |
| aEC->Abort(); |
| aEC = NULL; |
| } |
| } |
| |
| SubscriptionClient * SubscriptionEngine::FindClient(const uint64_t aPeerNodeId, const uint64_t aSubscriptionId) |
| { |
| SubscriptionClient * result = NULL; |
| |
| for (size_t i = 0; i < kMaxNumSubscriptionClients; ++i) |
| { |
| if ((mClients[i].mCurrentState >= SubscriptionClient::kState_Subscribing_IdAssigned) && |
| (mClients[i].mCurrentState <= SubscriptionClient::kState_SubscriptionEstablished_Confirming)) |
| { |
| if ((aPeerNodeId == mClients[i].mBinding->GetPeerNodeId()) && (mClients[i].mSubscriptionId == aSubscriptionId)) |
| { |
| result = &mClients[i]; |
| break; |
| } |
| } |
| } |
| |
| return result; |
| } |
| |
| bool SubscriptionEngine::UpdateClientLiveness(const uint64_t aPeerNodeId, const uint64_t aSubscriptionId, const bool aKill) |
| { |
| WEAVE_ERROR err = WEAVE_NO_ERROR; |
| bool found = false; |
| SubscriptionClient * pClient = FindClient(aPeerNodeId, aSubscriptionId); |
| |
| if (NULL != pClient) |
| { |
| found = true; |
| |
| if (aKill) |
| { |
| err = WEAVE_ERROR_TRANSACTION_CANCELED; |
| } |
| else |
| { |
| WeaveLogDetail(DataManagement, "Client[%d] [%5.5s] liveness confirmed", |
| GetClientId(pClient), pClient->GetStateStr()); |
| |
| // ignore incorrect state error, otherwise, let it flow through |
| err = pClient->RefreshTimer(); |
| if (WEAVE_ERROR_INCORRECT_STATE == err) |
| { |
| err = WEAVE_NO_ERROR; |
| |
| WeaveLogDetail(DataManagement, "Client[%d] [%5.5s] liveness confirmation failed, ignore", |
| GetClientId(pClient), pClient->GetStateStr()); |
| } |
| } |
| |
| if (WEAVE_NO_ERROR != err) |
| { |
| WeaveLogDetail(DataManagement, "Client[%d] [%5.5s] bound mutual subscription is going away", |
| GetClientId(pClient), pClient->GetStateStr()); |
| |
| pClient->HandleSubscriptionTerminated(pClient->IsRetryEnabled(), err, NULL); |
| } |
| } |
| |
| return found; |
| } |
| |
| |
| #endif // WDM_ENABLE_SUBSCRIPTION_CLIENT |
| |
| #if WDM_ENABLE_SUBSCRIPTION_PUBLISHER |
| |
| uint16_t SubscriptionEngine::GetHandlerId (const SubscriptionHandler * const apHandler) const |
| { |
| return static_cast<uint16_t>(apHandler - mHandlers); |
| } |
| |
| uint16_t SubscriptionEngine::GetCommandObjId (const Command * const apHandle) const |
| { |
| return static_cast<uint16_t>(apHandle - mCommandObjs); |
| } |
| |
| bool SubscriptionEngine::UpdateHandlerLiveness(const uint64_t aPeerNodeId, const uint64_t aSubscriptionId, const bool aKill) |
| { |
| WEAVE_ERROR err = WEAVE_NO_ERROR; |
| bool found = false; |
| SubscriptionHandler * pHandler = FindHandler(aPeerNodeId, aSubscriptionId); |
| if (NULL != pHandler) |
| { |
| found = true; |
| |
| if (aKill) |
| { |
| err = WEAVE_ERROR_TRANSACTION_CANCELED; |
| } |
| else |
| { |
| WeaveLogDetail(DataManagement, "Handler[%d] [%5.5s] liveness confirmed", |
| GetHandlerId(pHandler), pHandler->GetStateStr()); |
| |
| // ignore incorrect state error, otherwise, let it flow through |
| err = pHandler->RefreshTimer(); |
| if (WEAVE_ERROR_INCORRECT_STATE == err) |
| { |
| err = WEAVE_NO_ERROR; |
| |
| WeaveLogDetail(DataManagement, "Handler[%d] [%5.5s] liveness confirmation failed, ignore", |
| GetHandlerId(pHandler), pHandler->GetStateStr()); |
| } |
| } |
| |
| if (WEAVE_NO_ERROR != err) |
| { |
| WeaveLogDetail(DataManagement, "Handler[%d] [%5.5s] bound mutual subscription is going away", |
| GetHandlerId(pHandler), pHandler->GetStateStr()); |
| |
| pHandler->HandleSubscriptionTerminated(err, NULL); |
| } |
| } |
| |
| return found; |
| } |
| |
| SubscriptionHandler * SubscriptionEngine::FindHandler(const uint64_t aPeerNodeId, const uint64_t aSubscriptionId) |
| { |
| SubscriptionHandler * result = NULL; |
| |
| for (size_t i = 0; i < kMaxNumSubscriptionHandlers; ++i) |
| { |
| if ((mHandlers[i].mCurrentState >= SubscriptionHandler::kState_SubscriptionInfoValid_Begin) && |
| (mHandlers[i].mCurrentState <= SubscriptionHandler::kState_SubscriptionInfoValid_End)) |
| { |
| if ((aPeerNodeId == mHandlers[i].mBinding->GetPeerNodeId()) && (aSubscriptionId == mHandlers[i].mSubscriptionId)) |
| { |
| result = &mHandlers[i]; |
| break; |
| } |
| } |
| } |
| |
| return result; |
| } |
| |
| WEAVE_ERROR SubscriptionEngine::GetMinEventLogPosition(size_t &outLogPosition) const |
| { |
| WEAVE_ERROR err = WEAVE_NO_ERROR; |
| |
| for (size_t subIdx = 0; subIdx < kMaxNumSubscriptionHandlers; ++subIdx) |
| { |
| const SubscriptionHandler *subHandler = &(mHandlers[subIdx]); |
| if (subHandler->mCurrentState == SubscriptionHandler::kState_Free) |
| { |
| continue; |
| } |
| |
| if (subHandler->mBytesOffloaded < outLogPosition) |
| { |
| outLogPosition = subHandler->mBytesOffloaded; |
| } |
| } |
| |
| return err; |
| } |
| |
| void SubscriptionEngine::ReclaimTraitInfo(SubscriptionHandler * const aHandlerToBeReclaimed) |
| { |
| SubscriptionHandler::TraitInstanceInfo* const traitInfoList = aHandlerToBeReclaimed->mTraitInstanceList; |
| const uint16_t numTraitInstances = aHandlerToBeReclaimed->mNumTraitInstances; |
| size_t numTraitInstancesToBeAffected; |
| |
| aHandlerToBeReclaimed->mTraitInstanceList = NULL; |
| aHandlerToBeReclaimed->mNumTraitInstances = 0; |
| |
| if (!numTraitInstances) |
| { |
| WeaveLogDetail(DataManagement, "No trait instances allocated for this subscription"); |
| ExitNow(); |
| } |
| |
| // make sure everything is still sane |
| WeaveLogIfFalse(traitInfoList >= mTraitInfoPool); |
| WeaveLogIfFalse(numTraitInstances <= mNumTraitInfosInPool); |
| |
| // mPathGroupPool + kMaxNumPathGroups is a pointer which points to the last+1byte of this array |
| // traitInfoList is a pointer to the first trait instance to be released |
| // the result of subtraction is the number of trait instances from traitInfoList to the end of this array |
| numTraitInstancesToBeAffected = (mTraitInfoPool + mNumTraitInfosInPool) - traitInfoList; |
| |
| // Shrink the traitInfosInPool by the number of trait instances in this subscription. |
| mNumTraitInfosInPool -= numTraitInstances; |
| SYSTEM_STATS_DECREMENT_BY_N(nl::Weave::System::Stats::kWDMNext_NumTraits, numTraitInstances); |
| |
| if (numTraitInstances == numTraitInstancesToBeAffected) |
| { |
| WeaveLogDetail(DataManagement, "Releasing the last block of trait instances"); |
| ExitNow(); |
| } |
| |
| WeaveLogDetail(DataManagement, "Moving %u trait instances forward", static_cast<unsigned int>(numTraitInstancesToBeAffected - numTraitInstances)); |
| |
| memmove(traitInfoList, traitInfoList + numTraitInstances, |
| sizeof(SubscriptionHandler::TraitInstanceInfo) * |
| (numTraitInstancesToBeAffected - numTraitInstances)); |
| |
| for (size_t i = 0; i < kMaxNumSubscriptionHandlers; ++i) |
| { |
| SubscriptionHandler * const pHandler = mHandlers + i; |
| |
| if ((aHandlerToBeReclaimed != pHandler) && (pHandler->mTraitInstanceList > traitInfoList)) |
| { |
| pHandler->mTraitInstanceList -= numTraitInstances; |
| } |
| } |
| |
| exit: |
| WeaveLogDetail(DataManagement, "Number of allocated trait instances: %u", mNumTraitInfosInPool); |
| } |
| |
| WEAVE_ERROR SubscriptionEngine::EnablePublisher(IWeavePublisherLock *aLock, TraitCatalogBase<TraitDataSource>* const aPublisherCatalog) |
| { |
| // force abandon all subscription first, so we can have a clean slate |
| DisablePublisher(); |
| |
| mLock = aLock; |
| |
| // replace catalog |
| mPublisherCatalog = aPublisherCatalog; |
| |
| mIsPublisherEnabled = true; |
| |
| mNextHandlerToNotify = 0; |
| |
| return WEAVE_NO_ERROR; |
| } |
| |
| WEAVE_ERROR SubscriptionEngine::Lock() |
| { |
| if (mLock) { |
| return mLock->Lock(); |
| } |
| |
| return WEAVE_NO_ERROR; |
| } |
| |
| WEAVE_ERROR SubscriptionEngine::Unlock() |
| { |
| if (mLock) { |
| return mLock->Unlock(); |
| } |
| |
| return WEAVE_NO_ERROR; |
| } |
| |
| void SubscriptionEngine::DisablePublisher () |
| { |
| mIsPublisherEnabled = false; |
| mPublisherCatalog = NULL; |
| |
| for (size_t i = 0; i < kMaxNumSubscriptionHandlers; ++i) |
| { |
| switch (mHandlers[i].mCurrentState) |
| { |
| case SubscriptionHandler::kState_Free: |
| case SubscriptionHandler::kState_Aborted: |
| break; |
| default: |
| mHandlers[i].AbortSubscription(); |
| } |
| } |
| |
| // Note that the command objects are not closed when publisher is disabled. |
| // This is because the processing flow of commands are not directly linked |
| // with subscriptions. |
| } |
| |
| WEAVE_ERROR SubscriptionEngine::NewSubscriptionHandler(SubscriptionHandler **subHandler) |
| { |
| WEAVE_ERROR err = WEAVE_ERROR_NO_MEMORY; |
| |
| *subHandler = NULL; |
| |
| WEAVE_FAULT_INJECT(FaultInjection::kFault_WDM_SubscriptionHandlerNew, ExitNow()); |
| |
| for (size_t i = 0; i < kMaxNumSubscriptionHandlers; ++i) |
| { |
| if (SubscriptionHandler::kState_Free == mHandlers[i].mCurrentState) |
| { |
| WeaveLogIfFalse(0 == mHandlers[i].mRefCount); |
| *subHandler = &mHandlers[i]; |
| err = WEAVE_NO_ERROR; |
| |
| SYSTEM_STATS_INCREMENT(nl::Weave::System::Stats::kWDMNext_NumSubscriptionHandlers); |
| |
| break; |
| } |
| } |
| |
| ExitNow(); // silence warnings about unused labels. |
| exit: |
| return err; |
| } |
| |
| void SubscriptionEngine::OnSubscribeRequest (nl::Weave::ExchangeContext *aEC, const nl::Inet::IPPacketInfo *aPktInfo, |
| const nl::Weave::WeaveMessageInfo *aMsgInfo, uint32_t aProfileId, |
| uint8_t aMsgType, PacketBuffer *aPayload) |
| { |
| WEAVE_ERROR err = WEAVE_NO_ERROR; |
| SubscriptionEngine* const pEngine = reinterpret_cast<SubscriptionEngine *>(aEC->AppState); |
| SubscriptionHandler* handler = NULL; |
| uint32_t reasonProfileId = nl::Weave::Profiles::kWeaveProfile_Common; |
| uint16_t reasonStatusCode = nl::Weave::Profiles::Common::kStatus_InternalServerProblem; |
| InEventParam inParam; |
| OutEventParam outParam; |
| Binding* binding; |
| uint64_t subscriptionId = 0; |
| |
| // Note that there is no event callback nor App state assigned to this newly allocated binding |
| // We will need to assign a callback handler when binding actually generates useful events |
| binding = pEngine->mExchangeMgr->NewBinding(); |
| if (NULL == binding) |
| { |
| // log as error as it might be difficult to estimate how many bindings are needed on a system |
| WeaveLogError(DataManagement, "%s: Out of Binding", __func__); |
| ExitNow(err = WEAVE_ERROR_NO_MEMORY); |
| } |
| |
| err = binding->BeginConfiguration() |
| .ConfigureFromMessage(aMsgInfo, aPktInfo) |
| .PrepareBinding(); |
| SuccessOrExit(err); |
| |
| // If the peer requested an ACK, we need to ensure that the exchange context will automatically |
| // request an ACK when we send messages out on this exchange. |
| // |
| // In future exchanges that we initiate to this peer, the binding will automatically vend out exchange |
| // contexts with this auto-ack bit set due to the binding configuration that happens in the line above. |
| if (aMsgInfo->Flags & kWeaveMessageFlag_PeerRequestedAck) |
| { |
| aEC->SetAutoRequestAck(true); |
| } |
| |
| if (pEngine->mIsPublisherEnabled && (NULL != pEngine->mEventCallback)) |
| { |
| outParam.mIncomingSubscribeRequest.mAutoClosePriorSubscription = true; |
| outParam.mIncomingSubscribeRequest.mRejectRequest = false; |
| outParam.mIncomingSubscribeRequest.mpReasonProfileId = &reasonProfileId; |
| outParam.mIncomingSubscribeRequest.mpReasonStatusCode = &reasonStatusCode; |
| |
| inParam.mIncomingSubscribeRequest.mEC = aEC; |
| inParam.mIncomingSubscribeRequest.mPktInfo = aPktInfo; |
| inParam.mIncomingSubscribeRequest.mMsgInfo = aMsgInfo; |
| inParam.mIncomingSubscribeRequest.mPayload = aPayload; |
| inParam.mIncomingSubscribeRequest.mBinding = binding; |
| |
| // note the binding is exposed to app layer for configuration here, and again later after |
| // the request is fully parsed |
| pEngine->mEventCallback(pEngine->mAppState, kEvent_OnIncomingSubscribeRequest, inParam, outParam); |
| |
| // Make sure messages sent through this EC are sent with proper re-transmission/timeouts settings |
| // This is mainly for rejections, as the EC would be configured again in SubscriptionHandler::AcceptSubscribeRequest |
| |
| err = binding->AdjustResponseTimeout(aEC); |
| SuccessOrExit(err); |
| } |
| else |
| { |
| ExitNow(err = WEAVE_ERROR_NO_MESSAGE_HANDLER); |
| } |
| |
| if (outParam.mIncomingSubscribeRequest.mRejectRequest) |
| { |
| // reject this request (without touching existing subscriptions) |
| ExitNow(err = WEAVE_ERROR_TRANSACTION_CANCELED); |
| } |
| else |
| { |
| if (outParam.mIncomingSubscribeRequest.mAutoClosePriorSubscription) |
| { |
| // if not rejected, default behavior is to abort any prior communication with this node id |
| for (size_t i = 0; i < kMaxNumSubscriptionHandlers; ++i) |
| { |
| if ((pEngine->mHandlers[i].mCurrentState >= SubscriptionHandler::kState_SubscriptionInfoValid_Begin) && |
| (pEngine->mHandlers[i].mCurrentState <= SubscriptionHandler::kState_SubscriptionInfoValid_End)) |
| { |
| uint64_t nodeId = pEngine->mHandlers[i].GetPeerNodeId(); |
| |
| if (nodeId == aEC->PeerNodeId) |
| { |
| pEngine->mHandlers[i].HandleSubscriptionTerminated(err, NULL); |
| } |
| } |
| } |
| } |
| |
| err = nl::Weave::Platform::Security::GetSecureRandomData((uint8_t *)&subscriptionId, sizeof(subscriptionId)); |
| SuccessOrExit(err); |
| |
| err = pEngine->NewSubscriptionHandler(&handler); |
| if (err != WEAVE_NO_ERROR) |
| { |
| // try to give slightly more detail on the issue for this potentially common problem |
| reasonStatusCode = (err == WEAVE_ERROR_NO_MEMORY ? |
| nl::Weave::Profiles::Common::kStatus_OutOfMemory : |
| nl::Weave::Profiles::Common::kStatus_InternalServerProblem); |
| |
| ExitNow(); |
| } |
| else |
| { |
| handler->mAppState = outParam.mIncomingSubscribeRequest.mHandlerAppState; |
| handler->mEventCallback = outParam.mIncomingSubscribeRequest.mHandlerEventCallback; |
| uint32_t maxSize = WDM_MAX_NOTIFICATION_SIZE; |
| |
| WEAVE_FAULT_INJECT_WITH_ARGS(FaultInjection::kFault_WDM_NotificationSize, |
| // Code executed with the Manager's lock: |
| if (numFaultArgs > 0) |
| { |
| maxSize = static_cast<uint32_t>(faultArgs[0]); |
| } |
| else |
| { |
| maxSize = WDM_MAX_NOTIFICATION_SIZE/2; |
| } |
| , |
| // Code executed withouth the Manager's lock: |
| WeaveLogDetail(DataManagement, "Handler[%d] Payload size set to %d", pEngine->GetHandlerId(handler), maxSize) |
| ); |
| |
| handler->SetMaxNotificationSize(maxSize); |
| |
| handler->InitWithIncomingRequest(binding, subscriptionId, aEC, aPktInfo, aMsgInfo, aPayload); |
| aEC = NULL; |
| aPayload = NULL; |
| } |
| } |
| |
| exit: |
| WeaveLogFunctError(err); |
| |
| if (NULL != aPayload) |
| { |
| PacketBuffer::Free(aPayload); |
| aPayload = NULL; |
| } |
| |
| if (NULL != aEC) |
| { |
| err = SendStatusReport(aEC, reasonProfileId, reasonStatusCode); |
| WeaveLogFunctError(err); |
| |
| aEC->Close(); |
| aEC = NULL; |
| } |
| |
| if (NULL != binding) |
| { |
| binding->Release(); |
| } |
| } |
| |
| void SubscriptionEngine::OnSubscribeConfirmRequest (nl::Weave::ExchangeContext *aEC, const nl::Inet::IPPacketInfo *aPktInfo, |
| const nl::Weave::WeaveMessageInfo *aMsgInfo, uint32_t aProfileId, |
| uint8_t aMsgType, PacketBuffer *aPayload) |
| { |
| WEAVE_ERROR err = WEAVE_NO_ERROR; |
| SubscriptionEngine* const pEngine = reinterpret_cast<SubscriptionEngine *>(aEC->AppState); |
| uint32_t reasonProfileId = nl::Weave::Profiles::kWeaveProfile_Common; |
| uint16_t reasonStatusCode = nl::Weave::Profiles::Common::kStatus_InternalServerProblem; |
| uint64_t subscriptionId; |
| |
| { |
| nl::Weave::TLV::TLVReader reader; |
| SubscribeConfirmRequest::Parser request; |
| |
| reader.Init(aPayload); |
| |
| err = reader.Next(); |
| SuccessOrExit(err); |
| |
| err = request.Init(reader); |
| SuccessOrExit(err); |
| |
| err = request.GetSubscriptionID(&subscriptionId); |
| SuccessOrExit(err); |
| } |
| |
| if (pEngine->mIsPublisherEnabled) |
| { |
| // find a matching subscription |
| bool found = false; |
| |
| #if WDM_ENABLE_SUBSCRIPTION_CLIENT |
| if (pEngine->UpdateClientLiveness(aEC->PeerNodeId, subscriptionId)) |
| { |
| found = true; |
| } |
| #endif // WDM_ENABLE_SUBSCRIPTION_CLIENT |
| |
| #if WDM_ENABLE_SUBSCRIPTION_PUBLISHER |
| if (pEngine->UpdateHandlerLiveness(aEC->PeerNodeId, subscriptionId)) |
| { |
| found = true; |
| } |
| #endif // WDM_ENABLE_SUBSCRIPTION_PUBLISHER |
| |
| if (found) |
| { |
| reasonStatusCode = nl::Weave::Profiles::Common::kStatus_Success; |
| } |
| else |
| { |
| reasonProfileId = nl::Weave::Profiles::kWeaveProfile_WDM; |
| reasonStatusCode = kStatus_InvalidSubscriptionID; |
| } |
| } |
| else |
| { |
| reasonStatusCode = nl::Weave::Profiles::Common::kStatus_Busy; |
| } |
| |
| { |
| err = SendStatusReport(aEC, reasonProfileId, reasonStatusCode); |
| SuccessOrExit(err); |
| } |
| |
| exit: |
| WeaveLogFunctError(err); |
| |
| // aPayload is guaranteed to be non-NULL |
| PacketBuffer::Free(aPayload); |
| |
| // aEC is guaranteed to be non-NULL. |
| aEC->Abort(); |
| } |
| |
| #if WDM_PUBLISHER_ENABLE_CUSTOM_COMMANDS |
| void SubscriptionEngine::OnCustomCommandRequest (nl::Weave::ExchangeContext *aEC, const nl::Inet::IPPacketInfo *aPktInfo, |
| const nl::Weave::WeaveMessageInfo *aMsgInfo, uint32_t aProfileId, |
| uint8_t aMsgType, PacketBuffer *aPayload) |
| { |
| WEAVE_ERROR err = WEAVE_NO_ERROR; |
| SubscriptionEngine * const pEngine = reinterpret_cast<SubscriptionEngine *>(aEC->AppState); |
| Command * command = NULL; |
| uint32_t statusReportProfile = nl::Weave::Profiles::kWeaveProfile_WDM; |
| uint16_t statusReportCode = nl::Weave::Profiles::DataManagement::kStatus_InvalidPath; |
| |
| // WRM, in general, only works with WRM. Reject anything not requesting ACK immediately |
| // and do not propagate this request downstream. |
| VerifyOrExit(aEC->HasPeerRequestedAck(), err = WEAVE_ERROR_INVALID_TRANSFER_MODE); |
| |
| for (size_t i = 0; i < kMaxNumCommandObjs; ++i) |
| { |
| if (pEngine->mCommandObjs[i].IsFree()) |
| { |
| SYSTEM_STATS_INCREMENT(nl::Weave::System::Stats::kWDMNext_NumCommands); |
| command = &(pEngine->mCommandObjs[i]); |
| command->Init(aEC); |
| aEC = NULL; |
| break; |
| } |
| } |
| VerifyOrExit(NULL != command, err = WEAVE_ERROR_NO_MEMORY); |
| |
| if (!pEngine->mIsPublisherEnabled) |
| { |
| // Has to be a publisher to be processing a command request |
| statusReportProfile = nl::Weave::Profiles::kWeaveProfile_Common; |
| statusReportCode = nl::Weave::Profiles::Common::kStatus_UnsupportedMessage; |
| ExitNow(err = WEAVE_ERROR_INVALID_MESSAGE_TYPE); |
| } |
| |
| { |
| nl::Weave::TLV::TLVReader reader; |
| TraitDataSource *dataSource = NULL; |
| bool isVersionValid = false; |
| bool isExpiryTimeValid = false; |
| uint64_t commandType; |
| uint64_t mustBeVersion; |
| int64_t expiryTimeMicroSecond; |
| |
| reader.Init(aPayload); |
| |
| err = reader.Next(); |
| SuccessOrExit(err); |
| |
| { |
| CustomCommandRequest::Parser request; |
| TraitDataHandle traitDataHandle; |
| nl::Weave::TLV::TLVReader pathReader; |
| SchemaVersionRange requestedSchemaVersion, computedVersionIntersection; |
| |
| err = request.Init(reader); |
| SuccessOrExit(err); |
| |
| #if WEAVE_CONFIG_DATA_MANAGEMENT_ENABLE_SCHEMA_CHECK |
| err = request.CheckSchemaValidity(); |
| SuccessOrExit(err); |
| #endif // WEAVE_CONFIG_DATA_MANAGEMENT_ENABLE_SCHEMA_CHECK |
| |
| err = request.GetReaderOnPath(&pathReader); |
| SuccessOrExit(err); |
| |
| err = pEngine->mPublisherCatalog->AddressToHandle(pathReader, traitDataHandle, requestedSchemaVersion); |
| SuccessOrExit(err); |
| |
| err = SubscriptionEngine::GetInstance()->mPublisherCatalog->Locate(traitDataHandle, &dataSource); |
| SuccessOrExit(err); |
| |
| if (!dataSource->GetSchemaEngine()->GetVersionIntersection(requestedSchemaVersion, computedVersionIntersection)) { |
| WeaveLogDetail(DataManagement, "Mismatch in requested version on handle %u (requested: %u, %u)", traitDataHandle, requestedSchemaVersion.mMaxVersion, requestedSchemaVersion.mMinVersion); |
| |
| statusReportProfile = nl::Weave::Profiles::kWeaveProfile_WDM; |
| statusReportCode = kStatus_IncompatibleDataSchemaVersion; |
| ExitNow(err = WEAVE_ERROR_INCOMPATIBLE_SCHEMA_VERSION); |
| } |
| |
| err = request.GetCommandType(&commandType); |
| SuccessOrExit(err); |
| |
| err = request.GetExpiryTimeMicroSecond(&expiryTimeMicroSecond); |
| if (WEAVE_NO_ERROR == err) |
| { |
| isExpiryTimeValid = true; |
| } |
| else if (WEAVE_END_OF_TLV == err) |
| { |
| err = WEAVE_NO_ERROR; |
| } |
| else |
| { |
| ExitNow(); |
| } |
| |
| err = request.GetMustBeVersion(&mustBeVersion); |
| if (WEAVE_NO_ERROR == err) |
| { |
| isVersionValid = true; |
| } |
| else if (WEAVE_END_OF_TLV == err) |
| { |
| err = WEAVE_NO_ERROR; |
| } |
| else |
| { |
| ExitNow(); |
| } |
| |
| err = request.GetReaderOnArgument(&reader); |
| SuccessOrExit(err); |
| } |
| |
| #if WDM_ENFORCE_EXPIRY_TIME |
| if (isExpiryTimeValid) |
| { |
| nl::Weave::Profiles::Time::timesync_t now_usec; |
| err = nl::Weave::Platform::Time::GetSystemTime(&now_usec); |
| if (WEAVE_ERROR_UNSUPPORTED_CLOCK == err) |
| { |
| statusReportCode = nl::Weave::Profiles::DataManagement::kStatus_ExpiryTimeNotSupported; |
| ExitNow(); |
| } |
| else if (WEAVE_ERROR_TIME_NOT_SYNCED_YET == err) |
| { |
| statusReportCode = nl::Weave::Profiles::DataManagement::kStatus_NotTimeSyncedYet; |
| ExitNow(); |
| } |
| else if (now_usec >= expiryTimeMicroSecond) |
| { |
| statusReportCode = nl::Weave::Profiles::DataManagement::kStatus_RequestExpiredInTime; |
| ExitNow(); |
| } |
| WeaveLogDetail(DataManagement, "Command ExpiryTime 0x%" PRIX64 ", now: 0x% " PRIX64 " ", expiryTimeMicroSecond, now_usec); |
| } |
| #endif // WDM_ENFORCE_EXPIRY_TIME |
| |
| if (isVersionValid) |
| { |
| uint64_t currentVersion = dataSource->GetVersion(); |
| |
| if (mustBeVersion != currentVersion) |
| { |
| WeaveLogDetail(DataManagement, "Version required 0x%" PRIX64 ", current: 0x% " PRIX64 " ", mustBeVersion, currentVersion); |
| statusReportCode = nl::Weave::Profiles::DataManagement::kStatus_VersionMismatch; |
| ExitNow(); |
| } |
| } |
| |
| // Note we cannot just use pathReader at here because the TDM related functions |
| // generally assume they can move the reader at their will. |
| // Note that callee is supposed to cache whatever is useful in the TLV stream into its own memory |
| // when this callback returns, we'd destroy the TLV object |
| dataSource->OnCustomCommand(command, aMsgInfo, aPayload, commandType, |
| isExpiryTimeValid, expiryTimeMicroSecond, isVersionValid, mustBeVersion, |
| reader); |
| command = NULL; |
| aPayload = NULL; |
| } |
| |
| exit: |
| WeaveLogFunctError(err); |
| |
| if (NULL != aPayload) |
| { |
| PacketBuffer::Free(aPayload); |
| aPayload = NULL; |
| } |
| |
| // Note that when dispatched == true, ownership of aEC is already passed on to OnCustomCommand, and hence set to NULL |
| if (NULL != command) |
| { |
| err = command->SendError(statusReportProfile, statusReportCode, err); |
| WeaveLogFunctError(err); |
| } |
| |
| if (NULL != aEC) |
| { |
| aEC->Close(); |
| aEC = NULL; |
| } |
| } |
| #endif // WDM_PUBLISHER_ENABLE_CUSTOM_COMMANDS |
| |
| #endif // #if WDM_ENABLE_SUBSCRIPTION_PUBLISHER |
| |
| }; // WeaveMakeManagedNamespaceIdentifier(DataManagement, kWeaveManagedNamespaceDesignation_Current) |
| }; // Profiles |
| }; // Weave |
| }; // nl |
| |
| #endif // WEAVE_CONFIG_ENABLE_RELIABLE_MESSAGING |
