[pulsar-client-go] branch master updated: support DisableReplication (#543)

[mmerli]

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

mmerli pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar-client-go.git


The following commit(s) were added to refs/heads/master by this push:
     new f33b594  support DisableReplication (#543)
f33b594 is described below

commit f33b594034c9433faae78a144983ab3c8f785add
Author: Yuto Furuta <mzq6mft...@gmail.com>
AuthorDate: Thu Jun 24 18:07:47 2021 +0900

    support DisableReplication (#543)
    
    Co-authored-by: Yuto Furuta <yfur...@yahoo-corp.jp>
---
 pulsar/message.go            | 3 +++
 pulsar/producer_partition.go | 5 +++++
 2 files changed, 8 insertions(+)

diff --git a/pulsar/message.go b/pulsar/message.go
index 2a4343f..23dfefb 100644
--- a/pulsar/message.go
+++ b/pulsar/message.go
@@ -48,6 +48,9 @@ type ProducerMessage struct {
        // ReplicationClusters override the replication clusters for this 
message.
        ReplicationClusters []string
 
+       // Disable the replication for this message
+       DisableReplication bool
+
        // SequenceID set the sequence id to assign to the current message
        SequenceID *int64
 
diff --git a/pulsar/producer_partition.go b/pulsar/producer_partition.go
index 19aa06e..7dd176c 100644
--- a/pulsar/producer_partition.go
+++ b/pulsar/producer_partition.go
@@ -411,6 +411,11 @@ func (p *partitionProducer) internalSend(request 
*sendRequest) {
        if !sendAsBatch {
                p.internalFlushCurrentBatch()
        }
+
+       if msg.DisableReplication {
+               msg.ReplicationClusters = []string{"__local__"}
+       }
+
        added := p.batchBuilder.Add(smm, p.sequenceIDGenerator, payload, 
request,
                msg.ReplicationClusters, deliverAt)
        if !added {