[jira] [Commented] (BOOKKEEPER-321) Message Filter Support

["Sijie Guo (JIRA)" <ji...@apache.org>]

    [ https://issues.apache.org/jira/browse/BOOKKEEPER-321?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=13409534#comment-13409534 ] 

Sijie Guo commented on BOOKKEEPER-321:
--------------------------------------

Message Filter Support in Hedwig:

1) providing a message header *MessageHeader* in *Message* : (BOOKKEEPER-78}

system properties are defined as fields in protobuf definition.
user properties are defined as a key/bytes map structure. applications could put their customized data in it.

{code}
+// common structure to store header or properties
+message Map {
+    message Entry {
+        optional string key  = 1;
+        optional bytes value = 2;
+    }
+    repeated Entry entries = 1;
+}
+
+// message header
+message MessageHeader {
+    // user customized fields used for message filter
+    optional Map properties = 1;
+    // following are system properties in message header
+    optional string messageType = 2;
+}
@@ -30,6 +47,8 @@ message Message {
     required bytes body = 1;
     optional bytes srcRegion = 2;
     optional MessageSeqId msgId = 3;
+    // message header
+    optional MessageHeader header = 4;
 }
{code}

2) add preferences for a subscription. (BOOKKEEPER-332)

*SubscriptionPreferences* is a structure to store all preferences for a subscription. The preferences could be passed as *SubscriptionOptions*, sent to hub server and stored in meta store.

system options are defined as fields in *SubscriptionPreferences* in protobuf definition. currently we had two system options, *messageBound* is used to limit number of messages to receive for a subscription,  *messageFilter* is used to support running server-side message filter.

user options are defined as key/bytes map structure. application could put their customized options in it.

{code}
+// record all preferences for a subscription,
+// would be serialized to be stored in meta store
+message SubscriptionPreferences {
+    // user customized subscription options
+    optional Map options = 1;
+
+    ///
+    /// system defined options
+    ///
+
+    // message bound
+    optional uint32 messageBound = 2;
+    // server-side message filter
+    optional string messageFilter = 3;
+}
+

 message SubscribeRequest{
     required bytes subscriberId = 2;

@@ -100,12 +135,22 @@ message SubscribeRequest{

        // wait for cross-regional subscriptions to be established before returning
        optional bool synchronous = 4 [default = false];
+        // @Deprecated. set message bound in SubscriptionPreferences
        optional uint32 messageBound = 5;
+
+        // subscription options
+        optional SubscriptionPreferences preferences = 6;
 }

+// used in client only
+// options are stored in SubscriptionPreferences structure
 message SubscriptionOptions {
     optional SubscribeRequest.CreateOrAttach createOrAttach = 2 [default = CREATE_OR_ATTACH];
     optional uint32 messageBound = 3 [default = 0];
+    // user customized subscription options
+    optional Map options = 4;
+    // server-side message filter
+    optional string messageFilter = 5;
 }

 message SubscriptionState {
     required MessageSeqId msgId = 1;
+    // @Deprecated.
+    // It is a bad idea to put fields that don't change frequently
+    // together with fields that change frequently
+    // so move it to subscription info structure
     optional uint32 messageBound = 2;
 }

+message SubscriptionData {
+    optional SubscriptionState state = 1;
+    optional SubscriptionPreferences preferences = 2;
+}
+
{code}

The subscription preferences would be returned as response body in subscribe request. The subscription preferences would be passed to message filter running in client-side.

{code}
+message SubscribeResponse {
+    optional SubscriptionPreferences preferences = 2;
+}
+
+message ResponseBody {
+    optional SubscribeResponse subscribeResponse = 2;
+}
+
 message PubSubResponse{
     required ProtocolVersion protocolVersion = 1;
     required StatusCode statusCode = 2;
@@ -142,8 +195,10 @@ message PubSubResponse{
     optional Message message = 5;
     optional bytes topic = 6;
     optional bytes subscriberId = 7;
-}

+    // the following fields are sent by other requests
+    optional ResponseBody respBody = 8;
+}
{code}

3) message filter (BOOKKEEPER-333)

MessageFilter is described as below, which extends the original internal interface.

initialize/uninitialize is used to initialize/uninitialize a message filter running on server-side.

setSubscriptionPreferences passed subscription preferences to the message filter when it starts to run on either server side or client side. so message filter could get the subscriber's preferences and might use them to do filtering.

testMessage is used to answer yes/no giving a message. all the messages return true could be delivered to subscriber.

{code}
+public interface MessageFilter {
+
+    /**
+     * Initialize the message filter.
+     *
+     * @param conf
+     *          Configuration Object. An <i>MessageFilter</i> might read settings from it.
+     * @return message filter
+     * @throws IOException when failed to initialize message filter
+     */
+    public MessageFilter initialize(Configuration conf)
+    throws ConfigurationException, IOException;
+
+    /**
+     * Uninitialize the message filter.
+     */
+    public void uninitialize();
+
+    /**
+     * Set subscription preferences.
+     *
+     * <code>preferences</code> of the subscriber will be passed to message filter when
+     * the message filter attaches to its subscription either in server-side or client-side.
+     *
+     * @param topic
+     *          Topic Name.
+     * @param subscriberId
+     *          Subscriber Id.
+     * @param preferences
+     *          Subscription Preferences.
+     * @return message filter
+     */
+    public MessageFilter setSubscriptionPreferences(ByteString topic, ByteString subscriberId,
+                                                    SubscriptionPreferences preferences);
+
+    /**
+     * Tests whether a particular message passes the filter or not
+     *
+     * @param message
+     * @return
+     */
+    public boolean testMessage(Message message);
+}
{code}

4) server-side filter (BOOKKEEPER-333)

subscriber sets the message filter class name in its subscription preferences when subscribe. hub server would instantiate the message filter for the subscriber when starting serving subscribe requests.

its subscription preferences would be passed to the message filter and all messages tried to be delivered to it would do #testMessage first. if the message returns true, it enters deliver progress; otherwise, hub server would deliver next message.

5) client-side filter (BOOKKEEPER-334 & BOOKKEEPER-335)

A message filter is attached when #startDelivery to receive messages. all messages would be delivered to messageHandler only they passed testing of the message filter. their subscription preferences are passed to the message filter when calling #startDelivery.

java-client:
{code}

     /**
+     * Begin delivery of messages from the server to us for this topic and
+     * subscriberId.
+     *
+     * Only the messages passed <code>messageFilter</code> could be delivered to
+     * <code>messageHandler</code>.
+     *
+     * @param topic
+     *            Topic name of the subscription
+     * @param subscriberId
+     *            ID of the subscriber
+     * @param messageHandler
+     *            Message Handler that will consume the subscribed messages
+     * @throws ClientNotSubscribedException
+     *             If the client is not currently subscribed to the topic
+     * @throws AlreadyStartDeliveryException
+     *             If someone started delivery a message handler before stopping existed one.
+     */
+    public void startDelivery(ByteString topic, ByteString subscriberId,
+                              MessageHandler messageHandler, MessageFilter messageFilter)
+            throws ClientNotSubscribedException, AlreadyStartDeliveryException;
+
{code}

cpp client:
{code}
+
+  typedef std::tr1::shared_ptr<SubscriptionPreferences> SubscriptionPreferencesPtr;
+
+  class MessageFilter {
+  public:
+    virtual void setSubscriptionPreferences(const std::string& topic, const std::string& subscriberId,
+                                            const SubscriptionPreferencesPtr& preferences) = 0;
+    virtual bool testMessage(const Message& message) = 0;
+
+    virtual ~MessageFilter() {};
+  };
+  typedef std::tr1::shared_ptr<MessageFilter> MessageFilterPtr;


+    virtual void startDelivery(const std::string& topic, const std::string& subscriberId,
+                               const MessageHandlerCallbackPtr& callback,
+                               const MessageFilterPtr& filter) = 0;
+
{code}


6) giving customized properties in MessageHeader and SubscriptionPreferences. different application could  leverage them to implement their own filtering logic.
                
> Message Filter Support
> ----------------------
>
>                 Key: BOOKKEEPER-321
>                 URL: https://issues.apache.org/jira/browse/BOOKKEEPER-321
>             Project: Bookkeeper
>          Issue Type: Improvement
>            Reporter: Sijie Guo
>            Assignee: Sijie Guo
>             Fix For: 4.2.0
>
>
> Support message filtering in hedwig.
> 1) add user-customized headers part in Message, which could be used for message filtering.
> 2) add user-customized subscription preferences, which could be used for message filtering.
> 3) support both server-side & client-side message filter.

--
This message is automatically generated by JIRA.
If you think it was sent incorrectly, please contact your JIRA administrators: https://issues.apache.org/jira/secure/ContactAdministrators!default.jspa
For more information on JIRA, see: http://www.atlassian.com/software/jira