Expose replicated from filed on message struct (apache#251)

Signed-off-by: xiaolong.ran <rxl@apache.org>

pulsar/consumer_partition.go

@@ -424,6 +424,7 @@ func (pc *partitionConsumer) MessageReceived(response *pb.CommandMessage, header
 				msgID:               msgID,
 				payLoad:             payload,
 				replicationClusters: msgMeta.GetReplicateTo(),
+				replicatedFrom:      msgMeta.GetReplicatedFrom(),
 				redeliveryCount:     response.GetRedeliveryCount(),
 			}
 		} else {
@@ -436,6 +437,7 @@ func (pc *partitionConsumer) MessageReceived(response *pb.CommandMessage, header
 				msgID:               msgID,
 				payLoad:             payload,
 				replicationClusters: msgMeta.GetReplicateTo(),
+				replicatedFrom:      msgMeta.GetReplicatedFrom(),
 				redeliveryCount:     response.GetRedeliveryCount(),
 			}
 		}

pulsar/impl_message.go

@@ -145,6 +145,7 @@ type message struct {
 	properties          map[string]string
 	topic               string
 	replicationClusters []string
+	replicatedFrom      string
 	redeliveryCount     uint32
 }
 
@@ -180,6 +181,14 @@ func (msg *message) RedeliveryCount() uint32 {
 	return msg.redeliveryCount
 }
 
+func (msg *message) IsReplicated() bool {
+	return msg.replicatedFrom != ""
+}
+
+func (msg *message) GetReplicatedFrom() string {
+	return msg.replicatedFrom
+}
+
 func newAckTracker(size int) *ackTracker {
 	var batchIDs *big.Int
 	if size <= 64 {

pulsar/message.go

@@ -93,6 +93,12 @@ type Message interface {
 	// Message redelivery increases monotonically in a broker, when topic switch ownership to a another broker
 	// redelivery count will be recalculated.
 	RedeliveryCount() uint32
+
+	// Check whether the message is replicated from other cluster.
+	IsReplicated() bool
+
+	// Get name of cluster, from which the message is replicated.
+	GetReplicatedFrom() string
 }
 
 // MessageID identifier for a particular message