[pulsar] branch master updated: add get message method in Record interface (#4341)

[je...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

jerrypeng pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git


The following commit(s) were added to refs/heads/master by this push:
     new 92f0cee  add get message method in Record interface (#4341)
92f0cee is described below

commit 92f0cee05cdaabf4975ff1ea82c6f8740634f0dc
Author: Boyang Jerry Peng <je...@gmail.com>
AuthorDate: Thu May 23 22:23:10 2019 -0700

    add get message method in Record interface (#4341)
---
 .../src/main/java/org/apache/pulsar/functions/api/Record.java       | 6 ++++++
 .../main/java/org/apache/pulsar/functions/source/PulsarRecord.java  | 5 +++++
 2 files changed, 11 insertions(+)

diff --git a/pulsar-functions/api-java/src/main/java/org/apache/pulsar/functions/api/Record.java b/pulsar-functions/api-java/src/main/java/org/apache/pulsar/functions/api/Record.java
index b99c4dc..3c3d7e8 100644
--- a/pulsar-functions/api-java/src/main/java/org/apache/pulsar/functions/api/Record.java
+++ b/pulsar-functions/api-java/src/main/java/org/apache/pulsar/functions/api/Record.java
@@ -18,6 +18,8 @@
  */
 package org.apache.pulsar.functions.api;
 
+import org.apache.pulsar.client.api.Message;
+
 import java.util.Collections;
 import java.util.Map;
 import java.util.Optional;
@@ -104,4 +106,8 @@ public interface Record<T> {
     default Optional<String> getDestinationTopic() {
         return Optional.empty();
     }
+
+    default Optional<Message<T>> getMessage() {
+        return Optional.empty();
+    }
 }
diff --git a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/source/PulsarRecord.java b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/source/PulsarRecord.java
index c03ed9f..3af9a41 100644
--- a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/source/PulsarRecord.java
+++ b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/source/PulsarRecord.java
@@ -106,4 +106,9 @@ public class PulsarRecord<T> implements RecordWithEncryptionContext<T> {
     public void fail() {
         this.failFunction.run();
     }
+
+    @Override
+    public Optional<Message<T>> getMessage() {
+        return Optional.of(message);
+    }
 }