/*
 *
 *    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 notification handler for Weave
 *      Data Management (WDM) profile.
 *
 */

// __STDC_FORMAT_MACROS must be defined for PRIX64 to be defined for pre-C++11 clib
#ifndef __STDC_FORMAT_MACROS
#define __STDC_FORMAT_MACROS
#endif // __STDC_FORMAT_MACROS

// __STDC_LIMIT_MACROS must be defined for UINT8_MAX and INT32_MAX to be defined for pre-C++11 clib
#ifndef __STDC_LIMIT_MACROS
#define __STDC_LIMIT_MACROS
#endif // __STDC_LIMIT_MACROS

// __STDC_CONSTANT_MACROS must be defined for INT64_C and UINT64_C to be defined for pre-C++11 clib
#ifndef __STDC_CONSTANT_MACROS
#define __STDC_CONSTANT_MACROS
#endif // __STDC_CONSTANT_MACROS

#include <Weave/Profiles/data-management/Current/WdmManagedNamespace.h>
#include <Weave/Profiles/data-management/DataManagement.h>
#include <Weave/Support/WeaveFaultInjection.h>

#include <Weave/Profiles/status-report/StatusReportProfile.h>

#if WEAVE_CONFIG_ENABLE_RELIABLE_MESSAGING

namespace nl {
namespace Weave {
namespace Profiles {
namespace WeaveMakeManagedNamespaceIdentifier(DataManagement, kWeaveManagedNamespaceDesignation_Current) {

SubscriptionHandler::SubscriptionHandler()
{
}

void SubscriptionHandler::InitAsFree ()
{
    // These variables are going to be changed and reset along with subscription state machine
    // Must update SubscriptionHandler::Abort for anything added here
    mAppState = NULL;
    mEventCallback = NULL;
    mCurrentState = kState_Free;
    mEC = NULL;
    mLivenessTimeoutMsec = kNoTimeout;
    mPeerNodeId = 0;
    mSubscriptionId = 0;
    mBinding = NULL;
    mRefCount = 0;
    mIsInitiator = false;
    mTraitInstanceList = NULL;
    mNumTraitInstances = 0;
    mMaxNotificationSize = 0;
    mSubscribeToAllEvents = false;
    mCurProcessingTraitInstanceIdx = 0;
    mCurrentImportance = kImportanceType_Invalid;
    mBytesOffloaded = 0;

    memset(mSelfVendedEvents, 0, sizeof(mSelfVendedEvents));
    memset(mLastScheduledEventId, 0, sizeof(mLastScheduledEventId));
}

WEAVE_ERROR SubscriptionHandler::AcceptSubscribeRequest(const uint32_t aLivenessTimeoutSec)
{
    WEAVE_ERROR err = WEAVE_NO_ERROR;

    WeaveLogDetail(DataManagement, "Handler[%u] [%5.5s] %s Ref(%d)",
            SubscriptionEngine::GetInstance()->GetHandlerId(this), GetStateStr(), __func__, mRefCount);

    VerifyOrExit(kState_Subscribing_Evaluating == mCurrentState, err = WEAVE_ERROR_INCORRECT_STATE);

    if (mIsInitiator)
    {
        // Do nothing, as the handler on the initiator side of a mutual subscription doesn't need to timeout
        // The client machinery would kill both if anything happens
    }
    else
    {
        // We can only change the timeout spec if we're a responder
        if (aLivenessTimeoutSec <= kMaxTimeoutSec)
        {
            mLivenessTimeoutMsec = aLivenessTimeoutSec * 1000;
        }
        else
        {
            ExitNow(err = WEAVE_ERROR_INVALID_ARGUMENT);
        }
    }

    // make sure the changes in Binding in event callback
    // is reflected onto the active EC before we send out the first notify request

    err = mBinding->AdjustResponseTimeout(mEC);
    SuccessOrExit(err);

    // walk through the path list, prime the client
    MoveToState(kState_Subscribing);

    // Note that the call to NotificationEngine::Run could actually cause this particular handler to be aborted
    SubscriptionEngine::GetInstance()->GetNotificationEngine()->Run();

exit:
    WeaveLogFunctError(err);

    if (WEAVE_NO_ERROR != err)
    {
        AbortSubscription();
    }
    return err;
}

void SubscriptionHandler::DefaultEventHandler(EventID aEvent, const InEventParam & aInParam, OutEventParam & aOutParam)
{
    IgnoreUnusedVariable(aInParam);
    IgnoreUnusedVariable(aOutParam);

    WeaveLogDetail(DataManagement, "%s event: %d", __func__, aEvent);

    switch (aEvent)
    {
    case kEvent_OnSubscribeRequestParsed:
        // reject, don't care about current state should be in kState_Subscribing
        (void)aInParam.mSubscribeRequestParsed.mHandler->EndSubscription(nl::Weave::Profiles::kWeaveProfile_Common,
            nl::Weave::Profiles::Common::kStatus_UnsupportedMessage);
        break;

    default:
        // ignore
        break;
    }
}

WEAVE_ERROR SubscriptionHandler::EndSubscription(const uint32_t aReasonProfileId, const uint16_t aReasonStatusCode)
{
    WEAVE_ERROR err = WEAVE_NO_ERROR;
    bool abortOnError = true;

    WeaveLogDetail(DataManagement, "Handler[%u] [%5.5s] %s Ref(%d)",
            SubscriptionEngine::GetInstance()->GetHandlerId(this), GetStateStr(), __func__, mRefCount);

    switch (mCurrentState)
    {
    case kState_Subscribing_Evaluating:
    case kState_Subscribing:
        // reject the request with status report without any callback to app layer
        {
            uint8_t * p;
            uint8_t statusReportLen = 6;
            PacketBuffer *msgBuf = PacketBuffer::NewWithAvailableSize(statusReportLen);
            VerifyOrExit(NULL != msgBuf, err = WEAVE_ERROR_NO_MEMORY);

            p = msgBuf->Start();
            nl::Weave::Encoding::LittleEndian::Write32(p, aReasonProfileId);
            nl::Weave::Encoding::LittleEndian::Write16(p, aReasonStatusCode);
            msgBuf->SetDataLength(statusReportLen);

            err = mEC->SendMessage(nl::Weave::Profiles::kWeaveProfile_Common,
                nl::Weave::Profiles::Common::kMsgType_StatusReport,
                msgBuf,
                mEC->HasPeerRequestedAck() ? nl::Weave::ExchangeContext::kSendFlag_RequestAck : 0);
            msgBuf = NULL;
            SuccessOrExit(err);

            // Close our EC, but not abort it since the status report will still be in flight.
            FlushExistingExchangeContext(false);

            // This will clean-up the handler and reset it to the right state, but leave the EC untouched since we've already closed it out above
            AbortSubscription();

            ExitNow ();
        }
        break;

    case kState_Subscribing_Notifying:
    case kState_Subscribing_Responding:
    case kState_SubscriptionEstablished_Notifying:
    case kState_Canceling:
        // message in flight - for now, we're not going to take any action in these cases since we haven't
        // spent enough time plumbing the depths here.
        abortOnError = false;
        ExitNow (err = WEAVE_ERROR_INCORRECT_STATE);
        break;

    case kState_SubscriptionEstablished_Idle:
#if WDM_ENABLE_SUBSCRIPTION_CANCEL
        err = Cancel ();
#else // WDM_ENABLE_SUBSCRIPTION_CANCEL
        AbortSubscription();
#endif // WDM_ENABLE_SUBSCRIPTION_CANCEL
        break;

    default:
        // nothing we can do
        abortOnError = false;
        ExitNow (err = WEAVE_ERROR_INCORRECT_STATE);
        break;
    }

exit:
    WeaveLogFunctError(err);

    if (abortOnError && (err != WEAVE_NO_ERROR)) {
        AbortSubscription();
    }

    return err;
}

inline WEAVE_ERROR SubscriptionHandler::ParsePathVersionEventLists(
        SubscribeRequest::Parser & aRequest,
        uint32_t & aRejectReasonProfileId,
        uint16_t & aRejectReasonStatusCode)
{
    WEAVE_ERROR err = WEAVE_NO_ERROR;
    bool parsingCompletedSuccessfully = false;
    bool IsVersionListPresent = false;
    nl::Weave::TLV::TLVReader pathListIterator;
    PathList::Parser pathList;
    VersionList::Parser versionList;

    aRejectReasonProfileId = nl::Weave::Profiles::kWeaveProfile_Common;
    aRejectReasonStatusCode = nl::Weave::Profiles::Common::kStatus_BadRequest;

    err = aRequest.GetPathList(&pathList);
    SuccessOrExit(err);

    pathList.GetReader(&pathListIterator);

    err = aRequest.GetVersionList(&versionList);
    switch (err)
    {
    case WEAVE_NO_ERROR:
        IsVersionListPresent = true;
        break;
    case WEAVE_END_OF_TLV:
        err = WEAVE_NO_ERROR;
        break;
    default:
        ExitNow();
    }

    while (WEAVE_NO_ERROR == (err = pathListIterator.Next()))
    {
        TraitDataHandle traitDataHandle;
        PropertyPathHandle propertyPathHandle;
        TraitDataSource *dataSource;
        TraitInstanceInfo *traitInstance = NULL;
        nl::Weave::TLV::TLVReader pathReader;
        SchemaVersionRange requestedSchemaVersionRange, computedVersionIntersection;
        SchemaVersion computedForwardRequestedVersion;

        if (IsVersionListPresent)
        {
            // verify that we still have something to read in version list
            if (WEAVE_NO_ERROR != (err = versionList.Next()))
            {
                // Failed at reading from version list, reject with the default reason of bad request
                ExitNow();
            }
        }

        // Make a copy here
        pathReader.Init(pathListIterator);

        err = SubscriptionEngine::GetInstance()->mPublisherCatalog->AddressToHandle(pathReader, traitDataHandle, requestedSchemaVersionRange);

        if (WEAVE_ERROR_INVALID_PROFILE_ID == err)
        {
            // note we can safely 'continue' directly from here because 'pathReader' is a copy of 'reader'
            // so the actual reader is advancing irrespective of what happened during parsing

            WeaveLogDetail(DataManagement, "Unknown profile ID in the subscribe request, ignore.");
            continue;
        }

        SuccessOrExit(err);

        err = SubscriptionEngine::GetInstance()->mPublisherCatalog->Locate(traitDataHandle, &dataSource);
        SuccessOrExit(err);

        if (dataSource->GetSchemaEngine()->GetVersionIntersection(requestedSchemaVersionRange, computedVersionIntersection)) {
            computedForwardRequestedVersion = dataSource->GetSchemaEngine()->GetHighestForwardVersion(computedVersionIntersection.mMaxVersion);
        }
        else {
            WeaveLogDetail(DataManagement, "Mismatch in requested version on handle %u (requested: %u, %u)", traitDataHandle, requestedSchemaVersionRange.mMaxVersion, requestedSchemaVersionRange.mMinVersion);
            continue;
        }

        err = dataSource->GetSchemaEngine()->MapPathToHandle(pathReader, propertyPathHandle);
        SuccessOrExit(err);

        if (propertyPathHandle != kRootPropertyPathHandle)
        {
            WeaveLogError(DataManagement, "Device only supports subscriptions to root!\n");
            ExitNow();
        }

        for (size_t i = 0; i < mNumTraitInstances; ++i)
        {
            if (mTraitInstanceList[i].mTraitDataHandle == traitDataHandle)
            {
                // we found a prior trait instance which has the same root (trait instance)
                traitInstance = &mTraitInstanceList[i];
                break;
            }
        }

        if (!traitInstance)
        {
            // allocate a new trait instance
            WEAVE_FAULT_INJECT(FaultInjection::kFault_WDM_TraitInstanceNew, SuccessOrExit(err = WEAVE_ERROR_NO_MEMORY));

            if (SubscriptionEngine::GetInstance()->mNumTraitInfosInPool < SubscriptionEngine::kMaxNumPathGroups)
            {
                traitInstance = SubscriptionEngine::GetInstance()->mTraitInfoPool + SubscriptionEngine::GetInstance()->mNumTraitInfosInPool;
                ++mNumTraitInstances;
                ++(SubscriptionEngine::GetInstance()->mNumTraitInfosInPool);
                SYSTEM_STATS_INCREMENT(nl::Weave::System::Stats::kWDMNext_NumTraits);

                traitInstance->Init();
            }
            else
            {
                // we run out of trait instances, abort
                // Note it might help the client understanding what's going on with an error status like
                // "out of memory" or "internal error", but it's pretty common that a server doesn't disclose
                // too much internal status to clients
                SuccessOrExit(err = WEAVE_ERROR_NO_MEMORY);
            }
        }

        traitInstance->mTraitDataHandle = traitDataHandle;
        traitInstance->mRequestedVersion = computedForwardRequestedVersion;

        if (NULL == mTraitInstanceList)
        {
            // this the first trait instance for this subscription
            // mNumTraitInstanceList has already be incremented
            mTraitInstanceList = traitInstance;
        }

        if (!IsVersionListPresent)
        {
            // no existing version
            WeaveLogDetail(DataManagement, "Handler[%u] Syncing is requested for trait[%u].path[%u]",
                SubscriptionEngine::GetInstance()->GetHandlerId(this), traitDataHandle, propertyPathHandle);

            traitInstance->SetDirty();
        }
        else
        {
            if (versionList.IsNull())
            {
                // no existing version

                WeaveLogDetail(DataManagement, "Handler[%u] Syncing is requested for trait[%u].path[%u]",
                    SubscriptionEngine::GetInstance()->GetHandlerId(this), traitDataHandle, propertyPathHandle);

                traitInstance->SetDirty();
            }
            else
            {
                uint64_t existingVersion;
                uint64_t datasourceVersion = dataSource->GetVersion();
                err = versionList.GetVersion(&existingVersion);
                SuccessOrExit(err);

                if (existingVersion != datasourceVersion)
                {
                    WeaveLogDetail(DataManagement, "Handler[%u] Syncing is necessary for trait[%u].path[%u]",
                        SubscriptionEngine::GetInstance()->GetHandlerId(this), traitDataHandle, propertyPathHandle);

                    WeaveLogIfFalse(existingVersion < datasourceVersion);
                    traitInstance->SetDirty();
                }
                else
                {
                    WeaveLogDetail(DataManagement, "Handler[%u] Syncing is NOT necessary for trait[%u].path[%u]",
                        SubscriptionEngine::GetInstance()->GetHandlerId(this), traitDataHandle, propertyPathHandle);
                }
            }

#if WEAVE_CONFIG_WDM_NEXT_SUBSCRIPTION_HANDLER_DEBUG
            // we're going to log some warning if
            // 1. this is not the first encounter with this trait instance
            // 2. we're changing the decision made for this trait instance
            // this warning implies the versions for diff paths provided in subscribe request are inconsistent
            if ((!isFirstPathInGroup) && (previousSyncDecision != pathGroup->mFurtherSyncNeededForThisSubscription))
            {
                WeaveLogDetail(DataManagement, "Handler[%u] Inconsistent version information for trait[%u], reject",
                    SubscriptionEngine::GetInstance()->GetHandlerId(this), root);

                RejectReasonProfileId = nl::Weave::Profiles::kWeaveProfile_WDM;
                RejectReasonStatusCode = nl::Weave::Profiles::DataManagement::kStatus_VersionMismatch;
                ExitNow();
            }
#endif // WEAVE_CONFIG_WDM_NEXT_SUBSCRIPTION_HANDLER_DEBUG
        }
    }

    WeaveLogDetail(DataManagement, "Number allocated of trait info instances: %u", SubscriptionEngine::GetInstance()->mNumTraitInfosInPool);

    // Check if we still have anything in version list after we run out of paths
    if ((WEAVE_END_OF_TLV == err) && IsVersionListPresent)
    {
        // everything is fine, but we just ran out of paths
        if (IsVersionListPresent && (WEAVE_END_OF_TLV != (err = versionList.Next())))
        {
            // how come the version list has not been exhausted?
            WeaveLogDetail(DataManagement, "Path has been exhausted unexpectedly, rejecting");
            ExitNow();
        }
    }

    // Setting it to false is not absolutely necessary, as mSubscribeToAllEvents is reset to false in
    // InitAsFree, and Abort, and again in GetSubscribeToAllEvents
    mSubscribeToAllEvents = false;
    err = aRequest.GetSubscribeToAllEvents(&mSubscribeToAllEvents);
    if (WEAVE_END_OF_TLV == err)
    {
        err = WEAVE_NO_ERROR;
    }
    VerifyOrExit (WEAVE_NO_ERROR == err, /* no-op */);

    memset(mSelfVendedEvents, 0, sizeof(mSelfVendedEvents));

    if (mSubscribeToAllEvents)
    {
        EventList::Parser eventList;
        uint64_t localNodeId = mEC->ExchangeMgr->FabricState->LocalNodeId;
        uint64_t sourceId, importance, eventId;
        err = aRequest.GetLastObservedEventIdList(&eventList);

        if (WEAVE_NO_ERROR == err)
        {
            while ( (err = eventList.Next()) == WEAVE_NO_ERROR )
            {
                nl::Weave::TLV::TLVReader eventReader;
                Event::Parser event;

                eventList.GetReader(&eventReader);

                err = event.Init(eventReader);
                SuccessOrExit(err);

                err = event.GetSourceId(&sourceId);
                SuccessOrExit(err);

                err = event.GetImportance(&importance);
                SuccessOrExit(err);

                err = event.GetEventId(&eventId);
                SuccessOrExit(err);

                // At the moment, we don't support event aggregation in subscription

                if ((sourceId == localNodeId) &&
                        (static_cast<uint32_t>(importance) >= static_cast<uint32_t>(kImportanceType_First)) &&
                        (static_cast<uint32_t>(importance) <= static_cast<uint32_t>(kImportanceType_Last)))
                {
                    mSelfVendedEvents[static_cast<uint32_t>(importance)-static_cast<uint32_t>(kImportanceType_First)] = static_cast<event_id_t>(eventId);
                }
                else
                {
                    ExitNow();
                }
            }
            if (WEAVE_END_OF_TLV == err)
            {
                err = WEAVE_NO_ERROR;
            }
        }
        else if (WEAVE_END_OF_TLV == err)
        {
            err = WEAVE_NO_ERROR;
        }
        else
        {
            ExitNow();
        }
    }

    // Great! We've successfully processed path, version, and event lists
    parsingCompletedSuccessfully = true;

exit:
    WeaveLogFunctError(err);

    if (!parsingCompletedSuccessfully)
    {
        if (WEAVE_NO_ERROR == err)
        {
            err = WEAVE_ERROR_BAD_REQUEST;
        }
    }

    return err;
}

WEAVE_ERROR SubscriptionHandler::ParseSubscriptionId(
        SubscribeRequest::Parser & aRequest,
        uint32_t & aRejectReasonProfileId,
        uint16_t & aRejectReasonStatusCode,
        const uint64_t aRandomNumber)
{
    WEAVE_ERROR err = WEAVE_NO_ERROR;
    bool parsingCompletedSuccessfully = false;

    // From now on, the main reason for rejection is invalid subscription ID
    aRejectReasonProfileId = nl::Weave::Profiles::kWeaveProfile_WDM;
    aRejectReasonStatusCode = nl::Weave::Profiles::DataManagement::kStatus_InvalidSubscriptionID;

    err = aRequest.GetSubscriptionID(&mSubscriptionId);
    if (WEAVE_END_OF_TLV == err)
    {
        // good, this is a normal request
        // use the generated subscription id

        mSubscriptionId = aRandomNumber;
        err = WEAVE_NO_ERROR;
    }
    else if (WEAVE_NO_ERROR == err)
    {
#if WDM_ENABLE_SUBSCRIPTION_CLIENT

        // Note FindHandler is not to find this particular handler, for we're still in evaluating state
        if (NULL == SubscriptionEngine::GetInstance()->FindHandler(mPeerNodeId, mSubscriptionId))
        {
            if (NULL != SubscriptionEngine::GetInstance()->FindClient(mPeerNodeId, mSubscriptionId))
            {
                // this is the second half of a mutual subscription (and we did find the first half)
                // continue using id in the request
                mIsInitiator = true;
            }
            else
            {
                // This incoming request carries some subscription ID, which implies there should be an existing client
                // associated with the remote node already. Reject it if we couldn't find one
                WeaveLogError(DataManagement, "No matching subscription found for incoming mutual subscription");
                ExitNow();
            }
        }
        else
        {
            // This incoming request carries some subscription ID, but we already have an existing
            // subscription with the same client and the same ID. It's not obvious which one we should keep,
            // and we choose to keep the existing one here.
            WeaveLogError(DataManagement, "Mutual subscription with duplicated ID");
            ExitNow();
        }

#else //#if WDM_ENABLE_SUBSCRIPTION_CLIENT
        WeaveLogError(DataManagement, "Mutual subscription is not supported");
        RejectReasonProfileId = nl::Weave::Profiles::kWeaveProfile_Common;
        RejectReasonStatusCode = nl::Weave::Profiles::Common::kStatus_BadRequest;
        ExitNow();
#endif //#if WDM_ENABLE_SUBSCRIPTION_CLIENT
    }
    else
    {
        aRejectReasonProfileId = nl::Weave::Profiles::kWeaveProfile_Common;
        aRejectReasonStatusCode = nl::Weave::Profiles::Common::kStatus_BadRequest;
        ExitNow();
    }

    parsingCompletedSuccessfully = true;

exit:
    WeaveLogFunctError(err);

    if (!parsingCompletedSuccessfully)
    {
        if (WEAVE_NO_ERROR == err)
        {
            err = WEAVE_ERROR_BAD_REQUEST;
        }
    }

    return err;
}

void SubscriptionHandler::InitWithIncomingRequest (Binding * const aBinding,
    const uint64_t aRandomNumber, nl::Weave::ExchangeContext * aEC, const nl::Inet::IPPacketInfo *aPktInfo,
    const nl::Weave::WeaveMessageInfo *aMsgInfo, PacketBuffer * aPayload)
{
    WEAVE_ERROR err = WEAVE_NO_ERROR;
    uint32_t RejectReasonProfileId = nl::Weave::Profiles::kWeaveProfile_Common;
    uint16_t RejectReasonStatusCode = nl::Weave::Profiles::Common::kStatus_BadRequest;
    nl::Weave::TLV::TLVReader reader;
    SubscribeRequest::Parser request;

    WeaveLogDetail(DataManagement, "Handler[%u] [%5.5s] %s Ref(%d)",
            SubscriptionEngine::GetInstance()->GetHandlerId(this), GetStateStr(), __func__, mRefCount);

    WeaveLogIfFalse(0 == mRefCount);

    _AddRef();

    aBinding->AddRef();
    mBinding = aBinding;

    mPeerNodeId = aMsgInfo->SourceNodeId;
    mBytesOffloaded = 0;

    mEC = aEC;
    // The ownership has been transferred to this subscription
    aEC = NULL;

    // Add reference when we enter this initial state
    // This is needed because the app layer doesn't automatically get hold of this instance,
    // but we need this instance to be around until we clear the protocol state machine (by entering aborted)
    _AddRef();
    MoveToState(kState_Subscribing_Evaluating);

    InitExchangeContext();

    // First stage: Initial parsing
    // Convert all path lists to TargetHandles and PathHandles
    // Note that notification rejection is not supported, so version is only validated but not tracked
    // This job doesn't require application level intervention
    // If any conversion fails (which means schema error), just reject this 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

    // Second stage: Subscription ID
    err = ParseSubscriptionId(request, RejectReasonProfileId, RejectReasonStatusCode, aRandomNumber);
    SuccessOrExit(err);

    // Third stage: path, version, and events
    err = ParsePathVersionEventLists(request, RejectReasonProfileId, RejectReasonStatusCode);
    SuccessOrExit(err);

    // Final stage: app callback
    {
        InEventParam inParam;
        OutEventParam outParam;
        inParam.mSubscribeRequestParsed.mEC = mEC;
        inParam.mSubscribeRequestParsed.mPktInfo = aPktInfo;
        inParam.mSubscribeRequestParsed.mMsgInfo = aMsgInfo;
        inParam.mSubscribeRequestParsed.mTraitInstanceList = mTraitInstanceList;
        inParam.mSubscribeRequestParsed.mNumTraitInstances = mNumTraitInstances;
        inParam.mSubscribeRequestParsed.mSubscribeToAllEvents = mSubscribeToAllEvents;

        err = request.GetSubscribeTimeoutMin(&inParam.mSubscribeRequestParsed.mTimeoutSecMin);
        if (WEAVE_END_OF_TLV == err)
        {
            err = WEAVE_NO_ERROR;
            inParam.mSubscribeRequestParsed.mTimeoutSecMin = kNoTimeout;
        }
        else if (WEAVE_NO_ERROR != err)
        {
            ExitNow();
        }

        err = request.GetSubscribeTimeoutMax(&inParam.mSubscribeRequestParsed.mTimeoutSecMax);
        if (WEAVE_END_OF_TLV == err)
        {
            err = WEAVE_NO_ERROR;
            inParam.mSubscribeRequestParsed.mTimeoutSecMax = kNoTimeout;
        }
        else if (WEAVE_NO_ERROR != err)
        {
            ExitNow();
        }

        // Note that err must be WEAVE_NO_ERROR now, otherwise we should just reject and not call to app layer
        SuccessOrExit(err);

        inParam.mSubscribeRequestParsed.mHandler = this;
        inParam.mSubscribeRequestParsed.mIsSubscriptionIdValid = mIsInitiator;
        inParam.mSubscribeRequestParsed.mSubscriptionId = mSubscriptionId;

        // From now on, the app layer has to make the explicit call to accept or reject,
        // and then Free this handler later
        mEventCallback(mAppState, kEvent_OnSubscribeRequestParsed, inParam, outParam);
        // Note that either Abort or EndSubscription could be called.
        // There is no need to explicitly reject the request in this function.
    }

exit:
    WeaveLogFunctError(err);

    // Release the pbuf first, guaranteed to be non-NULL
    PacketBuffer::Free(aPayload);

    if (WEAVE_NO_ERROR != err)
    {
        // reject the request if we encountered any error
        (void)EndSubscription(RejectReasonProfileId, RejectReasonStatusCode);
    }

    _Release ();
}

WEAVE_ERROR SubscriptionHandler::SendNotificationRequest(PacketBuffer * aMsgBuf)
{
    WEAVE_ERROR err = WEAVE_NO_ERROR;

    WeaveLogDetail(DataManagement, "Handler[%u] [%5.5s] %s Ref(%d)",
        SubscriptionEngine::GetInstance()->GetHandlerId(this), GetStateStr(), __func__, mRefCount);

    WeaveLogIfFalse((kState_Subscribing == mCurrentState) || (kState_SubscriptionEstablished_Idle == mCurrentState));

    // Make sure we're not freed by accident.
    _AddRef();

    // Create new exchange context when we're idle (otherwise we must be using the existing EC)
    if (kState_SubscriptionEstablished_Idle == mCurrentState)
    {
        err = ReplaceExchangeContext ();
        SuccessOrExit(err);
    }

    // Note we're sending back a message using an EC initiated by the client
    err = mEC->SendMessage(nl::Weave::Profiles::kWeaveProfile_WDM, kMsgType_NotificationRequest, aMsgBuf,
        nl::Weave::ExchangeContext::kSendFlag_ExpectResponse);
    aMsgBuf = NULL;
    SuccessOrExit(err);

    mCurrentState = (mCurrentState == kState_Subscribing) ? kState_Subscribing_Notifying : kState_SubscriptionEstablished_Notifying;

exit:
    WeaveLogFunctError(err);

    if (NULL != aMsgBuf)
    {
        PacketBuffer::Free(aMsgBuf);
        aMsgBuf = NULL;
    }

    if (WEAVE_NO_ERROR != err)
    {
        HandleSubscriptionTerminated(err, NULL);
    }

    _Release();

    return err;
}

void SubscriptionHandler::OnNotifyProcessingComplete(
    const bool aPossibleLossOfEvent,
    const LastVendedEvent aLastVendedEventList[],
    const size_t aLastVendedEventListSize)
{
    if (mCurrentState == kState_Subscribing) {
        SendSubscribeResponse(aPossibleLossOfEvent, aLastVendedEventList, aLastVendedEventListSize);
    }
}

WEAVE_ERROR SubscriptionHandler::SendSubscribeResponse(
    const bool aPossibleLossOfEvent,
    const LastVendedEvent aLastVendedEventList[],
    const size_t aLastVendedEventListSize)
{
    WEAVE_ERROR err = WEAVE_NO_ERROR;
    PacketBuffer * msgBuf = NULL;
    nl::Weave::TLV::TLVWriter writer;
    SubscribeResponse::Builder response;

    WeaveLogDetail(DataManagement, "Handler[%u] [%5.5s] %s Ref(%d)",
            SubscriptionEngine::GetInstance()->GetHandlerId(this), GetStateStr(), __func__, mRefCount);

    // Make sure we're not freed by accident.
    _AddRef();

    msgBuf = PacketBuffer::New();
    VerifyOrExit(NULL != msgBuf, err = WEAVE_ERROR_NO_MEMORY);

    writer.Init(msgBuf);

    response.Init(&writer);
    response.SubscriptionID(mSubscriptionId);

    if (!mIsInitiator)
    {
        // If we're the initiator in a mutual subscription,
        // the subscribe response must not carry a timeout spec
        if (kNoTimeout != mLivenessTimeoutMsec)
        {
            response.SubscribeTimeout(mLivenessTimeoutMsec / 1000);
        }
    }

    if (aPossibleLossOfEvent)
    {
        response.PossibleLossOfEvents(aPossibleLossOfEvent);
    }

    if (aLastVendedEventListSize > 0)
    {
        EventList::Builder & eventList = response.CreateLastVendedEventIdListBuilder();

        for (size_t n = 0; n < aLastVendedEventListSize; ++n)
        {
            Event::Builder & event = eventList.CreateEventBuilder();
            event.SourceId(aLastVendedEventList[n].mSourceId)
                .Importance(aLastVendedEventList[n].mImportance)
                .EventId(aLastVendedEventList[n].mEventId)
                .EndOfEvent ();
            SuccessOrExit(err = event.GetError());
        }
        eventList.EndOfEventList ();
        SuccessOrExit(err = eventList.GetError());
    }

    response.EndOfResponse();

    err = response.GetError();
    SuccessOrExit(err);

    err = writer.Finalize();
    SuccessOrExit(err);

    // Note we're sending back a message using an EC initiated by the client
    err = mEC->SendMessage(nl::Weave::Profiles::kWeaveProfile_WDM, kMsgType_SubscribeResponse, msgBuf,
        nl::Weave::ExchangeContext::kSendFlag_RequestAck);
    msgBuf = NULL;
    SuccessOrExit(err);

    // Wait for Ack to move to alive state
    MoveToState(SubscriptionHandler::kState_Subscribing_Responding);

exit:
    WeaveLogFunctError(err);

    if (NULL != msgBuf)
    {
        PacketBuffer::Free(msgBuf);
        msgBuf = NULL;
    }

    if (WEAVE_NO_ERROR != err)
    {
        HandleSubscriptionTerminated(err, NULL);
    }

    _Release();

    return err;
}

Binding * SubscriptionHandler::GetBinding(void) const
{
    return mBinding;
}

uint64_t SubscriptionHandler::GetPeerNodeId(void) const
{
    return mPeerNodeId;
}

void SubscriptionHandler::InitExchangeContext()
{
    mEC->AppState = this;
    mEC->OnResponseTimeout = OnResponseTimeout;
    mEC->OnSendError = OnSendError;
    mEC->OnAckRcvd = OnAckReceived;
    mEC->OnMessageReceived = OnMessageReceivedFromLocallyHeldExchange;
}

WEAVE_ERROR SubscriptionHandler::ReplaceExchangeContext ()
{
    WEAVE_ERROR err = WEAVE_NO_ERROR;
    InEventParam inParam;
    OutEventParam outParam;

    WeaveLogDetail(DataManagement, "Handler[%u] [%5.5s] %s Ref(%d)",
            SubscriptionEngine::GetInstance()->GetHandlerId(this), GetStateStr(), __func__, mRefCount);

    // Make sure we're not freed by accident.
    _AddRef();

    FlushExistingExchangeContext();

    err = mBinding->NewExchangeContext(mEC);
    SuccessOrExit(err);

    InitExchangeContext();

    inParam.mExchangeStart.mEC = mEC;
    inParam.mExchangeStart.mHandler = this;
    mEventCallback(mAppState, kEvent_OnExchangeStart, inParam, outParam);

exit:
    WeaveLogFunctError(err);

    _Release();

    return err;
}

void SubscriptionHandler::FlushExistingExchangeContext (const bool aAbortNow)
{
    if (NULL != mEC)
    {
        mEC->AppState = NULL;
        mEC->OnMessageReceived = NULL;
        mEC->OnResponseTimeout = NULL;
        mEC->OnSendError = NULL;
        mEC->OnAckRcvd = NULL;
        if (aAbortNow)
        {
            mEC->Abort();
        }
        else
        {
            mEC->Close();
        }
        mEC = NULL;
    }
}

#if WDM_ENABLE_SUBSCRIPTION_CANCEL
WEAVE_ERROR SubscriptionHandler::Cancel ()
{
    WEAVE_ERROR err = WEAVE_NO_ERROR;
    PacketBuffer * msgBuf = NULL;
    bool cancel = false;

    WeaveLogDetail(DataManagement, "Handler[%u] [%5.5s] %s Ref(%d)",
            SubscriptionEngine::GetInstance()->GetHandlerId(this), GetStateStr(), __func__, mRefCount);

    // Make sure we're not freed by accident.
    _AddRef();

    switch (mCurrentState)
    {
    case kState_SubscriptionEstablished_Notifying:
        // abort whatever we're doing (notification request)
        FlushExistingExchangeContext(true);
        cancel = true;

        break;

    case kState_SubscriptionEstablished_Idle:
        // send a cancel request
        cancel = true;
        break;

    default:
        ExitNow(err = WEAVE_ERROR_INCORRECT_STATE);
    }

    if (cancel)
    {
        nl::Weave::TLV::TLVWriter writer;
        SubscribeCancelRequest::Builder request;

        msgBuf = PacketBuffer::NewWithAvailableSize(request.kBaseMessageSubscribeId_PayloadLen);
        VerifyOrExit(NULL != msgBuf, err = WEAVE_ERROR_NO_MEMORY);
        writer.Init(msgBuf);

        err = ReplaceExchangeContext ();
        SuccessOrExit(err);

        request.Init(&writer);

        request.SubscriptionID(mSubscriptionId)
               .EndOfRequest();

        err = request.GetError();
        SuccessOrExit (err);

        err = writer.Finalize();
        SuccessOrExit (err);

        // NOTE: State could be changed if there is a sync error callback from message layer
        err = mEC->SendMessage(nl::Weave::Profiles::kWeaveProfile_WDM, kMsgType_SubscribeCancelRequest,
            msgBuf, nl::Weave::ExchangeContext::kSendFlag_ExpectResponse);
        msgBuf = NULL;
        SuccessOrExit(err);

        MoveToState(SubscriptionHandler::kState_Canceling);
    }

exit:
    WeaveLogFunctError(err);

    if (NULL != msgBuf)
    {
        PacketBuffer::Free(msgBuf);
        msgBuf = NULL;
    }

    _Release();

    return err;
}
#endif // WDM_ENABLE_SUBSCRIPTION_CANCEL

WEAVE_ERROR SubscriptionHandler::GetSubscriptionId (uint64_t * const apSubscriptionId)
{
    WEAVE_ERROR err =  WEAVE_NO_ERROR;

    *apSubscriptionId = 0;

    if ((mCurrentState >= kState_SubscriptionInfoValid_Begin) && (mCurrentState <= kState_SubscriptionInfoValid_End))
    {
        *apSubscriptionId = mSubscriptionId;
    }
    else
    {
        ExitNow(err = WEAVE_ERROR_INCORRECT_STATE);
    }

exit:
    WeaveLogFunctError(err);

    return err;
}

void SubscriptionHandler::_AddRef ()
{
    WeaveLogIfFalse(mRefCount < INT8_MAX);

    ++mRefCount;
}

void SubscriptionHandler::_Release ()
{
    WeaveLogIfFalse(mRefCount > 0);

    --mRefCount;

    if (0 == mRefCount)
    {
        SYSTEM_STATS_DECREMENT(nl::Weave::System::Stats::kWDMNext_NumSubscriptionHandlers);
        AbortSubscription();
    }
}

void SubscriptionHandler::HandleSubscriptionTerminated(WEAVE_ERROR aReason,
        StatusReporting::StatusReport *aStatusReportPtr)
{
    void * const appState = mAppState;
    EventCallback callbackFunc = mEventCallback;

    WeaveLogDetail(DataManagement, "Handler[%u] [%5.5s] %s Ref(%d)",
        SubscriptionEngine::GetInstance()->GetHandlerId(this), GetStateStr(), __func__, mRefCount);

    _AddRef();

    // Flush most internal states, except for mRefCount and mCurrentState
    // move to kState_Aborted
    AbortSubscription();

    // Make app layer callback in kState_Aborted
    // Note that mRefCount should still be more than 0 when we make this callback.
    // Since this function does not call _AddRef, it is possible if the app
    // calls Free() that we'll exit the app callback in a free state.
    // In that case, we won't have any cleanup to do.
    // On the other hand, if the caller of this function has AddRef'ed,
    // the state after exiting the app callback could be Aborted.
    // In that case, we still won't have cleanup, since everything will
    // be cleaned up upon _Release by the caller of this func.
    if (NULL != callbackFunc)
    {
        InEventParam inParam;
        OutEventParam outParam;

        inParam.Clear();
        outParam.Clear();

        inParam.mSubscriptionTerminated.mReason = aReason;
        inParam.mSubscriptionTerminated.mHandler = this;
        if (aStatusReportPtr != NULL)
        {
            inParam.mSubscriptionTerminated.mStatusProfileId = aStatusReportPtr->mProfileId;
            inParam.mSubscriptionTerminated.mStatusCode = aStatusReportPtr->mStatusCode;
            inParam.mSubscriptionTerminated.mAdditionalInfoPtr = &(aStatusReportPtr->mAdditionalInfo);
        }

        callbackFunc(appState, kEvent_OnSubscriptionTerminated, inParam, outParam);
    }
    else
    {
        WeaveLogDetail(DataManagement, "Handler[%u] [%5.5s] %s Ref(%d) app layer callback skipped",
            SubscriptionEngine::GetInstance()->GetHandlerId(this), GetStateStr(), __func__, mRefCount);
    }

    _Release();
}

void SubscriptionHandler::AbortSubscription()
{
    WEAVE_ERROR err = WEAVE_NO_ERROR;
    const bool nullReference = (0 == mRefCount);

    if (!nullReference)
    {
        // only do verbose logging if we're still referenced by someone
        WeaveLogDetail(DataManagement, "Handler[%u] [%5.5s] %s Ref(%d)",
            SubscriptionEngine::GetInstance()->GetHandlerId(this), GetStateStr(), __func__, mRefCount);

        // Make sure we're not freed by accident.
        // NOTE: In the last Abort call from _Release, mRefCount is already 0.
        // In that case, we do not need this AddRef/Release pair, and we move to FREE state.
        _AddRef();
    }

    if (kState_Free == mCurrentState)
    {
        // This must not happen
        ExitNow(err = WEAVE_ERROR_INCORRECT_STATE);
    }
    else if (kState_Aborted == mCurrentState || kState_Aborting == mCurrentState)
    {
        // we're already aborted, so there is nothing else to flush
        ExitNow();
    }
    else
    {
        // This is an intermediate state for external calls during the abort process
        uint64_t peerNodeId = mPeerNodeId;
        uint64_t subscriptionId = mSubscriptionId;
        bool notifyEngineCleanupNeeded = false;

        // If we were previously in a notifying state, we must tell NE so that it can do some clean-up.
        // However, we should wait till the subscription state has moved to aborting before attempting
        // to do so.
        if (IsNotifying()) {
            notifyEngineCleanupNeeded = true;
        }

        MoveToState(kState_Aborting);

        if (notifyEngineCleanupNeeded) {
            const bool notifySuccessfullyDelivered = true;

            SubscriptionEngine::GetInstance()->GetNotificationEngine()->OnNotifyConfirm(this, !notifySuccessfullyDelivered);
        }

        mBinding->SetProtocolLayerCallback(NULL, NULL);

        mBinding->Release();
        mBinding = NULL;

        FlushExistingExchangeContext(true);
        mLivenessTimeoutMsec = kNoTimeout;
        mPeerNodeId = 0;
        mSubscriptionId = 0;
        mIsInitiator = false;
        mAppState = NULL;
        mEventCallback = NULL;
        mMaxNotificationSize = 0;
        mSubscribeToAllEvents = false;
        mCurProcessingTraitInstanceIdx = 0;
        mCurrentImportance = kImportanceType_Invalid;
        (void)RefreshTimer();

        // release all trait instances back to the shared pool
        SubscriptionEngine::GetInstance()->ReclaimTraitInfo(this);

        mTraitInstanceList = NULL;
        mNumTraitInstances = 0;

        memset(mSelfVendedEvents, 0, sizeof(mSelfVendedEvents));
        memset(mLastScheduledEventId, 0, sizeof(mLastScheduledEventId));

        MoveToState(kState_Aborted);

        // decrease the ref we added when we first started the protocol state machine
        // now we've cleared this state machine (nothing for the protocol anymore),
        // this instance doesn't need to be around any more
        _Release();

#if WDM_ENABLE_SUBSCRIPTION_CLIENT
        (void)SubscriptionEngine::GetInstance()->UpdateClientLiveness(peerNodeId, subscriptionId, true);
#endif // WDM_ENABLE_SUBSCRIPTION_CLIENT

    }

exit:
    WeaveLogFunctError(err);

    if (nullReference)
    {
        // No one is referencing us, move to FREE
        MoveToState(kState_Free);
    }
    else
    {
        // Note that this call could lower ref count to 0 and cause this instance to be freed
        // this is because we have an extra _Release when we first move into Aborting state
        _Release();
    }
}

#if WDM_ENABLE_SUBSCRIPTION_CANCEL
void SubscriptionHandler::CancelRequestHandler (nl::Weave::ExchangeContext *aEC, const nl::Inet::IPPacketInfo *aPktInfo,
    const nl::Weave::WeaveMessageInfo *aMsgInfo, PacketBuffer *aPayload)
{
    WEAVE_ERROR err = WEAVE_NO_ERROR;
    uint8_t statusReportLen = 6;
    PacketBuffer *msgBuf = PacketBuffer::NewWithAvailableSize(statusReportLen);
    uint8_t * p;
    bool canceled = true;
    uint32_t statusProfile = nl::Weave::Profiles::kWeaveProfile_Common;
    uint16_t statusCode = nl::Weave::Profiles::Common::kStatus_Success;

    WeaveLogDetail(DataManagement, "Handler[%u] [%5.5s] %s Ref(%d)",
            SubscriptionEngine::GetInstance()->GetHandlerId(this), GetStateStr(), __func__, mRefCount);

    // Make sure we're not freed by accident.
    _AddRef();

    mBinding->AdjustResponseTimeout(aEC);

    VerifyOrExit(NULL != msgBuf, err = WEAVE_ERROR_NO_MEMORY);

    // Verify the cancel request is truly from the client.  If not, reject the request with
    // "invalid subscription id" to avoid revealing the existence of the subscription.
    if (!mBinding->IsAuthenticMessageFromPeer(aMsgInfo))
    {
        WeaveLogDetail(DataManagement, "Rejecting SubscribeCancelRequest from unauthorized source");
        canceled = false;
        statusProfile = nl::Weave::Profiles::kWeaveProfile_WDM;
        statusCode = kStatus_InvalidSubscriptionID;
    }

    p = msgBuf->Start();
    nl::Weave::Encoding::LittleEndian::Write32(p, statusProfile);
    nl::Weave::Encoding::LittleEndian::Write16(p, statusCode);
    msgBuf->SetDataLength(statusReportLen);

    err = aEC->SendMessage(nl::Weave::Profiles::kWeaveProfile_Common,
        nl::Weave::Profiles::Common::kMsgType_StatusReport,
        msgBuf,
        aEC->HasPeerRequestedAck() ? nl::Weave::ExchangeContext::kSendFlag_RequestAck : 0);
    msgBuf = NULL;
    SuccessOrExit(err);

exit:
    WeaveLogFunctError(err);

    if (NULL != msgBuf)
    {
        PacketBuffer::Free(msgBuf);
        msgBuf = NULL;
    }

    if (canceled)
    {
        HandleSubscriptionTerminated(WEAVE_ERROR_TRANSACTION_CANCELED, NULL);
    }

    _Release();
}
#endif // WDM_ENABLE_SUBSCRIPTION_CANCEL

void SubscriptionHandler::OnTimerCallback(System::Layer* aSystemLayer, void *aAppState, System::Error)
{
    SubscriptionHandler * const pHandler = reinterpret_cast<SubscriptionHandler*>(aAppState);

    pHandler->TimerEventHandler();
}

WEAVE_ERROR SubscriptionHandler::RefreshTimer (void)
{
    WEAVE_ERROR err = WEAVE_NO_ERROR;

    WeaveLogDetail(DataManagement, "Handler[%u] [%5.5s] %s Ref(%d)",
        SubscriptionEngine::GetInstance()->GetHandlerId(this), GetStateStr(), __func__, mRefCount);

    // Cancel timer first
    SubscriptionEngine::GetInstance()->GetExchangeManager()->MessageLayer->
            SystemLayer->CancelTimer(OnTimerCallback, this);

    // Arm timer according to current state
    switch (mCurrentState)
    {
    case kState_SubscriptionEstablished_Idle:
    case kState_SubscriptionEstablished_Notifying:
        if (mIsInitiator)
        {
            WeaveLogDetail(DataManagement, "Handler[%u] [%5.5s] %s Ref(%d) Ignored for handler on initiator",
                SubscriptionEngine::GetInstance()->GetHandlerId(this), GetStateStr(), __func__, mRefCount);
        }
        else if (kNoTimeout != mLivenessTimeoutMsec)
        {
            WeaveLogDetail(DataManagement, "Handler[%u] [%5.5s] %s Ref(%d) Set timer for liveness confirmation to %" PRIu32 " msec",
                SubscriptionEngine::GetInstance()->GetHandlerId(this), GetStateStr(), __func__, mRefCount, mLivenessTimeoutMsec);

            err = SubscriptionEngine::GetInstance()->GetExchangeManager()->MessageLayer->
                    SystemLayer->StartTimer(mLivenessTimeoutMsec, OnTimerCallback, this);

            VerifyOrExit(WEAVE_SYSTEM_NO_ERROR == err, /* no-op */);
        }
        break;
    case kState_Aborting:
        // Do nothing
        break;
    default:
        ExitNow(err = WEAVE_ERROR_INCORRECT_STATE);
    }

exit:
    WeaveLogFunctError(err);

    return err;
}

void SubscriptionHandler::TimerEventHandler (void)
{
    bool ShouldCleanUp = false;
    bool skipTimerCheck = false;

    if (0 == mRefCount)
    {
        skipTimerCheck = true;
        ExitNow();
    }

    // Make sure we're not freed by accident.
    _AddRef();

    if (kState_SubscriptionEstablished_Idle == mCurrentState)
    {
        ShouldCleanUp = true;

        WeaveLogDetail(DataManagement, "Handler[%u] [%5.5s] %s Ref(%d) Timeout",
            SubscriptionEngine::GetInstance()->GetHandlerId(this), GetStateStr(), __func__, mRefCount);
    }
    else
    {
        WeaveLogDetail(DataManagement, "Handler[%u] [%5.5s] %s Ref(%d) Timer event fired at wrong state, ignore",
            SubscriptionEngine::GetInstance()->GetHandlerId(this), GetStateStr(), __func__, mRefCount);
    }

exit:
    if (ShouldCleanUp)
    {
        HandleSubscriptionTerminated(WEAVE_ERROR_TIMEOUT, NULL);
    }

    if (!skipTimerCheck)
    {
        _Release();
    }
}

bool SubscriptionHandler::CheckEventUpToDate(LoggingManagement &inLogger)
{
    bool retval = true;

    if (inLogger.IsValid())
    {
        for (size_t i = 0; i < sizeof(mSelfVendedEvents)/sizeof(event_id_t); i++)
        {
            event_id_t eid = inLogger.GetLastEventID(static_cast<ImportanceType>(i+kImportanceType_First));
            if ((eid != 0) && (eid >= mSelfVendedEvents[i]))
            {
                retval = false;
                break;
            }
        }
    }

    return retval;
}

ImportanceType SubscriptionHandler::FindNextImportanceForTransfer(void)
{
    ImportanceType retval = kImportanceType_Invalid;

    for (size_t i = 0; i < sizeof(mSelfVendedEvents)/sizeof(event_id_t); i++)
    {
        if ((mLastScheduledEventId[i] != 0) && mSelfVendedEvents[i] <= mLastScheduledEventId[i])
        {
            retval = static_cast<ImportanceType>(i+kImportanceType_First);
            break;
        }
    }

    return retval;
}


WEAVE_ERROR SubscriptionHandler::SetEventLogEndpoint(LoggingManagement &inLogger)
{
    WEAVE_ERROR err = WEAVE_NO_ERROR;

    VerifyOrExit(inLogger.IsValid(), err = WEAVE_ERROR_INCORRECT_STATE);

    err = inLogger.SetLoggingEndpoint(&(mLastScheduledEventId[0]), kImportanceType_Last - kImportanceType_First + 1, mBytesOffloaded);

exit:
    return err;
}


#if WEAVE_DETAIL_LOGGING
const char * SubscriptionHandler::GetStateStr() const
{
    switch (mCurrentState)
    {
    case kState_Free:
        return "FREE";

    case kState_Subscribing_Evaluating:
        return "EVAL";

    case kState_Subscribing:
        return "PRIME";

    case kState_Subscribing_Notifying:
        return "pNOTF";

    case kState_Subscribing_Responding:
        return "pRESP";

    case kState_SubscriptionEstablished_Idle:
        return "ALIVE";

    case kState_SubscriptionEstablished_Notifying:
        return "NOTIF";

    case kState_Canceling:
        return "CANCL";

    case kState_Aborting:
        return "ABTNG";

    case kState_Aborted:
        return "ABORT";
    }
    return "N/A";
}
#else // WEAVE_DETAIL_LOGGING
const char * SubscriptionHandler::GetStateStr() const
{
    return "N/A";
}
#endif // WEAVE_DETAIL_LOGGING

void SubscriptionHandler::MoveToState(const HandlerState aTargetState)
{
    mCurrentState = aTargetState;
    WeaveLogDetail(DataManagement, "Handler[%u] Moving to [%5.5s] Ref(%d)",
            SubscriptionEngine::GetInstance()->GetHandlerId(this), GetStateStr(), mRefCount);

#if WEAVE_DETAIL_LOGGING
    if (kState_Free == mCurrentState)
    {
        SubscriptionEngine::GetInstance()->LogSubscriptionFreed();
    }
#endif // #if WEAVE_DETAIL_LOGGING
}

void SubscriptionHandler::OnAckReceived(ExchangeContext * aEC, void * aMsgSpecificContext)
{
    WEAVE_ERROR err = WEAVE_NO_ERROR;
    SubscriptionHandler * const pHandler = reinterpret_cast<SubscriptionHandler *>(aEC->AppState);

    WeaveLogDetail(DataManagement, "Handler[%u] [%5.5s] %s Ref(%d)",
        SubscriptionEngine::GetInstance()->GetHandlerId(pHandler), pHandler->GetStateStr(), __func__, pHandler->mRefCount);

    // Make sure we're not freed by accident.
    pHandler->_AddRef();

    switch (pHandler->mCurrentState)
    {
    case kState_Subscribing_Responding:

        pHandler->FlushExistingExchangeContext();

        pHandler->MoveToState(SubscriptionHandler::kState_SubscriptionEstablished_Idle);

        err = pHandler->RefreshTimer();
        SuccessOrExit(err);

        {
            InEventParam inParam;
            OutEventParam outParam;
            inParam.mSubscriptionEstablished.mSubscriptionId = pHandler->mSubscriptionId;
            inParam.mSubscriptionEstablished.mHandler = pHandler;

            // Note we could be aborted in this callback
            pHandler->mEventCallback(pHandler->mAppState, kEvent_OnSubscriptionEstablished, inParam, outParam);
        }

        // Run NE since things may have changed.
        SubscriptionEngine::GetInstance()->GetNotificationEngine()->Run();
        break;

    default:
        break;
    }

exit:

    if (WEAVE_NO_ERROR != err)
    {
        pHandler->HandleSubscriptionTerminated(err, NULL);
    }

    pHandler->_Release();
}

void SubscriptionHandler::OnSendError (ExchangeContext *aEC, WEAVE_ERROR aErrorCode, void *aMsgSpecificContext)
{
    SubscriptionHandler * const pHandler = reinterpret_cast<SubscriptionHandler *>(aEC->AppState);

    WeaveLogDetail(DataManagement, "Handler[%u] [%5.5s] %s Ref(%d)",
            SubscriptionEngine::GetInstance()->GetHandlerId(pHandler), pHandler->GetStateStr(), __func__, pHandler->mRefCount);

    // Make sure we're not freed by accident.
    pHandler->_AddRef();
    pHandler->HandleSubscriptionTerminated(aErrorCode, NULL);
    pHandler->_Release();

    // Run it again to do more useful work.
    // Note that the call to NotificationEngine::Run could actually cause this particular handler to be aborted
    SubscriptionEngine::GetInstance()->GetNotificationEngine()->Run();
}

void SubscriptionHandler::OnResponseTimeout (nl::Weave::ExchangeContext *aEC)
{
    SubscriptionHandler * const pHandler = reinterpret_cast<SubscriptionHandler *>(aEC->AppState);

    WeaveLogDetail(DataManagement, "Handler[%u] [%5.5s] %s Ref(%d)",
            SubscriptionEngine::GetInstance()->GetHandlerId(pHandler), pHandler->GetStateStr(), __func__, pHandler->mRefCount);

    // Make sure we're not freed by accident.
    pHandler->_AddRef();
    pHandler->HandleSubscriptionTerminated(WEAVE_ERROR_TIMEOUT, NULL);
    pHandler->_Release();
}

void SubscriptionHandler::OnMessageReceivedFromLocallyHeldExchange (nl::Weave::ExchangeContext *aEC, const nl::Inet::IPPacketInfo *aPktInfo,
    const nl::Weave::WeaveMessageInfo *aMsgInfo, uint32_t aProfileId,
    uint8_t aMsgType, PacketBuffer *aPayload)
{
    // Status Report for Notification request during priming
    // Status Report for Notification request when subscription is alive
    // Status Report for Cancel request

    WEAVE_ERROR err = WEAVE_NO_ERROR;
    SubscriptionHandler * const pHandler = reinterpret_cast<SubscriptionHandler *>(aEC->AppState);

    bool RetainExchangeContext = false;
    bool isStatusReportValid = false;
    bool isNotificationRejectedForInvalidValue = false;
    nl::Weave::Profiles::StatusReporting::StatusReport status;

    WeaveLogDetail(DataManagement, "Handler[%u] [%5.5s] %s Ref(%d)",
            SubscriptionEngine::GetInstance()->GetHandlerId(pHandler), pHandler->GetStateStr(), __func__, pHandler->mRefCount);

    // Make sure we're not freed by accident.
    pHandler->_AddRef();

    VerifyOrExit(aEC == pHandler->mEC, err = WEAVE_ERROR_INCORRECT_STATE);

    if ((nl::Weave::Profiles::kWeaveProfile_Common == aProfileId) && (nl::Weave::Profiles::Common::kMsgType_StatusReport == aMsgType))
    {
        // Note that payload is not freed in this call to parse
        err = nl::Weave::Profiles::StatusReporting::StatusReport::parse(aPayload, status);
        SuccessOrExit(err);
        isStatusReportValid = true;
        WeaveLogDetail(DataManagement, "Received Status Report 0x%" PRIX32 " : 0x%" PRIX16, status.mProfileId, status.mStatusCode);

        if ((nl::Weave::Profiles::kWeaveProfile_WDM == status.mProfileId) && (kStatus_InvalidValueInNotification == status.mStatusCode))
        {
            isNotificationRejectedForInvalidValue = true;
        }
    }

    switch (pHandler->mCurrentState)
    {
    case kState_Subscribing_Notifying:
        // response for notification request during priming, don't close the exchange context, for more notification requests might
        // need to go through this same exchange context
        VerifyOrExit(isStatusReportValid, err = WEAVE_ERROR_INVALID_MESSAGE_TYPE);

        if (status.success())
        {
            // good
        }
        else if (isNotificationRejectedForInvalidValue)
        {
            // rejected for invalid value
            // we don't really support this, assume it's accepted and continue.
            WeaveLogDetail(DataManagement, "Notification rejected, ignore rejection");
        }
        else
        {
            ExitNow(err = WEAVE_ERROR_STATUS_REPORT_RECEIVED);
        }

        // only retain the EC if we're good to continue processing
        RetainExchangeContext = true;

        // Only prompt the NotificationEngine if we received a status report indicating success. Otherwise, the subscription will get torn down and as part of that
        // clean-up, a similar invocation of OnNotifyConfirm will happen.
        SubscriptionEngine::GetInstance()->GetNotificationEngine()->OnNotifyConfirm(pHandler, status.success());

        // kick back from kState_Subscribing_Notifying to kState_Subscribing and evaluate again
        pHandler->MoveToState(kState_Subscribing);

        // kick notification engine again
        // Note that the call to NotificationEngine::Run could actually cause this particular handler to be aborted
        SubscriptionEngine::GetInstance()->GetNotificationEngine()->Run();
        break;

    case kState_SubscriptionEstablished_Notifying:
        // response for notification request while subscription is alive. close the exchange context
        VerifyOrExit(isStatusReportValid, err = WEAVE_ERROR_INVALID_MESSAGE_TYPE);

        if (status.success())
        {
            // good
        }
        else if (isNotificationRejectedForInvalidValue)
        {
            // rejected for invalid value
            // we don't really support this, assume it's accepted and continue.
            WeaveLogDetail(DataManagement, "Notification rejected, ignore rejection");
        }
        else
        {
            ExitNow(err = WEAVE_ERROR_STATUS_REPORT_RECEIVED);
        }

        // don't call flush for us in the end
        RetainExchangeContext = true;

        // Only prompt the NotificationEngine if we received a status report indicating success. Otherwise, the subscription will get torn down and as part of that
        // clean-up, a similar invocation of OnNotifyConfirm will happen.
        SubscriptionEngine::GetInstance()->GetNotificationEngine()->OnNotifyConfirm(pHandler, status.success());

        // flush right here, for NotificationEngine::Run() only flushes when a new notification is needed
        // make it clear that we do not need this EC anymore
        pHandler->FlushExistingExchangeContext();
        aEC = NULL;

        // kick back from kState_SubscriptionEstablished_Notifying to kState_SubscriptionEstablished_Idle and evaluate again
        pHandler->MoveToState(kState_SubscriptionEstablished_Idle);

        err = pHandler->RefreshTimer();
        SuccessOrExit(err);

#if WDM_ENABLE_SUBSCRIPTION_CLIENT
        (void)SubscriptionEngine::GetInstance()->UpdateClientLiveness(pHandler->mPeerNodeId, pHandler->mSubscriptionId);
#endif // WDM_ENABLE_SUBSCRIPTION_CLIENT

        // kick notification engine again
        // Note: we could have a new pHandler->mEC here, for sending out the next notification request.
        // when that happens, we're again in kState_SubscriptionEstablished_Notifying
        // Note that the call to NotificationEngine::Run could actually cause this particular handler to be aborted
        SubscriptionEngine::GetInstance()->GetNotificationEngine()->Run();
        break;

#if WDM_ENABLE_SUBSCRIPTION_CANCEL
    case kState_Canceling:
        // response for cancel request while subscription is alive. close the exchange context
        // call abort silently without callback to upper layer, for this subscription was canceled by the upper layer
        pHandler->AbortSubscription();
        break;
#endif // WDM_ENABLE_SUBSCRIPTION_CANCEL

    // We must not receive any callback in any of these states
    default:
        WeaveLogDetail(DataManagement, "Received message in some wrong state, ignore");
        ExitNow(err = WEAVE_ERROR_INCORRECT_STATE);
    }

exit:
    WeaveLogFunctError(err);

    if (NULL != aPayload)
    {
        PacketBuffer::Free(aPayload);
        aPayload = NULL;
    }

    if (!RetainExchangeContext)
    {
        pHandler->FlushExistingExchangeContext();
    }

    if (WEAVE_NO_ERROR != err)
    {
        pHandler->HandleSubscriptionTerminated(err, isStatusReportValid ? &status : NULL);
    }

    pHandler->_Release();
}

void SubscriptionHandler::SetMaxNotificationSize(const uint32_t aMaxSize)
{
    if (aMaxSize > UINT16_MAX)
        mMaxNotificationSize = 0;
    else
        mMaxNotificationSize = aMaxSize;
}

}; // WeaveMakeManagedNamespaceIdentifier(DataManagement, kWeaveManagedNamespaceDesignation_Current)
}; // Profiles
}; // Weave
}; // nl

#endif // WEAVE_CONFIG_ENABLE_RELIABLE_MESSAGING