モジュール ncache.client.services.MessagingService

クラス

class MessagingService (value)

このクラスには、MessagingServiceに必要なプロパティとメソッドが含まれています。

ソースコードを展開する

class MessagingService:
    """
    This class contains properties and methods required for Messaging Service.
    """
    def __init__(self, value):
        self.__messagingservice = value

    def get_topic(self, topicname, searchoptions=None):
        """
        Retrieves the topic instance against the TopicSearchOptions and provided name or pattern.

        :param topicname: Name or pattern to identify topic.
        :type topicname: str
        :param searchoptions: TopicSearchOptions specifies to search topic by name or pattern.
        :type searchoptions: TopicSearchOptions
        :return: Returns the topic instance, None if it does not exist.
        :rtype: Topic or None
        """
        ValidateType.is_string(topicname, self.get_topic)
        javatopicname = TypeCaster.to_java_primitive_type(topicname)

        if searchoptions is not None:
            ValidateType.type_check(searchoptions, TopicSearchOptions, self.get_topic)
            javasearchoptions = EnumUtil.get_topic_search_options(searchoptions.value)

            result = self.__messagingservice.getTopic(javatopicname, javasearchoptions)

        else:
            result = self.__messagingservice.getTopic(javatopicname)

        if result is not None:
            result = Topic(result)

        return result

    def create_topic(self, topicname, topicpriority=None):
        """
        Creates and retrieves the topic instance against the specified topic name and priority.

        :param topicname: Name or pattern to identify topic.
        :type topicname: str
        :param topicpriority: Specifies the relative priority of the topic stored in cache.
        :type topicpriority: TopicPriority
        :return: The topic instance.
        :rtype: Topic
        """
        ValidateType.is_string(topicname, self.create_topic)
        javatopicname = TypeCaster.to_java_primitive_type(topicname)

        if topicpriority is not None:
            ValidateType.type_check(topicpriority, TopicPriority, self.create_topic)
            javatopicpriority = EnumUtil.get_topic_priority(topicpriority.value)

            result = self.__messagingservice.createTopic(javatopicname, javatopicpriority)

        else:
            result = self.__messagingservice.createTopic(javatopicname)

        if result is not None:
            result = Topic(result)

        return result

    def delete_topic(self, topicname):
        """
        Deletes the topic instance against the specified topic name.

        :param topicname: Name or pattern to identify topic.
        :type topicname: str
        """
        ValidateType.is_string(topicname, self.delete_topic)
        javatopicname = TypeCaster.to_java_primitive_type(topicname)

        self.__messagingservice.deleteTopic(javatopicname)

    def delete_topic_async(self, topicname):
        """
        Deletes the topic instance asynchronously against the specified topic name.

        :param topicname: Name or pattern to identify topic.
        :type topicname: str
        :return: Future task that performs a delete topic operation in background.Methods of Task can be used to
            determine status of the task.
        :rtype: Task
        """
        ValidateType.is_string(topicname, self.delete_topic_async)
        return asyncio.create_task(self.__return_coroutine(self.delete_topic, topicname))

    def register_cq(self, query):
        """
        Registers the specified continuous query with the cache server. You can use this method multiple times in your
        application depending on its need to receive the notifications for a change in the dataset of your query. This
        method takes as argument an object of ContinuousQuery which has the query and the listeners registered to it.

        :param query: SQL-like query to be executed on cache.
        :type query: ContinuousQuery
        """
        ValidateType.type_check(query, ContinuousQuery, self.register_cq)

        self.__messagingservice.registerCQ(query.get_instance())

    def un_register_cq(self, query):
        """
        Unregisters an already registered continuous query to deactivate it on the cache server. Like registerCQ, it
        takes as argument an object of ContinuousQuery to unregister the listeners which are no more fired after this
        call.

        :param query: SQL-like query to be executed on cache.
        :type query: ContinuousQuery
        """
        ValidateType.type_check(query, ContinuousQuery, self.un_register_cq)

        self.__messagingservice.unRegisterCQ(query.get_instance())

    def add_cache_notification_listener(self, callablefunction, eventtypes, eventdatafilter, keys=None):
        """
        Registers cache notification EventType of type item added, updated or removed against specified key(s) in cache.
        Notification is register against all the keys if no key is specified.

        :param callablefunction: Callable function to be invoked when an item is updated or removed. This
            function should follow this signature: callablefunction(key: str, eventarg: CacheEventArg)
        :type callablefunction: Callable
        :param eventtypes: List of Event types the listener is registered against.
        :type eventtypes: list
        :param eventdatafilter: Tells whether to receive metadata, data with metadata or none when a notification is
            triggered.
        :type eventdatafilter: EventDataFilter
        :param keys: Key or list of keys to identify the cache item.
        :type keys: list or str
        :return: Instance of CacheEventDescriptor if no key is specified. it is required to unregister the notifications.
        :rtype: CacheEventDescriptor or None
        """
        ValidateType.params_check(callablefunction, 2, self.add_cache_notification_listener)
        ValidateType.type_check(eventtypes, list, self.add_cache_notification_listener)
        for eventtype in eventtypes:
            ValidateType.type_check(eventtype, EventType, self.add_cache_notification_listener)
        ValidateType.type_check(eventdatafilter, EventDataFilter, self.add_cache_notification_listener)

        eventlistener = EventsListenerHelper.get_listener(callablefunction, CacheDataModificationListener)
        javaeventtypes = EventsListenerHelper.get_event_type_enum_set(eventtypes)
        javaeventdatafilter = EnumUtil.get_event_data_filter(eventdatafilter.value)

        if keys is not None:
            if type(keys) is str:
                javakey = TypeCaster.to_java_primitive_type(keys)

                self.__messagingservice.addCacheNotificationListener(javakey, eventlistener, javaeventtypes, javaeventdatafilter)
                return

            elif type(keys) is list:
                for key in keys:
                    ValidateType.is_string(key, self.add_cache_notification_listener)

                javakeys = TypeCaster.to_java_array_list(keys, isjavatype=True)

                self.__messagingservice.addCacheNotificationListener(javakeys, eventlistener, javaeventtypes, javaeventdatafilter)
                return

            else:
                raise TypeError(ExceptionHandler.get_invalid_keys_exception_message(self.add_cache_notification_listener))

        else:
            result = self.__messagingservice.addCacheNotificationListener(eventlistener, javaeventtypes, javaeventdatafilter)

            if result is not None:
                descriptor = CacheEventDescriptor(result)
                return descriptor

    def remove_cache_notification_listener(self, descriptor=None, keys=None, callablefunction=None, eventtypes=None):
        """
        Unregisters cache notification against specified key(s) in cache. If no key is specified, this method
        unregisters a cache level event that may have been registered.

        :param descriptor: The descriptor returned when the general event was registered.
        :type descriptor: CacheEventDescriptor
        :param keys: Key or list of keys to identify the cache item.
        :type keys: list or str
        :param callablefunction: Callable function that is invoked when specified event(s) are triggered in cache.
        :type callablefunction: Callable
        :param eventtypes: List of Event types the listener is registered against.
        :type eventtypes: list
        """
        if descriptor is not None and eventtypes is None and callablefunction is None and keys is None:
            ValidateType.type_check(descriptor, CacheEventDescriptor, self.remove_cache_notification_listener)
            javadescriptor = descriptor.get_instance()

            self.__messagingservice.removeCacheNotificationListener(javadescriptor)
            return

        elif eventtypes is not None and callablefunction is not None and keys is not None:
            ValidateType.params_check(callablefunction, 2, self.add_cache_notification_listener)
            ValidateType.type_check(eventtypes, list, self.add_cache_notification_listener)
            for eventtype in eventtypes:
                ValidateType.type_check(eventtype, EventType, self.add_cache_notification_listener)

            eventlistener = EventsListenerHelper.get_listener(callablefunction, CacheDataModificationListener)
            javaeventtypes = EventsListenerHelper.get_event_type_enum_set(eventtypes)

            if type(keys) is str:
                javakey = TypeCaster.to_java_primitive_type(keys)

                self.__messagingservice.removeCacheNotificationListener(javakey, eventlistener, javaeventtypes)
                return

            elif type(keys) is list:
                for key in keys:
                    ValidateType.is_string(key, self.remove_cache_notification_listener)

                javakeys = TypeCaster.to_java_array_list(keys, isjavatype=True)

                self.__messagingservice.removeCacheNotificationListener(javakeys, eventlistener, javaeventtypes)
                return

            else:
                raise TypeError(ExceptionHandler.get_invalid_keys_exception_message(self.remove_cache_notification_listener))

        else:
            raise ValueError(ExceptionHandler.exceptionmessages.get("MessagingService.remove_cache_notification_listener"))

    @staticmethod
    async def __return_coroutine(function, arg):
        return function(arg)
        # For the time being, we have only 1 possible argument. This function has to be made generic if needed in future.

メソッド

def add_cache_notification_listener(self, callablefunction, eventtypes, eventdatafilter, keys=None)

キャッシュ内の指定されたキーに対して追加、更新、または削除されたタイプ項目のキャッシュ通知 EventType を登録します。 キーが指定されていない場合は、すべてのキーに対して通知が登録されます。

:param callablefunction: 項目が更新または削除されたときに呼び出される呼び出し可能関数。 この関数は次のシグネチャに従う必要があります: callablefunction(key: str,eventarg: CacheEventArg) :type callablefunction: Callable :parameventtypes: リスナーが登録されているイベント タイプのリスト。 :typeeventtypes:list :parameventdatafilter:通知がトリガーされたときに、メタデータを受信するか、メタデータを含むデータを受信するか、または何も受信しないかを指定します。 :typeeventdatafilter:EventDataFilter :paramkeys:キャッシュ項目を識別するためのキーまたはキーのリスト。 :type key: list または str :return: キーが指定されていない場合は、CacheEventDescriptor のインスタンス。 通知の登録を解除する必要があります。 :rtype: CacheEventDescriptor または None

ソースコードを展開する

def add_cache_notification_listener(self, callablefunction, eventtypes, eventdatafilter, keys=None):
    """
    Registers cache notification EventType of type item added, updated or removed against specified key(s) in cache.
    Notification is register against all the keys if no key is specified.

    :param callablefunction: Callable function to be invoked when an item is updated or removed. This
        function should follow this signature: callablefunction(key: str, eventarg: CacheEventArg)
    :type callablefunction: Callable
    :param eventtypes: List of Event types the listener is registered against.
    :type eventtypes: list
    :param eventdatafilter: Tells whether to receive metadata, data with metadata or none when a notification is
        triggered.
    :type eventdatafilter: EventDataFilter
    :param keys: Key or list of keys to identify the cache item.
    :type keys: list or str
    :return: Instance of CacheEventDescriptor if no key is specified. it is required to unregister the notifications.
    :rtype: CacheEventDescriptor or None
    """
    ValidateType.params_check(callablefunction, 2, self.add_cache_notification_listener)
    ValidateType.type_check(eventtypes, list, self.add_cache_notification_listener)
    for eventtype in eventtypes:
        ValidateType.type_check(eventtype, EventType, self.add_cache_notification_listener)
    ValidateType.type_check(eventdatafilter, EventDataFilter, self.add_cache_notification_listener)

    eventlistener = EventsListenerHelper.get_listener(callablefunction, CacheDataModificationListener)
    javaeventtypes = EventsListenerHelper.get_event_type_enum_set(eventtypes)
    javaeventdatafilter = EnumUtil.get_event_data_filter(eventdatafilter.value)

    if keys is not None:
        if type(keys) is str:
            javakey = TypeCaster.to_java_primitive_type(keys)

            self.__messagingservice.addCacheNotificationListener(javakey, eventlistener, javaeventtypes, javaeventdatafilter)
            return

        elif type(keys) is list:
            for key in keys:
                ValidateType.is_string(key, self.add_cache_notification_listener)

            javakeys = TypeCaster.to_java_array_list(keys, isjavatype=True)

            self.__messagingservice.addCacheNotificationListener(javakeys, eventlistener, javaeventtypes, javaeventdatafilter)
            return

        else:
            raise TypeError(ExceptionHandler.get_invalid_keys_exception_message(self.add_cache_notification_listener))

    else:
        result = self.__messagingservice.addCacheNotificationListener(eventlistener, javaeventtypes, javaeventdatafilter)

        if result is not None:
            descriptor = CacheEventDescriptor(result)
            return descriptor

def create_topic(self, topicname, topicpriority=None)

指定されたトピック名と優先度に対してトピックインスタンスを作成および取得します。

:param topicname: トピックを識別するための名前またはパターン。 :type topicname: str :param topicpriority: キャッシュに格納されているトピックの相対的な優先度を指定します。 :type topicpriority: TopicPriority :return: トピックのインスタンス。 :rtype: トピック

ソースコードを展開する

def create_topic(self, topicname, topicpriority=None):
    """
    Creates and retrieves the topic instance against the specified topic name and priority.

    :param topicname: Name or pattern to identify topic.
    :type topicname: str
    :param topicpriority: Specifies the relative priority of the topic stored in cache.
    :type topicpriority: TopicPriority
    :return: The topic instance.
    :rtype: Topic
    """
    ValidateType.is_string(topicname, self.create_topic)
    javatopicname = TypeCaster.to_java_primitive_type(topicname)

    if topicpriority is not None:
        ValidateType.type_check(topicpriority, TopicPriority, self.create_topic)
        javatopicpriority = EnumUtil.get_topic_priority(topicpriority.value)

        result = self.__messagingservice.createTopic(javatopicname, javatopicpriority)

    else:
        result = self.__messagingservice.createTopic(javatopicname)

    if result is not None:
        result = Topic(result)

    return result

def delete_topic(self, topicname)

指定されたトピック名に対してトピックインスタンスを削除します。

:param topicname: トピックを識別するための名前またはパターン。 :type トピック名: str

ソースコードを展開する

def delete_topic(self, topicname):
    """
    Deletes the topic instance against the specified topic name.

    :param topicname: Name or pattern to identify topic.
    :type topicname: str
    """
    ValidateType.is_string(topicname, self.delete_topic)
    javatopicname = TypeCaster.to_java_primitive_type(topicname)

    self.__messagingservice.deleteTopic(javatopicname)

def delete_topic_async(self, topicname)

指定されたトピック名に対して非同期的にトピックインスタンスを削除します。

:param topicname: トピックを識別するための名前またはパターン。 :type topicname: str :return: バックグラウンドでトピック削除操作を実行する将来のタスク。タスクのメソッドを使用してタスクのステータスを確認できます。 :rtype: タスク

ソースコードを展開する

def delete_topic_async(self, topicname):
    """
    Deletes the topic instance asynchronously against the specified topic name.

    :param topicname: Name or pattern to identify topic.
    :type topicname: str
    :return: Future task that performs a delete topic operation in background.Methods of Task can be used to
        determine status of the task.
    :rtype: Task
    """
    ValidateType.is_string(topicname, self.delete_topic_async)
    return asyncio.create_task(self.__return_coroutine(self.delete_topic, topicname))

def get_topic(self, topicname, searchoptions=None)

TopicSearchOptionsおよび指定された名前またはパターンに対してトピックインスタンスを取得します。

:param topicname: トピックを識別するための名前またはパターン。 :type topicname: str :param searchoptions: TopicSearchOptions は、名前またはパターンでトピックを検索することを指定します。 :type searchoptions: TopicSearchOptions :return: トピック インスタンスを返します。存在しない場合は None を返します。 :rtype: トピックまたはなし

ソースコードを展開する

def get_topic(self, topicname, searchoptions=None):
    """
    Retrieves the topic instance against the TopicSearchOptions and provided name or pattern.

    :param topicname: Name or pattern to identify topic.
    :type topicname: str
    :param searchoptions: TopicSearchOptions specifies to search topic by name or pattern.
    :type searchoptions: TopicSearchOptions
    :return: Returns the topic instance, None if it does not exist.
    :rtype: Topic or None
    """
    ValidateType.is_string(topicname, self.get_topic)
    javatopicname = TypeCaster.to_java_primitive_type(topicname)

    if searchoptions is not None:
        ValidateType.type_check(searchoptions, TopicSearchOptions, self.get_topic)
        javasearchoptions = EnumUtil.get_topic_search_options(searchoptions.value)

        result = self.__messagingservice.getTopic(javatopicname, javasearchoptions)

    else:
        result = self.__messagingservice.getTopic(javatopicname)

    if result is not None:
        result = Topic(result)

    return result

def register_cq(self, query)

指定された継続クエリをキャッシュサーバーに登録します。 クエリのデータセットの変更に関する通知を受信する必要性に応じて、アプリケーション内でこのメソッドを複数回使用できます。 このメソッドは、クエリとそれに登録されたリスナーを含む ContinuousQuery のオブジェクトを引数として受け取ります。

:param query: キャッシュ上で実行される SQL のようなクエリ。 :type クエリ: ContinuousQuery

ソースコードを展開する

def register_cq(self, query):
    """
    Registers the specified continuous query with the cache server. You can use this method multiple times in your
    application depending on its need to receive the notifications for a change in the dataset of your query. This
    method takes as argument an object of ContinuousQuery which has the query and the listeners registered to it.

    :param query: SQL-like query to be executed on cache.
    :type query: ContinuousQuery
    """
    ValidateType.type_check(query, ContinuousQuery, self.register_cq)

    self.__messagingservice.registerCQ(query.get_instance())

def remove_cache_notification_listener(self, descriptor=None, keys=None, callablefunction=None, eventtypes=None)

キャッシュ内の指定されたキーに対するキャッシュ通知の登録を解除します。 キーが指定されていない場合、このメソッドは登録されている可能性のあるキャッシュ レベル イベントを登録解除します。

:param descriptor: 一般イベントが登録されたときに返される記述子。 :type descriptor: CacheEventDescriptor :param key: キャッシュ項目を識別するためのキーまたはキーのリスト。 :type key: list または str :param callablefunction: 指定されたイベントがキャッシュ内でトリガーされたときに呼び出される呼び出し可能な関数。 :type callablefunction: 呼び出し可能 :parameventtypes: リスナーが登録されているイベント タイプのリスト。 :type イベントタイプ: リスト

ソースコードを展開する

def remove_cache_notification_listener(self, descriptor=None, keys=None, callablefunction=None, eventtypes=None):
    """
    Unregisters cache notification against specified key(s) in cache. If no key is specified, this method
    unregisters a cache level event that may have been registered.

    :param descriptor: The descriptor returned when the general event was registered.
    :type descriptor: CacheEventDescriptor
    :param keys: Key or list of keys to identify the cache item.
    :type keys: list or str
    :param callablefunction: Callable function that is invoked when specified event(s) are triggered in cache.
    :type callablefunction: Callable
    :param eventtypes: List of Event types the listener is registered against.
    :type eventtypes: list
    """
    if descriptor is not None and eventtypes is None and callablefunction is None and keys is None:
        ValidateType.type_check(descriptor, CacheEventDescriptor, self.remove_cache_notification_listener)
        javadescriptor = descriptor.get_instance()

        self.__messagingservice.removeCacheNotificationListener(javadescriptor)
        return

    elif eventtypes is not None and callablefunction is not None and keys is not None:
        ValidateType.params_check(callablefunction, 2, self.add_cache_notification_listener)
        ValidateType.type_check(eventtypes, list, self.add_cache_notification_listener)
        for eventtype in eventtypes:
            ValidateType.type_check(eventtype, EventType, self.add_cache_notification_listener)

        eventlistener = EventsListenerHelper.get_listener(callablefunction, CacheDataModificationListener)
        javaeventtypes = EventsListenerHelper.get_event_type_enum_set(eventtypes)

        if type(keys) is str:
            javakey = TypeCaster.to_java_primitive_type(keys)

            self.__messagingservice.removeCacheNotificationListener(javakey, eventlistener, javaeventtypes)
            return

        elif type(keys) is list:
            for key in keys:
                ValidateType.is_string(key, self.remove_cache_notification_listener)

            javakeys = TypeCaster.to_java_array_list(keys, isjavatype=True)

            self.__messagingservice.removeCacheNotificationListener(javakeys, eventlistener, javaeventtypes)
            return

        else:
            raise TypeError(ExceptionHandler.get_invalid_keys_exception_message(self.remove_cache_notification_listener))

    else:
        raise ValueError(ExceptionHandler.exceptionmessages.get("MessagingService.remove_cache_notification_listener"))

def un_register_cq(self, query)

すでに登録されている連続クエリを登録解除して、キャッシュ サーバー上で非アクティブ化します。 registerCQ と同様に、引数として ContinuousQuery のオブジェクトを受け取り、この呼び出し後に起動されなくなったリスナーの登録を解除します。

:param query: キャッシュ上で実行される SQL のようなクエリ。 :type クエリ: ContinuousQuery

ソースコードを展開する

def un_register_cq(self, query):
    """
    Unregisters an already registered continuous query to deactivate it on the cache server. Like registerCQ, it
    takes as argument an object of ContinuousQuery to unregister the listeners which are no more fired after this
    call.

    :param query: SQL-like query to be executed on cache.
    :type query: ContinuousQuery
    """
    ValidateType.type_check(query, ContinuousQuery, self.un_register_cq)

    self.__messagingservice.unRegisterCQ(query.get_instance())