/*

 * Copyright (c) 2020-2022 Huawei Device Co., Ltd.

 * 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.

 */



#include "pub_sub_feature.h"

#include <ohos_init.h>

#include "samgr_lite.h"

#include "thread_adapter.h"

#include "pub_sub_implement.h"



static const char *GetName(Feature *feature);

static void OnInitialize(Feature *feature, Service *parent, Identity identity);

static void OnStop(Feature *feature, Identity identity);

static BOOL OnMessage(Feature *feature, Request *request);

static Relation *GetRelation(PubSubFeature *feature, const Topic *topic);

static BOOL IsTopicEqual(const Topic *current, const Topic *other);

static PubSubFeature g_broadcastFeature = {

    .GetName = GetName,

    .OnInitialize = OnInitialize,

    .OnStop = OnStop,

    .OnMessage = OnMessage,

    .GetRelation = GetRelation,

};



static void Init(void)

{

    PubSubFeature *feature = &g_broadcastFeature;

    feature->relations.topic = -1;

    feature->relations.callbacks.consumer = NULL;

    UtilsListInit(&feature->relations.callbacks.node);

    UtilsListInit(&feature->relations.node);

    feature->mutex = MUTEX_InitValue();

    SAMGR_GetInstance()->RegisterFeature(BROADCAST_SERVICE, (Feature *)feature);



    PubSubImplement *apiEntry = BCE_CreateInstance((Feature *)feature);

    SAMGR_GetInstance()->RegisterFeatureApi(BROADCAST_SERVICE, PUB_SUB_FEATURE, GET_IUNKNOWN(*apiEntry));

}

SYS_FEATURE_INIT(Init);



static const char *GetName(Feature *feature)

{

    (void)feature;

    return PUB_SUB_FEATURE;

}



static void OnInitialize(Feature *feature, Service *parent, Identity identity)

{

    (void)parent;

    ((PubSubFeature *)feature)->identity = identity;

}



static void OnStop(Feature *feature, Identity identity)

{

    (void)feature;

    (void)identity;

}



static BOOL OnMessage(Feature *feature, Request *request)

{

    PubSubFeature *broadcast = (PubSubFeature *)feature;

    switch (request->msgId) {

        case MSG_PUBLISH: {

            Topic topic = request->msgValue;

            Relation *relation = broadcast->GetRelation(broadcast, &topic);

            if (relation == NULL) {

                return FALSE;

            }

            MUTEX_Lock(broadcast->mutex);

            ConsumerNode *item = NULL;

            UTILS_DL_LIST_FOR_EACH_ENTRY(item, &relation->callbacks.node, ConsumerNode, node) {

                if (item->consumer->identity == NULL) {

                    item->consumer->Notify(item->consumer, &topic, request);

                }

            }

            MUTEX_Unlock(broadcast->mutex);

        }

            break;

        default:

            break;

    }

    return TRUE;

}



static Relation *GetRelation(PubSubFeature *feature, const Topic *topic)

{

    if (feature == NULL || topic == NULL) {

        return NULL;

    }



    UTILS_DL_LIST *list = &feature->relations.node;

    Relation *item = NULL;

    MUTEX_Lock(feature->mutex);

    UTILS_DL_LIST_FOR_EACH_ENTRY(item, list, Relation, node) {

        if (IsTopicEqual(&item->topic, topic)) {

            MUTEX_Unlock(feature->mutex);

            return item;

        }

    }

    MUTEX_Unlock(feature->mutex);

    return NULL;

}



static BOOL IsTopicEqual(const Topic *current, const Topic *other)

{

    return *current == *other;

}