消息通知之系统层事件发布相关流程

前言

Openharmony 3.1Release中存在消息通知的处理，消息通知包括系统层事件发布、消息订阅、消息投递与处理，为了开发者能够熟悉消息的处理流程，本篇文章主要介绍系统层事件发布的相关流程。

整体流程

代码流程

发布消息

{
eventAction)want.SetAction("usual.event.license.LIC_EXPIRED");
EventFwk::CommonEventPublishInfo publishInfo;
    CommonEventData commonData;
    commonData.SetWant(want);
    if (!CommonEventManager::PublishCommonEvent(commonData, publishInfo)) {
        LICENSE_LOGI("failed to publish event[%{public}d]", eventAction);
        return false;
    }
    return true;
    
}

CommonEventManager函数处理，调用PublishCommonEvent

{
    CommonEventPublishInfo publishInfo;
    return PublishCommonEventAsUser(data, publishInfo, nullptr, UNDEFINED_USER);
}

调用PublishCommonEventAsUser

    const CommonEventPublishInfo &publishInfo, const std::shared_ptr<CommonEventSubscriber> &subscriber,
    const int32_t &userId)
{
    EVENT_LOGI("enter");
    return DelayedSingleton<CommonEvent>::GetInstance()->PublishCommonEventAsUser(data, publishInfo, subscriber,
        userId);
}

调用CommonEvent的PublishCommonEventAsUser

    const std::shared_ptr<CommonEventSubscriber> &subscriber, const int32_t &userId)
{
    EVENT_LOGI("enter");
    sptr<IRemoteObject> commonEventListener = nullptr;
    if (!PublishParameterCheck(data, publishInfo, subscriber, commonEventListener)) {
        return false;
    }
    EVENT_LOGD("before PublishCommonEvent proxy valid state is %{public}d", isProxyValid_);
    return commonEventProxy_->PublishCommonEvent(data, publishInfo, commonEventListener, userId);
}

CommonEventProxy调用PublishCommonEvent向服务端发送CES_PUBLISH_COMMON_EVENT消息

    const sptr<IRemoteObject> &commonEventListener, const int32_t &userId)
{
    EVENT_LOGD("start");

    MessageParcel data;
    MessageParcel reply;

     …….

    bool ret = SendRequest(ICommonEvent::Message::CES_PUBLISH_COMMON_EVENT, data, reply);
    if (ret) {
        ret = reply.ReadBool();
    }

    EVENT_LOGD("end");
    return ret;
}

服务端接收CES_PUBLISH_COMMON_EVENT消息

{
    if (data.ReadInterfaceToken() != GetDescriptor()) {
        EVENT_LOGE("local descriptor is not equal to remote");
        return ERR_TRANSACTION_FAILED;
    }

    switch (code) {
        case static_cast<uint32_t>(ICommonEvent::Message::CES_PUBLISH_COMMON_EVENT): {
            std::unique_ptr<CommonEventData> event(data.ReadParcelable<CommonEventData>());
            std::unique_ptr<CommonEventPublishInfo> publishinfo(data.ReadParcelable<CommonEventPublishInfo>());
            sptr<IRemoteObject> commonEventListener = nullptr;
            bool hasLastSubscriber = data.ReadBool();
            if (hasLastSubscriber) {
                sptr<IRemoteObject> commonEventListener = data.ReadRemoteObject();
            }
            int32_t userId = data.ReadInt32();
            if (!event) {
                EVENT_LOGE("Failed to ReadParcelable<CommonEventData>");
                return ERR_INVALID_VALUE;
            }
            if (!publishinfo) {
                EVENT_LOGE("Failed to ReadParcelable<CommonEventPublishInfo>");
                return ERR_INVALID_VALUE;
            }
            bool ret = PublishCommonEvent(*event, *publishinfo, commonEventListener, userId);
            if (!reply.WriteBool(ret)) {
                EVENT_LOGE("Failed to write reply ");
                return ERR_INVALID_VALUE;
            }
            break;
        }

        ……..

        default:
            EVENT_LOGW("unknown, code = %{public}u, flags= %{public}u", code, option.GetFlags());
            return IPCObjectStub::OnRemoteRequest(code, data, reply, option);
    }
    return NO_ERROR;
}

调用服务端PublishCommonEvent函数

    const CommonEventPublishInfo &publishinfo, const sptr<IRemoteObject> &commonEventListener, const int32_t &userId)
{
    EVENT_LOGI("enter");

    if (!IsReady()) {
        return false;
    }

    return PublishCommonEventDetailed(event, publishinfo, commonEventListener, IPCSkeleton::GetCallingPid(),
                   IPCSkeleton::GetCallingUid(),userId);
}

PublishCommonEventDetailed绑定PublishCommonEvent,然后进行事件投递

    const CommonEventPublishInfo &publishinfo, const sptr<IRemoteObject> &commonEventListener, const pid_t &pid,
    const uid_t &uid, const int32_t &userId)
{
    EVENT_LOGI("enter");

    struct tm recordTime = {0};
    if (!GetSystemCurrentTime(&recordTime)) {
        EVENT_LOGE("Failed to GetSystemCurrentTime");
        return false;
    }

    std::string bundleName = DelayedSingleton<BundleManagerHelper>::GetInstance()->GetBundleName(uid);

    if (DelayedSingleton<PublishManager>::GetInstance()->CheckIsFloodAttack(uid)) {
        EVENT_LOGE("Too many common events have been sent in a short period from %{public}s (pid = %{public}d, uid = "
                   "%{public}d, userId = %{public}d)", bundleName.c_str(), pid, uid, userId);
        return false;
    }

    Security::AccessToken::AccessTokenID callerToken = IPCSkeleton::GetCallingTokenID();

    std::function<void()> PublishCommonEventFunc = std::bind(&InnerCommonEventManager::PublishCommonEvent,
        innerCommonEventManager_, event, publishinfo, commonEventListener, recordTime, pid,
        uid, callerToken, userId, bundleName, this);
    return handler_->PostTask(PublishCommonEventFunc);
}

在事件投递中调用处理InnerCommonEventManager::PublishCommonEvent

    const sptr<IRemoteObject> &commonEventListener, const struct tm &recordTime, const pid_t &pid, const uid_t &uid,
    const Security::AccessToken::AccessTokenID &callerToken, const int32_t &userId, const std::string &bundleName,
    const sptr<IRemoteObject> &service)
{
    EVENT_LOGI("enter %{public}s(pid = %{public}d, uid = %{public}d), event = %{public}s to userId = %{public}d",
        bundleName.c_str(), pid, uid, data.GetWant().GetAction().c_str(), userId);

    if (data.GetWant().GetAction().empty()) {
        EVENT_LOGE("the commonEventdata action is null");
        return false;
    }

    if ((!publishInfo.IsOrdered()) && (commonEventListener != nullptr)) {
        EVENT_LOGE("When publishing unordered events, the subscriber object is not required.");
        return false;
    }

    std::string action = data.GetWant().GetAction();
    bool isSystemEvent = DelayedSingleton<CommonEventSupport>::GetInstance()->IsSystemEvent(action);
    …….
    if (!controlPtr_) {
        EVENT_LOGE("CommonEventControlManager ptr is nullptr");
        return false;
    }
    controlPtr_->PublishCommonEvent(eventRecord, commonEventListener);
        …….
    return true;
}

默认IsOrdered是false，参数不配置调用ProcessUnorderedEvent

    const CommonEventRecord &eventRecord, const sptr<IRemoteObject> &commonEventListener)
{
    EVENT_LOGI("enter");

    bool ret = false;

    if (!eventRecord.publishInfo->IsOrdered()) {
        ret = ProcessUnorderedEvent(eventRecord);
    } else {
        ret = ProcessOrderedEvent(eventRecord, commonEventListener);
    }

    return ret;
}

无序事件处理，投递事件在hander中调用NotifyUnorderedEvent

    const CommonEventRecord &eventRecord, const std::shared_ptr<EventSubscriberRecord> &subscriberRecord)
{
    …….

    std::function<void()> innerCallback =
        std::bind(&CommonEventControlManager::NotifyUnorderedEvent, this, eventRecordPtr);

    if (eventRecord.isSystemEvent) {
        ret = handler_->PostImmediateTask(innerCallback);
    } else {
        ret = handler_->PostTask(innerCallback);
    }

    return ret;
}

NotifyUnorderedEvent调用NotifyEvent

{
    ……
    for (auto vec : eventRecord->receivers) {
        size_t index = eventRecord->nextReceiver++;
        eventRecord->curReceiver = vec->commonEventListener;
        if (vec->isFreeze) {
            eventRecord->deliveryState[index] = OrderedEventRecord::SKIPPED;
            DelayedSingleton<CommonEventSubscriberManager>::GetInstance()->InsertFrozenEvents(vec, *eventRecord);
        } else {
            ……
            if (ret == OrderedEventRecord::DELIVERED) {
                eventRecord->state = OrderedEventRecord::RECEIVEING;
                commonEventListenerProxy->NotifyEvent(
                    *(eventRecord->commonEventData), false, eventRecord->publishInfo->IsSticky());
                eventRecord->state = OrderedEventRecord::RECEIVED;
            }
        }
    }
    ……
}

在EventReceiveProxy::NotifyEvent函数中发送消息CES_NOTIFY_COMMON_EVENT

{
    ……
    int32_t result = remote->SendRequest(
        static_cast<uint32_t>(IEventReceive::Message::CES_NOTIFY_COMMON_EVENT), data, reply, option);
    if (result != OHOS::NO_ERROR) {
        EVENT_LOGE("Failed to SendRequest, error code: %{public}d", result);
        return;
    }

    EVENT_LOGD("end");
}

服务端接收消息调用服务端NotifyEvent函数

{
    if (data.ReadInterfaceToken() != GetDescriptor()) {
        EVENT_LOGE("local descriptor is not equal to remote");
        return ERR_TRANSACTION_FAILED;
    }
    switch (code) {
        case static_cast<uint32_t>(IEventReceive::Message::CES_NOTIFY_COMMON_EVENT): {
            std::unique_ptr<CommonEventData> eventData(data.ReadParcelable<CommonEventData>());
            bool ordered = data.ReadBool();
            bool sticky = data.ReadBool();
            if (eventData == nullptr) {
                EVENT_LOGE("callback stub receive common event data is nullptr");
                return ERR_INVALID_VALUE;
            }
            NotifyEvent(*eventData, ordered, sticky);
            break;
        }

        default:
            EVENT_LOGW("event receive stub receives unknown code, code = %{public}u", code);
            return IPCObjectStub::OnRemoteRequest(code, data, reply, option);
    }

    return NO_ERROR;
}

服务端NotifyEvent函数，调用OnReceiveEvent

{
    EVENT_LOGI("enter");

    std::lock_guard<std::mutex> lock(mutex_);
    if (!IsReady()) {
        EVENT_LOGE("not ready");
        return;
    }

    std::function<void()> onReceiveEventFunc =
        std::bind(&CommonEventListener::OnReceiveEvent, this, commonEventData, ordered, sticky);
    handler_->PostTask(onReceiveEventFunc);
}

从上面我们就梳理整个系统层事件发布流程，希望对大家有所帮助。